data_analysis/Analysis/regression.py

import sys
sys.path.append("..")
from Analysis.pearsonr import DFMat, PearsonrMat
from Database.database_ import DatabaseParam
import config
import os
import json
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import Lasso, LassoCV, LinearRegression
from sklearn.model_selection import TimeSeriesSplit
import numpy as np
import matplotlib.pyplot as plt
from sklearn.metrics import r2_score
import scipy.stats as stats
from utils.tools import set_chinese_font
from sklearn.metrics import mean_squared_error, mean_absolute_error
from sklearn.model_selection import cross_val_score
from statsmodels.stats.outliers_influence import OLSInfluence
import statsmodels.api as sm

class RegressionBox(PearsonrMat):
    """Lasso回归模型+OLS最小回归"""
    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.lasso_info = {"help":"x，自变量名；y，因变量名；alpha，最佳参数；coef，自变量权重；intercept，截距；n_iter，迭代次数；dual_gap，对偶间隙；tol，对偶容忍"}
        self.ols_info =  {"help":"x，自变量名；y，因变量名；最佳参数；coef，自变量权重；intercept，截距；n_iter，迭代次数；score，R2决定系数；"}
        self.ols_model = None  # 最终的线性OLS回归模型

    def read_features_file(self):
        """加载特征文件，确定因变量Y和自变量X的标签"""
        path = config.LASSO_FEATURE_FILE_PATH
        if not os.path.exists(path):
            raise FileNotFoundError('文件未发现:', path)
        with open(path, "r", encoding="utf-8") as f:
            json_data = json.load(f)
        return json_data.get('targets'), json_data.get('features')

    def load_features(self)->tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
        y_label_name, x_label_name = self.read_features_file()
        # name转换为code
        y_label_code = [self.name_2code_dict.get(i) for i in y_label_name if self.name_2code_dict.get(i) in self.df_merge.columns.tolist()]
        x_label_code = [self.name_2code_dict.get(i) for i in x_label_name if self.name_2code_dict.get(i) in self.df_merge.columns.tolist()]
        if len(y_label_code) ==0 or len(x_label_code) == 0:
            raise ValueError('需要拟合的特征为空，请检查建模字段是否存在', (y_label_code, x_label_code))
        targets = self.df_merge.loc[:, y_label_code].copy()
        features = self.df_merge.loc[:, x_label_code].copy()
        time = self.df_merge.loc[:, ['time']].copy()
        return targets, features, time

    def select_features(self):
        pass

    def lasso_(self, y_value:np.ndarray, scaler_x_value:np.ndarray,n_splits:int=5, max_iter:int=10000):
        """实现Lasso回归分析，选择字段"""
        tscv = TimeSeriesSplit(n_splits=n_splits)
        # 寻找最优alphas
        lasso_model = LassoCV(alphas=100,
                              cv=tscv,
                              max_iter=max_iter,
                              random_state=42,
                              n_jobs=-1)
        lasso_model.fit(scaler_x_value, y_value)
        # 记录最优alphas
        self.lasso_info['alpha'] = lasso_model.alpha_
        # 记录截距
        self.lasso_info['intercept'] = lasso_model.intercept_
        # 记录迭代次数
        self.lasso_info['n_iter'] = lasso_model.n_iter_
        # 记录对偶间隙
        self.lasso_info['dual_gap'] = lasso_model.dual_gap_
        # 记录对偶容忍
        self.lasso_info['tol'] = lasso_model.tol
        # 记录权重
        self.lasso_info['coef'] = lasso_model.coef_

    def ols_(self, y_value:np.ndarray, scaler_x_value:np.ndarray)->LinearRegression:
        """OLS回归"""
        model = LinearRegression()
        model.fit(scaler_x_value, y_value)
        # 记录截距
        self.ols_info['intercept'] = model.intercept_
        # 记录权重
        self.ols_info['coef'] = model.coef_
        # 记录R²
        self.ols_info['score'] = model.score(scaler_x_value, y_value)
        return model

    def any_regression_full(self, target_name:str):
        """对任意输入字段进行全字段回归建模"""
        pass

    def any_regression_r_rank(self, target_name:str):
        """基于皮尔逊系数排序对字段进行回归建模"""
        # 所有需要建模的字段
        y_label_name = target_name
        x_label_name = self.query_r_rank_n(y_label_name)  # 根据皮尔逊排序挑选相关性字段
        # 剔除自身字段
        if y_label_name in x_label_name:
            x_label_name.remove(y_label_name)

        # 拿到数据
        y_label_code = self.name_2code_dict[y_label_name]
        x_label_code = [self.name_2code_dict.get(name) for name in x_label_name]

        y = self.df_merge.loc[:, y_label_code].copy()  # 真实值
        y = y.to_numpy()
        x = self.df_merge.loc[:, x_label_code].copy()  # 预测值
        t = self.df_merge.loc[:, 'time'].copy()   # 时间序列

        # 标准化
        scaler = StandardScaler()
        x = scaler.fit_transform(x)

        # Lasso回归，选择字段
        self.lasso_(y_value=y, scaler_x_value=x)
        self.lasso_info['x'] = x_label_name
        self.lasso_info['y'] = y_label_name

        # Lasso模型诊断与可视化
        print('\n===========Lasso训练结果==================')
        print(f'最优lambda：{self.lasso_info.get('alpha')}')
        print(f'Y：{self.lasso_info.get('y')}')
        print(f"Lasso系数:")
        for feat, coef in zip(x_label_name, self.lasso_info.get('coef')):
            print(f"  {feat}: {coef}")
        print(f'截距:{self.lasso_info.get('intercept')}')
        print(f'迭代次数：{self.lasso_info.get('n_iter')}')
        print(f'对偶间隙：{self.lasso_info.get('dual_gap')}')
        print(f'对偶间隙容忍：{self.lasso_info.get('tol')}')

        # OLS回归，筛选系数不为零的向量
        mask = self.lasso_info.get('coef') != 0
        x_label_name = list(np.array(x_label_name)[mask])
        x_label_code = list(np.array(x_label_code)[mask])

        x = self.df_merge.loc[:, x_label_code]  # 没进行归一化/标准化
        self.ols_model = self.ols_(y_value=y, scaler_x_value=x)
        self.ols_info['x'] = x_label_name
        self.ols_info['y'] = y_label_name


        # OLS模型诊断
        print('\n===========OLS训练结果==================')
        print(f"OLS 截距: {self.ols_info.get('intercept')}")
        print(f"OLS 系数:")
        for feat, coef in zip(x_label_name, self.ols_info.get('coef')):
            print(f"  {feat}: {coef}")
        print(f"OLS R² (训练集): {self.ols_info.get('score'):.4f}")

        # 基本指标评价
        y_pred = self.ols_model.predict(x)
        residuals = y - y_pred
        mse = mean_squared_error(y, y_pred)
        rmse = np.sqrt(mse)
        mae = mean_absolute_error(y, y_pred)
        r2 = r2_score(y, y_pred)
        # 调整R²
        n = len(y)
        p = x.shape[1]
        adj_r2 = 1 - (1 - r2) * (n - 1) / (n - p - 1)
        print("\n===========模型性能指标==================:")
        print(f"均方误差 (MSE): {mse:.4f}")
        print(f"均方根误差 (RMSE): {rmse:.4f}")
        print(f"平均绝对误差 (MAE): {mae:.4f}")
        print(f"决定系数 (R²): {r2:.4f}")
        print(f"调整R²: {adj_r2:.4f}")

        # 创建诊断图
        set_chinese_font()
        fig, axes = plt.subplots(2, 3, figsize=(15, 10))

        # 1. 残差 vs 拟合值图（检查同方差性和线性关系）
        axes[0, 0].scatter(y_pred, residuals, alpha=0.6)
        axes[0, 0].axhline(y=0, color='red', linestyle='--')
        axes[0, 0].set_xlabel('拟合值')
        axes[0, 0].set_ylabel('残差')
        axes[0, 0].set_title('残差 vs 拟合值')

        # 2. 正态Q-Q图（检查残差正态性）
        stats.probplot(residuals, dist="norm", plot=axes[0, 1])
        axes[0, 1].set_title('Q-Q图（检查正态性）')

        # 3. 残差直方图
        axes[0, 2].hist(residuals, bins=30, density=True, alpha=0.7)
        axes[0, 2].set_xlabel('残差')
        axes[0, 2].set_ylabel('密度')
        axes[0, 2].set_title('残差分布')

        # 4. 观测值 vs 拟合值
        axes[1, 0].scatter(y, y_pred, alpha=0.6)
        min_val = min(y.min(), y_pred.min())
        max_val = max(y.max(), y_pred.max())
        axes[1, 0].plot([min_val, max_val], [min_val, max_val], 'red', linestyle='--')
        axes[1, 0].set_xlabel('实际值')
        axes[1, 0].set_ylabel('预测值')
        axes[1, 0].set_title('实际值 vs 预测值')
        r2 = r2_score(y, y_pred)
        axes[1, 0].text(0.05, 0.95, f'R² = {r2:.3f}', transform=axes[1, 0].transAxes)

        # 5. 残差的时间序列图（如果是时间序列数据）
        axes[1, 1].plot(residuals)
        axes[1, 1].axhline(y=0, color='red', linestyle='--')
        axes[1, 1].set_xlabel('时间/观测序号')
        axes[1, 1].set_ylabel('残差')
        axes[1, 1].set_title('残差时间序列')

        # 6. 尺度-位置图（检查同方差性）
        standardized_residuals = residuals / np.std(residuals)
        axes[1, 2].scatter(y_pred, np.sqrt(np.abs(standardized_residuals)), alpha=0.6)
        axes[1, 2].set_xlabel('拟合值')
        axes[1, 2].set_ylabel('√|标准化残差|')
        axes[1, 2].set_title('尺度-位置图')

        plt.tight_layout()
        plt.show()
        pass



    def any_regression_custom(self, target_name:str, path:str):
        """基于自定义字段进行回归建模，从文件读入建模字段"""

    def auto_fit(self, x_label_name:str, y_label_name:str, is_use_lasso:bool=True):
        """回归分析"""


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)

    my_box = RegressionBox(
        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))
    # 计算皮尔逊
    my_box.calculate_pearsonr_mat()
    # 进行回归分析
    my_box.any_regression_r_rank("RO1脱盐率")