data_analysis/Analysis/pearsonr.py

import csv
import sys

sys.path.append("..")
import os
import pandas as pd
import config
from Database.database_ import Database, DatabaseParam
import json
from scipy import stats
import numpy as np
import pickle
from utils.tools import cal_vari_without_zero_nan, cal_vari_without_nan, df_is_symetry, quick_sort, load_transfer_file_name_code

class DFMat:
    """输入字段，实现从数据库中获取全部数据，核心的属性是pandas.Dataframe,融合了所有字段的数据,从数据库拿完数据后需要进行数据的清洗和预处理"""
    def __init__(self, keys_file_dir: str, min_records:int, db_param: DatabaseParam, transfer_file_dir:str, is_from_local:bool=True):
        self.bad_keys = config.EXCLUDE_WORDS
        self.keys_file_dir = keys_file_dir
        self.min_records = min_records
        self.keys = self.load_keys()  # 升序排序
        self.db_param = db_param
        self.transfer_file_dir = transfer_file_dir
        self.name_2code_dict, self.code_2name_dict = self.load_transfer_file()  # 转换字典
        self.diff_words = config.DIFF_WORDS  # 需要差分计算的字段，如果字段中包括这些字段就进行差分平稳化
        self.is_from_local = is_from_local
        # 本地保存数据库数据，避免重复查询
        self.local_df_merge_path = config.DF_MERGE_FILE_PATH
        self.df_merge = self.__construct()  # 构建数据部分，初始化时完成

    def load_keys(self):
        keys_list = []
        with open(self.keys_file_dir, "r", encoding="utf-8") as f:
            csv_reader = csv.reader(f)
            try:
                label = next(csv_reader)
            except StopIteration:
                print('文件不存在：', self.keys_file_dir)
            for row in csv_reader:
                records_num = int(row[6])
                records_name = row[0]
                if records_num < self.min_records: continue
                keys_list.append(records_name)
        # 升序排序
        keys_list = sorted(keys_list)
        # 剔除列表不需要的字段
        keys_list = self.exclude_keys(keys_list)
        return keys_list  # 升序排列

    def exclude_keys(self, keys_list:list):
        """根据剔除列表对键入的字段进行剔除"""
        new_keys = []
        for name in keys_list:
            flag = False
            for bad_key in self.bad_keys:
                if bad_key in name:
                    flag = True
                    break
            if flag: continue
            new_keys.append(name)
        return new_keys

    def load_transfer_file(self):
        """加载转换文件"""
        path = self.transfer_file_dir
        return load_transfer_file_name_code(path)
        # if not os.path.exists(self.transfer_file_dir):
        #     raise FileNotFoundError('文件未发现:', self.transfer_file_dir)
        # with open(self.transfer_file_dir, "r", encoding="utf-8") as f:
        #     json_data = json.load(f)
        # return json_data.get('name_2_code'), json_data.get('code_2_name')

    def save_df_merge(self, data:pd.DataFrame):
        """保存文件到本地"""
        with open(self.local_df_merge_path, 'wb') as f:
            pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
        print(f'mat_shape:{data.shape}，文件保存至：', self.local_df_merge_path)

    def load_from_local(self) -> pd.DataFrame:
        """从本地加载数据"""
        with open(self.local_df_merge_path, 'rb') as f:
            local_data = pickle.load(f)
        return local_data

    def normalize(self, data:pd.DataFrame)-> pd.DataFrame:
        """对数据进行归一化，消除量纲影响"""
        # 皮尔逊系数的计算不需要
        pass

    @staticmethod
    def diff_tool(data: pd.Series):
        """用于计算累计量的差分,单调递增"""
        data = data.copy()
        # 0值替换为Nan
        data.replace([np.inf, -np.inf, 0], np.nan, inplace=True)
        data = data.diff()
        # 检查负值，替换为nan
        data[data < 0] = np.nan
        data[0] = data.mean()
        # 将nan向前填充
        data.ffill(inplace=True)
        return data

    def stabilize(self, data:pd.DataFrame)-> pd.DataFrame:
        """数据差分使数据平稳化"""
        if len(self.diff_words) == 0: return data

        # 获取所有列标签, 仅保留需要做差分的字段
        col_label_list = data.columns.tolist()
        # 剔除时间戳字段
        if 'time' in col_label_list:
            col_label_list.remove('time')
        # 查找需要平稳化的字段
        diff_label_list = set()
        for col in col_label_list:
            name = self.code_2name_dict[col]
            for dword in self.diff_words:
                if dword in name:
                    diff_label_list.add(col)
        diff_label_list = list(diff_label_list)

        for col in diff_label_list:
            data.loc[:, col] = self.diff_tool(data.loc[:, col])

        return data

    @staticmethod
    def remove_outliers(data:pd.Series, fill_value=0, times:int=1)-> pd.Series:
        """剔除序列的离群点，使用fill_value进行填充"""
        data = data.copy(deep=True)
        for time in range(abs(times)): # 执行times次
            # 计算均值和方差
            mean, std_dev = cal_vari_without_nan(data)
            fill_value = mean
            threshold = 3 * std_dev
            limit_top = mean + threshold
            limit_low = mean - threshold
            # 处理离群点
            mask = data.notna() & (data != 0) & ((data < limit_low) | (data > limit_top))
            # 离群点填充
            data.loc[mask] = fill_value
            # for idx, v in enumerate(data):
            #     if pd.isna(v) or abs(v - 0.) < 1e-6: continue  # 0和nan不处理
            #     if v > limit_top or v < limit_low:
            #         data[idx] = fill_value  # 离群点填充
        return data

    def clean(self, data:pd.DataFrame)-> pd.DataFrame:
        """对数据进行清洗，把离群值和Nan替换为平均值,0不参与该过程"""
        # 获取所有列标签
        col_label_list = data.columns.tolist()
        if 'time' in col_label_list: col_label_list.remove('time')  # 不处理time列
        # 逐列处理离群点
        for col_label in col_label_list:
            # 拿到列数据
            col_series = data.loc[:, col_label]

            data.loc[:, col_label] = self.remove_outliers(col_series, times=1)
        # 统一处理nan值，使用平均值填充nan
        cols_mean = data[col_label_list].mean()  # 自动跳过平均值
        cols_mean = cols_mean.fillna(0)
        data[col_label_list] = data[col_label_list].fillna(cols_mean)
        return data

    def fetch(self)->pd.DataFrame:
        """从数据库中拿到数据，拿到原始数据，尽量不要在这里面清洗数据"""
        # 数据库操作应在内部
        data_names = self.keys
        data_codes = [self.name_2code_dict.get(name) for name in data_names]
        # 从数据库取数据
        with Database(self.db_param) as db:  # 连接数据库
            # 检查表是否存在
            if not db.sheet_exists(config.DB_SHEET_NAME):
                raise RuntimeError(f'表{config.DB_SHEET_NAME}不存在于数据库{config.DB_NAME}中！')
            # SQL查询数据
            group_df = db.query_sql_time_series_group2data_frame(
                code_name_dict=self.code_2name_dict,
                project_id=config.PROJECT_ID,
                sheet_name=config.DB_SHEET_NAME,
                data_codes=data_codes,
                start_year=config.CHECK_YEAR_START,
                end_year=config.CHECK_YEAR_END,
                start_month=config.CHECK_MONTH_START,
                end_month=config.CHECK_MONTH_END,
                start_day=config.CHECK_DAY_START,
                end_day=config.CHECK_DAY_END,
                start_hour=config.CHECK_HOUR_START,
                end_hour=config.CHECK_HOUR_END,
                start_minute=config.CHECK_MINUTE_START,
                end_minute=config.CHECK_MINUTE_END,
                start_second=config.CHECK_SECONDS_START,
                end_second=config.CHECK_SECONDS_END)
        return group_df

    def __construct(self):
        """构建所有满足条件的字段dataframe"""
        # 尝试从本地加载数据
        if self.is_from_local:
            if os.path.exists(self.local_df_merge_path):
                print(f'从本地{self.local_df_merge_path}加载数据库数据')
                return self.load_from_local()
            else:
                print(f'从本地{self.local_df_merge_path}加载失败，文件不存在！')
        # 先从数据库获取数据
        print("尝试从数据库获取数据！")
        group_df = self.fetch()
        # 清洗数据，消除Nan和离群值
        group_df = self.clean(group_df)  # 把自己的引用给自己
        # 平稳化
        group_df = self.stabilize(group_df)  # 此时数据不存在nan
        # 如果数据不存在就保存
        if not os.path.exists(self.local_df_merge_path):
            self.save_df_merge(group_df)
        return group_df
    def get_df_merge(self):
        return self.df_merge


class PearsonrMat(DFMat):
    """实现皮尔逊相关系数矩阵，核心属性为pandas.Dataframe，要求键入key，核心的df行和列也是按照给定的keys写入"""
    def __init__(self, keys_file_dir: str, min_records:int, db_param: DatabaseParam, transfer_file_dir:str, is_from_local:bool=True):
        super().__init__(keys_file_dir=keys_file_dir, min_records=min_records, db_param=db_param, transfer_file_dir=transfer_file_dir, is_from_local=is_from_local)
        self.r_mat = None
        self.lag_mat = None

    def r_mat_filter(self):
        """将mat中只和自己相关的字段过滤掉"""
        # 先找到需要删除的字段
        filter_label_list = []
        label_list = self.r_mat.columns.tolist()
        for label in label_list:
            r_col = self.r_mat.loc[:, label]
            non_zero_counter = 0
            for value in r_col:
                if value > config.PEARSONR_VALUE_THRESHOLD:
                    non_zero_counter += 1
            if non_zero_counter < 2:
                filter_label_list.append(label)
        self.r_mat.drop(filter_label_list, axis=0, inplace=True)
        self.r_mat.drop(filter_label_list, axis=1, inplace=True)

    def pearsonr_with_lag(self, a_series_data_label: str, b_series_data_label: str):
        """带滞后的皮尔逊计算"""

        lags = config.MAX_LAG

        if lags == 0:
            left_point = 0
            right_point = 1
        elif lags > 0:
            left_point = -lags
            right_point = lags
        else:
            raise ValueError('最大滞后不能为负数', lags)

        step = config.STEP
        # 不同滞后下的相关系数
        list_r_lag = []
        for lag in range(left_point, right_point, step):
            if lag < 0:  # a滞后于b
                series_a_shifted = self.df_merge.loc[:, a_series_data_label][-lag:]
                series_b_shifted = self.df_merge.loc[:, b_series_data_label][:lag]
            elif lag > 0:  # b滞后于a
                series_a_shifted = self.df_merge.loc[:, a_series_data_label][:-lag]
                series_b_shifted = self.df_merge.loc[:, b_series_data_label][lag:]
            elif lag == 0:  # 0滞后
                series_a_shifted = self.df_merge.loc[:, a_series_data_label]
                series_b_shifted = self.df_merge.loc[:, b_series_data_label]
            else:
                series_a_shifted = None
                series_b_shifted = None
            # 计算皮尔逊系数和显著性
            if series_a_shifted is None or series_b_shifted is None:
                raise RuntimeError('数据不应为None',series_a_shifted, series_b_shifted)
            r, p_value = stats.pearsonr(series_a_shifted, series_b_shifted)
            # 过滤不显著的数据
            if p_value <= config.P_VALUE_THRESHOLD:
                list_r_lag.append(np.float32(r))
        if len(list_r_lag) > 0:
            return max(list_r_lag)
        else:
            return 0

    def pearsonr_(self, a_series_data_label: str, b_series_data_label: str)->float:
        a_series_data = self.df_merge.loc[:, a_series_data_label]
        b_series_data = self.df_merge.loc[:, b_series_data_label]
        r, p_value = stats.pearsonr(a_series_data, b_series_data)
        if p_value <= config.P_VALUE_THRESHOLD:  # 结果显著
            return np.float32(r)
        else:
            return np.float32(0)

    def skip_tool(self, series_a_name:str, series_b_name:str)->bool:
        # 标签转换
        series_a_name = self.code_2name_dict.get(series_a_name)
        series_b_name = self.code_2name_dict.get(series_b_name)
        if '温度' in series_a_name and '温度' in series_b_name: return True
        if '次数' in series_a_name and '次数' in series_b_name: return True
        if '累计' in series_a_name and '累计' in series_b_name: return True
        if '电流' in series_a_name and '电流' in series_b_name: return True
        if '电压' in series_a_name and '电压' in series_b_name: return True
        if '电流' in series_a_name and '温度' in series_b_name: return True
        if '温度' in series_a_name and '电流' in series_b_name: return True
        if '累计电量' in series_a_name and '累计电量' in series_b_name: return True
        if '运行时间' in series_a_name and '累计电量' in series_b_name: return True
        if '累计电量' in series_a_name and '运行时间' in series_b_name: return True
        if '运行时间' in series_a_name and '运行时间' in series_b_name: return True
        if '时间设定' in series_a_name and '时间设定' in series_b_name: return True
        return False

    def calculate_pearsonr_mat(self):
        """计算pearson系数"""
        # 判断是否能够从本地读取，可以的话就不从新计算了
        if os.path.exists(config.R_MAT_JSON_PATH):
            print(f"皮尔逊系数矩阵从本地读取, {config.R_MAT_JSON_PATH}")
            with open(config.R_MAT_JSON_PATH, 'rb') as f:
                self.r_mat = pickle.load(f)
            return

        # 先算再使标签中文化
        all_labels_code = [k for k in self.df_merge.columns.tolist() if k != 'time']
        all_labels_name = sorted([self.code_2name_dict.get(l) for l in all_labels_code])  # 升序
        self.r_mat = pd.DataFrame(index=all_labels_name, columns=all_labels_name, dtype=np.float32)
        self.r_mat.fillna(0, inplace=True)  # 全部填充为0
        for a_label_idx in range(0, len(all_labels_code), 1):  # 行标签
            for b_label_idx in range(a_label_idx, len(all_labels_code), 1): # 列标签
                # 检查是否属于可跳过的字段组合
                a_label = all_labels_code[a_label_idx]
                b_label = all_labels_code[b_label_idx]
                if self.skip_tool(a_label, b_label):
                    print(f'跳过组合：{a_label}，{b_label}')
                    self.r_mat.loc[self.code_2name_dict.get(a_label), self.code_2name_dict.get(b_label)] = np.float32(0)
                # 正式计算
                if config.IS_LAG:
                    result = self.pearsonr_with_lag(a_label, b_label)
                else:
                    result = self.pearsonr_(a_label, b_label)
                # 要保证对称性
                self.r_mat.loc[self.code_2name_dict.get(a_label), self.code_2name_dict.get(b_label)] = result
                self.r_mat.loc[self.code_2name_dict.get(b_label), self.code_2name_dict.get(a_label)] = result
        # 筛选一些无关字段
        self.r_mat_filter()
        # 保存计算结果
        self.save_pearsonr_mat()
    def save_pearsonr_mat(self):
        path = config.R_MAT_JSON_PATH
        if os.path.exists(path):
            os.remove(path)
        with open(path, 'wb') as f:
            pickle.dump(self.r_mat, f, protocol=pickle.HIGHEST_PROTOCOL)
        print(f'mat_shape:{self.r_mat.shape}，文件保存至：',path)


    def query_r_rank_n(self, target:str, n:int=-1)->list[str]:
        """输入target字段，从皮尔逊系数矩阵中挑选排名前n的字段, n为-1表示取所有"""
        if self.r_mat is None:
            raise Exception('r_mat 为None，请先计算皮尔逊系数矩阵！')
        # 取出对应的列，皮尔逊矩阵为对称矩阵，因此取一列或者一行就可以了
        if not df_is_symetry(self.r_mat):
            raise RuntimeError('皮尔逊矩阵非对称，请检查计算过程!')
        # 准备排序
        label_list = self.r_mat.index.tolist()
        if target not in label_list:
            raise ValueError(f'查询字段不存在',target)
        # 检查输入参数是否合法
        if n == -1:
            n = np.sum(np.abs(self.r_mat.loc[:, target].to_numpy()) > 0 )
            n = int(n)
        elif n <= 0:
            raise RuntimeError('n输入值非法，应大于0',n)

        elements = []
        for row_label in label_list:
            elements.append((row_label, self.r_mat.loc[row_label, target]))
        # 按照皮尔逊相关系数的绝对值进行升序排序
        quick_sort(elements, 0, len(elements) - 1)
        # 反转list，由大到小排序
        elements = elements[::-1]

        elements = [elements[e][0] for e in range(n)]
        return elements

if __name__ == '__main__':
    # 数据库参数
    db_param = DatabaseParam(
        db_host=config.DB_HOST,
        db_user=config.DB_USER,
        db_password=config.DB_PASSWORD,
        db_name=config.DB_NAME,
        db_port=config.DB_PORT)
    # 先拿到所有的数据
    df_mat = PearsonrMat(keys_file_dir=os.path.join(config.STATISTICS_FILE_DIR, config.STATISTICS_FILE_NAME),
                   min_records=config.MIN_RECORDS, db_param=db_param,
                   transfer_file_dir=os.path.join(config.ALL_ITEMS_FILE_DIR, config.TRANSFER_JSON_NAME)
                         )
    # 计算皮尔逊系数和显著性p值（带滞后）
    df_mat.calculate_pearsonr_mat()

    # 测试函数
    # df_mat.query_r_rank_n('反渗透总产水电导')