Data Exploration Visualization

#!/usr/bin/env python3
"""
数据预处理器 (Data Preprocessor) - 智能数据清洗和转换模块

提供全面的数据预处理功能，包括：
- 数据清洗（缺失值处理、异常值检测和处理）
- 数据类型转换和标准化
- 特征工程（特征选择、创建、转换）
- 数据编码（标签编码、独热编码）
- 数据分割和平衡
- 数据质量评估和改进建议
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional, Union, Any
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler, LabelEncoder, OneHotEncoder
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.feature_selection import SelectKBest, f_classif, f_regression, RFE
from sklearn.ensemble import IsolationForest
from sklearn.model_selection import train_test_split
from imblearn.over_sampling import SMOTE, RandomOverSampler
from imblearn.under_sampling import RandomUnderSampler
import warnings
from pathlib import Path
import json

warnings.filterwarnings('ignore')


class DataPreprocessor:
    """数据预处理器 - 智能数据清洗和特征工程引擎"""

    def __init__(self, config: Optional[Dict] = None):
        """
        初始化数据预处理器

        Parameters:
        - config: 配置参数字典
        """
        self.config = config or {}
        self.preprocessing_steps = []
        self.scalers = {}
        self.encoders = {}
        self.imputers = {}
        self.feature_selectors = {}
        self.preprocessing_report = {}

        # 默认配置
        self.default_config = {
            'missing_threshold': 0.5,  # 缺失值阈值
            'outlier_method': 'isolation_forest',  # 异常值检测方法
            'outlier_contamination': 0.1,  # 异常值比例
            'scaling_method': 'standard',  # 标准化方法
            'encoding_method': 'auto',  # 编码方法
            'feature_selection': False,  # 是否进行特征选择
            'k_features': 10,  # 选择的特征数量
            'test_size': 0.2,  # 测试集比例
            'random_state': 42,  # 随机种子
            'balance_data': False,  # 是否平衡数据
            'balance_method': 'smote'  # 数据平衡方法
        }

        # 合并配置
        self.config = {**self.default_config, **self.config}

    def analyze_data_quality(self, data: pd.DataFrame) -> Dict:
        """
        分析数据质量

        Parameters:
        - data: 数据DataFrame

        Returns:
        - 数据质量报告
        """
        print("🔍 分析数据质量...")

        quality_report = {
            'shape': data.shape,
            'memory_usage': data.memory_usage(deep=True).sum() / 1024**2,  # MB
            'columns': {},
            'overall_score': 0,
            'issues': [],
            'recommendations': []
        }

        total_issues = 0
        total_checks = 0

        for col in data.columns:
            col_info = {
                'dtype': str(data[col].dtype),
                'non_null_count': data[col].count(),
                'null_count': data[col].isnull().sum(),
                'null_percentage': data[col].isnull().sum() / len(data) * 100,
                'unique_count': data[col].nunique(),
                'duplicate_count': data[col].duplicated().sum(),
                'issues': []
            }

            # 检查缺失值
            total_checks += 1
            if col_info['null_percentage'] > 0:
                total_issues += 1
                col_info['issues'].append(f"缺失值: {col_info['null_percentage']:.1f}%")

                if col_info['null_percentage'] > self.config['missing_threshold'] * 100:
                    quality_report['issues'].append(
                        f"列 '{col}' 缺失值过高 ({col_info['null_percentage']:.1f}%)"
                    )

            # 检查重复值
            if col_info['duplicate_count'] > 0:
                col_info['issues'].append(f"重复值: {col_info['duplicate_count']}")

            # 检查数据类型
            if data[col].dtype == 'object':
                # 检查可能的数值型分类变量
                try:
                    pd.to_numeric(data[col], errors='raise')
                    col_info['issues'].append("可能是数值型但存储为字符串")
                    quality_report['recommendations'].append(
                        f"考虑将列 '{col}' 转换为数值类型"
                    )
                except:
                    pass
            elif pd.api.types.is_numeric_dtype(data[col]):
                # 检查异常值
                Q1 = data[col].quantile(0.25)
                Q3 = data[col].quantile(0.75)
                IQR = Q3 - Q1
                lower_bound = Q1 - 1.5 * IQR
                upper_bound = Q3 + 1.5 * IQR
                outliers = ((data[col] < lower_bound) | (data[col] > upper_bound)).sum()

                if outliers > 0:
                    outlier_percentage = outliers / len(data) * 100
                    col_info['issues'].append(f"异常值: {outliers} ({outlier_percentage:.1f}%)")

            quality_report['columns'][col] = col_info

        # 计算整体质量分数
        quality_report['overall_score'] = max(0, 100 - (total_issues / total_checks) * 100)

        # 生成建议
        if quality_report['overall_score'] < 80:
            quality_report['recommendations'].append("数据质量较低，建议进行数据清洗")

        print(f"   ✓ 数据质量分析完成，质量分数: {quality_report['overall_score']:.1f}")
        return quality_report

    def clean_data(self, data: pd.DataFrame,
                   handle_missing: str = 'auto',
                   handle_outliers: str = 'auto',
                   handle_duplicates: bool = True) -> pd.DataFrame:
        """
        数据清洗

        Parameters:
        - data: 原始数据
        - handle_missing: 缺失值处理方法
        - handle_outliers: 异常值处理方法
        - handle_duplicates: 是否处理重复值

        Returns:
        - 清洗后的数据
        """
        print("🧹 开始数据清洗...")
        cleaned_data = data.copy()
        original_shape = cleaned_data.shape

        # 1. 处理重复值
        if handle_duplicates:
            before_count = len(cleaned_data)
            cleaned_data = cleaned_data.drop_duplicates()
            removed_duplicates = before_count - len(cleaned_data)
            if removed_duplicates > 0:
                print(f"   ✓ 移除了 {removed_duplicates} 个重复行")
                self.preprocessing_steps.append(f"移除重复值: {removed_duplicates} 行")

        # 2. 处理缺失值
        if handle_missing != 'none':
            cleaned_data = self._handle_missing_values(cleaned_data, handle_missing)

        # 3. 处理异常值
        if handle_outliers != 'none':
            cleaned_data = self._handle_outliers(cleaned_data, handle_outliers)

        final_shape = cleaned_data.shape
        print(f"   ✓ 数据清洗完成: {original_shape} -> {final_shape}")

        return cleaned_data

    def _handle_missing_values(self, data: pd.DataFrame, method: str) -> pd.DataFrame:
        """处理缺失值"""
        print("   处理缺失值...")
        cleaned_data = data.copy()

        for col in data.columns:
            missing_percentage = data[col].isnull().sum() / len(data) * 100

            if missing_percentage > 0:
                if missing_percentage > self.config['missing_threshold'] * 100:
                    # 删除缺失值过多的列
                    cleaned_data = cleaned_data.drop(columns=[col])
                    print(f"     - 删除列 '{col}' (缺失值 {missing_percentage:.1f}%)")
                    self.preprocessing_steps.append(f"删除列: {col} (缺失值过多)")
                    continue

                # 根据数据类型选择填充方法
                if method == 'auto':
                    if pd.api.types.is_numeric_dtype(data[col]):
                        fill_method = 'median'
                    else:
                        fill_method = 'mode'
                else:
                    fill_method = method

                if fill_method == 'mean' and pd.api.types.is_numeric_dtype(data[col]):
                    cleaned_data[col] = cleaned_data[col].fillna(cleaned_data[col].mean())
                elif fill_method == 'median' and pd.api.types.is_numeric_dtype(data[col]):
                    cleaned_data[col] = cleaned_data[col].fillna(cleaned_data[col].median())
                elif fill_method == 'mode':
                    mode_value = cleaned_data[col].mode()
                    if len(mode_value) > 0:
                        cleaned_data[col] = cleaned_data[col].fillna(mode_value[0])
                elif fill_method == 'knn' and pd.api.types.is_numeric_dtype(data[col]):
                    # 使用KNN填充
                    imputer = KNNImputer(n_neighbors=5)
                    cleaned_data[[col]] = imputer.fit_transform(cleaned_data[[col]])
                    self.imputers[col] = imputer
                elif fill_method == 'forward':
                    cleaned_data[col] = cleaned_data[col].fillna(method='ffill')
                elif fill_method == 'backward':
                    cleaned_data[col] = cleaned_data[col].fillna(method='bfill')

                print(f"     - 填充列 '{col}' 缺失值 (方法: {fill_method})")

        self.preprocessing_steps.append(f"处理缺失值: {method}")
        return cleaned_data

    def _handle_outliers(self, data: pd.DataFrame, method: str) -> pd.DataFrame:
        """处理异常值"""
        print("   处理异常值...")
        cleaned_data = data.copy()
        outlier_count = 0

        numeric_cols = data.select_dtypes(include=[np.number]).columns

        if method == 'auto':
            method = self.config['outlier_method']

        if method == 'iqr':
            for col in numeric_cols:
                Q1 = data[col].quantile(0.25)
                Q3 = data[col].quantile(0.75)
                IQR = Q3 - Q1
                lower_bound = Q1 - 1.5 * IQR
                upper_bound = Q3 + 1.5 * IQR

                outlier_mask = ((data[col] < lower_bound) | (data[col] > upper_bound))
                col_outliers = outlier_mask.sum()
                if col_outliers > 0:
                    outlier_count += col_outliers
                    # 用边界值替换异常值
                    cleaned_data[col] = np.where(data[col] < lower_bound, lower_bound, data[col])
                    cleaned_data[col] = np.where(data[col] > upper_bound, upper_bound, cleaned_data[col])

        elif method == 'isolation_forest':
            # 使用Isolation Forest检测异常值
            iso_forest = IsolationForest(contamination=self.config['outlier_contamination'],
                                       random_state=self.config['random_state'])

            # 只使用数值列进行检测
            numeric_data = data[numeric_cols].dropna()
            if len(numeric_data) > 0:
                outlier_labels = iso_forest.fit_predict(numeric_data)
                outlier_mask = outlier_labels == -1

                # 移除异常值行
                outlier_indices = numeric_data.index[outlier_mask]
                outlier_count = len(outlier_indices)
                cleaned_data = cleaned_data.drop(outlier_indices)

        print(f"     - 处理了 {outlier_count} 个异常值")
        self.preprocessing_steps.append(f"处理异常值: {method}")
        return cleaned_data

    def transform_data_types(self, data: pd.DataFrame, auto_detect: bool = True) -> pd.DataFrame:
        """
        转换数据类型

        Parameters:
        - data: 数据DataFrame
        - auto_detect: 是否自动检测数据类型

        Returns:
        - 类型转换后的数据
        """
        print("🔄 转换数据类型...")
        transformed_data = data.copy()
        type_conversions = []

        for col in data.columns:
            original_type = str(data[col].dtype)

            if auto_detect:
                # 尝试自动检测最佳类型
                if data[col].dtype == 'object':
                    # 尝试转换为数值类型
                    try:
                        numeric_data = pd.to_numeric(data[col], errors='raise')
                        if (numeric_data % 1 == 0).all():
                            transformed_data[col] = numeric_data.astype('int64')
                            type_conversions.append(f"{col}: {original_type} -> int64")
                        else:
                            transformed_data[col] = numeric_data.astype('float64')
                            type_conversions.append(f"{col}: {original_type} -> float64")
                    except:
                        # 尝试转换为日期时间
                        try:
                            transformed_data[col] = pd.to_datetime(data[col], errors='raise')
                            type_conversions.append(f"{col}: {original_type} -> datetime64")
                        except:
                            # 尝试转换为分类类型
                            unique_ratio = data[col].nunique() / len(data)
                            if unique_ratio < 0.5:  # 如果唯一值比例小于50%
                                transformed_data[col] = data[col].astype('category')
                                type_conversions.append(f"{col}: {original_type} -> category")

        if type_conversions:
            print("   数据类型转换:")
            for conversion in type_conversions:
                print(f"     ✓ {conversion}")
            self.preprocessing_steps.append("数据类型自动转换")

        return transformed_data

    def encode_categorical(self, data: pd.DataFrame, columns: Optional[List[str]] = None,
                          method: str = 'auto') -> pd.DataFrame:
        """
        编码分类变量

        Parameters:
        - data: 数据DataFrame
        - columns: 要编码的列名列表
        - method: 编码方法

        Returns:
        - 编码后的数据
        """
        print("🏷️ 编码分类变量...")
        encoded_data = data.copy()

        if columns is None:
            columns = data.select_dtypes(include=['object', 'category']).columns.tolist()

        if not columns:
            print("   ✓ 没有需要编码的分类变量")
            return encoded_data

        if method == 'auto':
            # 自动选择编码方法
            for col in columns:
                unique_count = data[col].nunique()

                if unique_count == 2:
                    # 二分类变量使用标签编码
                    encoder = LabelEncoder()
                    encoded_data[col] = encoder.fit_transform(data[col].astype(str))
                    self.encoders[col] = encoder
                    print(f"     ✓ 标签编码: {col} ({unique_count} 类别)")

                elif unique_count <= 10:
                    # 少量类别使用独热编码
                    dummies = pd.get_dummies(data[col], prefix=col)
                    encoded_data = pd.concat([encoded_data.drop(columns=[col]), dummies], axis=1)
                    print(f"     ✓ 独热编码: {col} ({unique_count} 类别)")

                else:
                    # 多类别使用标签编码
                    encoder = LabelEncoder()
                    encoded_data[col] = encoder.fit_transform(data[col].astype(str))
                    self.encoders[col] = encoder
                    print(f"     ✓ 标签编码: {col} ({unique_count} 类别)")

        else:
            # 使用指定的编码方法
            if method == 'label':
                for col in columns:
                    encoder = LabelEncoder()
                    encoded_data[col] = encoder.fit_transform(data[col].astype(str))
                    self.encoders[col] = encoder
                    print(f"     ✓ 标签编码: {col}")

            elif method == 'onehot':
                for col in columns:
                    dummies = pd.get_dummies(data[col], prefix=col)
                    encoded_data = pd.concat([encoded_data.drop(columns=[col]), dummies], axis=1)
                    print(f"     ✓ 独热编码: {col}")

        self.preprocessing_steps.append(f"分类变量编码: {method}")
        return encoded_data

    def scale_features(self, data: pd.DataFrame, columns: Optional[List[str]] = None,
                      method: str = None) -> pd.DataFrame:
        """
        特征缩放

        Parameters:
        - data: 数据DataFrame
        - columns: 要缩放的列名列表
        - method: 缩放方法

        Returns:
        - 缩放后的数据
        """
        print("📏 特征缩放...")
        scaled_data = data.copy()

        if columns is None:
            columns = data.select_dtypes(include=[np.number]).columns.tolist()

        if not columns:
            print("   ✓ 没有需要缩放的数值变量")
            return scaled_data

        if method is None:
            method = self.config['scaling_method']

        # 选择缩放器
        if method == 'standard':
            scaler = StandardScaler()
        elif method == 'minmax':
            scaler = MinMaxScaler()
        elif method == 'robust':
            scaler = RobustScaler()
        else:
            raise ValueError(f"不支持的缩放方法: {method}")

        # 应用缩放
        scaled_data[columns] = scaler.fit_transform(data[columns])
        self.scalers['feature_scaler'] = scaler

        print(f"   ✓ 使用 {method} 方法缩放了 {len(columns)} 个特征")
        self.preprocessing_steps.append(f"特征缩放: {method}")

        return scaled_data

    def select_features(self, data: pd.DataFrame, target_col: str,
                       method: str = 'univariate', k: int = None) -> Tuple[pd.DataFrame, List[str]]:
        """
        特征选择

        Parameters:
        - data: 数据DataFrame
        - target_col: 目标列名
        - method: 选择方法
        - k: 选择的特征数量

        Returns:
        - 选择后的数据和特征列表
        """
        print("🎯 特征选择...")

        if target_col not in data.columns:
            print(f"   ⚠️ 目标列 '{target_col}' 不存在，跳过特征选择")
            return data, data.columns.tolist()

        if k is None:
            k = self.config['k_features']

        # 准备数据
        X = data.drop(columns=[target_col])
        y = data[target_col]

        # 只使用数值列进行选择
        numeric_cols = X.select_dtypes(include=[np.number]).columns.tolist()
        X_numeric = X[numeric_cols]

        if len(numeric_cols) == 0:
            print("   ⚠️ 没有数值特征，跳过特征选择")
            return data, data.columns.tolist()

        if method == 'univariate':
            # 单变量统计选择
            if y.dtype == 'object' or len(y.unique()) < 10:
                # 分类问题
                selector = SelectKBest(score_func=f_classif, k=min(k, len(numeric_cols)))
            else:
                # 回归问题
                selector = SelectKBest(score_func=f_regression, k=min(k, len(numeric_cols)))

            X_selected = selector.fit_transform(X_numeric, y)
            selected_features = X_numeric.columns[selector.get_support()].tolist()

        elif method == 'rfe':
            # 递归特征消除
            from sklearn.linear_model import LogisticRegression, LinearRegression

            if y.dtype == 'object' or len(y.unique()) < 10:
                estimator = LogisticRegression(max_iter=1000)
            else:
                estimator = LinearRegression()

            selector = RFE(estimator=estimator, n_features_to_select=min(k, len(numeric_cols)))
            X_selected = selector.fit_transform(X_numeric, y)
            selected_features = X_numeric.columns[selector.get_support()].tolist()

        else:
            raise ValueError(f"不支持的特征选择方法: {method}")

        # 构建选择后的数据
        other_cols = [col for col in data.columns if col not in numeric_cols and col != target_col]
        selected_data = pd.concat([
            data[other_cols],
            pd.DataFrame(X_selected, columns=selected_features, index=data.index),
            data[[target_col]]
        ], axis=1)

        print(f"   ✓ 从 {len(numeric_cols)} 个特征中选择了 {len(selected_features)} 个")
        self.feature_selectors['feature_selector'] = selector
        self.preprocessing_steps.append(f"特征选择: {method} (选择了 {len(selected_features)} 个特征)")

        return selected_data, selected_features

    def split_data(self, data: pd.DataFrame, target_col: str,
                   test_size: float = None, stratify: bool = True) -> Dict:
        """
        数据分割

        Parameters:
        - data: 数据DataFrame
        - target_col: 目标列名
        - test_size: 测试集比例
        - stratify: 是否分层抽样

        Returns:
        - 分割后的数据字典
        """
        print("✂️ 分割数据...")

        if test_size is None:
            test_size = self.config['test_size']

        X = data.drop(columns=[target_col])
        y = data[target_col]

        # 分层抽样参数
        stratify_param = y if stratify and (y.dtype == 'object' or len(y.unique()) < 100) else None

        # 分割数据
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=test_size, random_state=self.config['random_state'],
            stratify=stratify_param
        )

        split_info = {
            'X_train': X_train,
            'X_test': X_test,
            'y_train': y_train,
            'y_test': y_test,
            'train_size': len(X_train),
            'test_size': len(X_test),
            'train_ratio': len(X_train) / len(data),
            'test_ratio': len(X_test) / len(data),
            'feature_count': X.shape[1],
            'target_classes': y.nunique() if y.dtype == 'object' else 'continuous'
        }

        print(f"   ✓ 训练集: {len(X_train)} 样本 ({len(X_train)/len(data):.1%})")
        print(f"   ✓ 测试集: {len(X_test)} 样本 ({len(X_test)/len(data):.1%})")

        self.preprocessing_steps.append(f"数据分割: 测试集比例 {test_size}")
        return split_info

    def balance_data(self, X_train: pd.DataFrame, y_train: pd.DataFrame,
                    method: str = None) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        平衡数据

        Parameters:
        - X_train: 训练特征
        - y_train: 训练标签
        - method: 平衡方法

        Returns:
        - 平衡后的数据
        """
        if method is None:
            method = self.config['balance_method']

        # 检查是否需要平衡
        if y_train.dtype == 'object' or len(y_train.unique()) < 100:
            class_counts = y_train.value_counts()
            min_count = class_counts.min()
            max_count = class_counts.max()

            if max_count / min_count <= 2:  # 如果类别比例小于2:1，认为已经平衡
                print("   ✓ 数据已经平衡，无需处理")
                return X_train, y_train

        print(f"⚖️ 平衡数据 (方法: {method})...")

        if method == 'smote':
            # SMOTE过采样
            smote = SMOTE(random_state=self.config['random_state'])
            X_balanced, y_balanced = smote.fit_resample(X_train, y_train)

        elif method == 'oversample':
            # 随机过采样
            ros = RandomOverSampler(random_state=self.config['random_state'])
            X_balanced, y_balanced = ros.fit_resample(X_train, y_train)

        elif method == 'undersample':
            # 随机欠采样
            rus = RandomUnderSampler(random_state=self.config['random_state'])
            X_balanced, y_balanced = rus.fit_resample(X_train, y_train)

        else:
            raise ValueError(f"不支持的平衡方法: {method}")

        print(f"   ✓ 平衡前: {X_train.shape[0]} 样本")
        print(f"   ✓ 平衡后: {X_balanced.shape[0]} 样本")

        # 转换回DataFrame
        if hasattr(X_balanced, 'toarray'):
            X_balanced = pd.DataFrame(X_balanced.toarray(), columns=X_train.columns)
        else:
            X_balanced = pd.DataFrame(X_balanced, columns=X_train.columns)

        y_balanced = pd.Series(y_balanced, name=y_train.name)

        self.preprocessing_steps.append(f"数据平衡: {method}")
        return X_balanced, y_balanced

    def auto_preprocess(self, data: pd.DataFrame, target_col: str,
                       save_report: bool = True) -> Dict:
        """
        自动化预处理流程

        Parameters:
        - data: 原始数据
        - target_col: 目标列名
        - save_report: 是否保存报告

        Returns:
        - 预处理结果字典
        """
        print("🚀 开始自动化数据预处理...")

        results = {
            'original_data': data,
            'preprocessed_data': None,
            'X_train': None,
            'X_test': None,
            'y_train': None,
            'y_test': None,
            'quality_report': None,
            'preprocessing_steps': [],
            'feature_info': {}
        }

        # 1. 数据质量分析
        quality_report = self.analyze_data_quality(data)
        results['quality_report'] = quality_report

        # 2. 数据清洗
        cleaned_data = self.clean_data(data)
        results['preprocessing_steps'].extend(self.preprocessing_steps)

        # 3. 数据类型转换
        transformed_data = self.transform_data_types(cleaned_data)

        # 4. 特征工程
        # 这里可以添加更多的特征工程步骤
        engineered_data = self._feature_engineering(transformed_data, target_col)

        # 5. 编码分类变量
        encoded_data = self.encode_categorical(engineered_data)

        # 6. 特征缩放
        numeric_cols = encoded_data.select_dtypes(include=[np.number]).columns.tolist()
        if target_col in numeric_cols:
            numeric_cols.remove(target_col)

        if numeric_cols:
            scaled_data = self.scale_features(encoded_data, numeric_cols)
        else:
            scaled_data = encoded_data

        results['preprocessed_data'] = scaled_data

        # 7. 特征选择（可选）
        if self.config['feature_selection'] and target_col in scaled_data.columns:
            selected_data, selected_features = self.select_features(
                scaled_data, target_col, k=self.config['k_features']
            )
            results['feature_info']['selected_features'] = selected_features
            final_data = selected_data
        else:
            final_data = scaled_data
            results['feature_info']['all_features'] = [
                col for col in final_data.columns if col != target_col
            ]

        # 8. 数据分割
        if target_col in final_data.columns:
            split_result = self.split_data(final_data, target_col)
            results.update(split_result)

        # 9. 数据平衡（可选，仅对分类问题）
        if (self.config['balance_data'] and
            target_col in final_data.columns and
            (results['y_train'].dtype == 'object' or len(results['y_train'].unique()) < 100)):

            X_balanced, y_balanced = self.balance_data(
                results['X_train'], results['y_train'], method=self.config['balance_method']
            )
            results['X_train'] = X_balanced
            results['y_train'] = y_balanced

        # 保存预处理报告
        if save_report:
            self.preprocessing_report = {
                'timestamp': pd.Timestamp.now().isoformat(),
                'original_shape': data.shape,
                'final_shape': results['preprocessed_data'].shape,
                'preprocessing_steps': self.preprocessing_steps,
                'quality_score': quality_report['overall_score'],
                'feature_count': len(results['feature_info'].get('selected_features',
                                                results['feature_info'].get('all_features', [])))
            }

        print(f"\n🎉 自动化预处理完成！")
        print(f"   原始数据: {data.shape}")
        print(f"   预处理后: {results['preprocessed_data'].shape}")
        print(f"   预处理步骤: {len(self.preprocessing_steps)}")

        return results

    def _feature_engineering(self, data: pd.DataFrame, target_col: str) -> pd.DataFrame:
        """基础特征工程"""
        engineered_data = data.copy()
        numeric_cols = data.select_dtypes(include=[np.number]).columns.tolist()

        # 创建交互特征（对于数值变量）
        if len(numeric_cols) >= 2:
            # 选择前几个重要变量创建交互项
            important_cols = numeric_cols[:min(3, len(numeric_cols))]

            for i, col1 in enumerate(important_cols):
                for col2 in important_cols[i+1:]:
                    # 乘积特征
                    engineered_data[f'{col1}_x_{col2}'] = data[col1] * data[col2]
                    # 比值特征（避免除零）
                    engineered_data[f'{col1}_div_{col2}'] = np.where(
                        data[col2] != 0, data[col1] / data[col2], 0
                    )

        # 创建多项式特征（对于重要变量）
        if len(important_cols) > 0:
            for col in important_cols[:2]:  # 只为前两个变量创建
                engineered_data[f'{col}_squared'] = data[col] ** 2
                engineered_data[f'{col}_sqrt'] = np.sqrt(np.abs(data[col]))

        print(f"   ✓ 创建了 {engineered_data.shape[1] - data.shape[1]} 个新特征")
        return engineered_data

    def get_preprocessing_summary(self) -> Dict:
        """
        获取预处理摘要

        Returns:
        - 预处理摘要信息
        """
        return {
            'preprocessing_steps': self.preprocessing_steps,
            'scalers': list(self.scalers.keys()),
            'encoders': list(self.encoders.keys()),
            'imputers': list(self.imputers.keys()),
            'feature_selectors': list(self.feature_selectors.keys()),
            'preprocessing_report': getattr(self, 'preprocessing_report', {}),
            'config': self.config
        }

    def save_preprocessing_objects(self, output_dir: str):
        """
        保存预处理对象

        Parameters:
        - output_dir: 输出目录
        """
        output_path = Path(output_dir)
        output_path.mkdir(exist_ok=True)

        # 保存scalers
        if self.scalers:
            import joblib
            for name, scaler in self.scalers.items():
                joblib.dump(scaler, output_path / f"{name}.pkl")

        # 保存encoders
        if self.encoders:
            import joblib
            for name, encoder in self.encoders.items():
                joblib.dump(encoder, output_path / f"{name}_encoder.pkl")

        # 保存预处理报告
        if hasattr(self, 'preprocessing_report'):
            with open(output_path / 'preprocessing_report.json', 'w', encoding='utf-8') as f:
                json.dump(self.preprocessing_report, f, ensure_ascii=False, indent=2)

        print(f"✅ 预处理对象已保存到 {output_dir}")

    def load_preprocessing_objects(self, input_dir: str):
        """
        加载预处理对象

        Parameters:
        - input_dir: 输入目录
        """
        input_path = Path(input_dir)

        # 加载scalers
        import joblib
        for scaler_file in input_path.glob("*.pkl"):
            if not str(scaler_file).endswith('_encoder.pkl'):
                name = scaler_file.stem
                self.scalers[name] = joblib.load(scaler_file)

        # 加载encoders
        for encoder_file in input_path.glob("*_encoder.pkl"):
            name = encoder_file.stem.replace('_encoder', '')
            self.encoders[name] = joblib.load(encoder_file)

        print(f"✅ 预处理对象已从 {input_dir} 加载")

    def transform_new_data(self, new_data: pd.DataFrame) -> pd.DataFrame:
        """
        对新数据应用相同的预处理

        Parameters:
        - new_data: 新数据

        Returns:
        - 预处理后的新数据
        """
        transformed_data = new_data.copy()

        # 应用缺失值处理
        for col, imputer in self.imputers.items():
            if col in transformed_data.columns:
                if hasattr(imputer, 'transform'):
                    transformed_data[[col]] = imputer.transform(transformed_data[[col]])

        # 应用编码
        for col, encoder in self.encoders.items():
            if col in transformed_data.columns:
                transformed_data[col] = encoder.transform(transformed_data[col].astype(str))

        # 应用缩放
        if 'feature_scaler' in self.scalers:
            numeric_cols = transformed_data.select_dtypes(include=[np.number]).columns.tolist()
            if numeric_cols:
                transformed_data[numeric_cols] = self.scalers['feature_scaler'].transform(
                    transformed_data[numeric_cols]
                )

        return transformed_data

"""
数据探索分析器 (EDA Analyzer) - 自动化探索性数据分析核心模块

提供全面的EDA功能，包括：
- 自动数据质量检查
- 统计描述分析
- 异常值检测
- 相关性分析
- 数据分布分析
- 智能洞察生成
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional, Union, Any
import warnings
from scipy import stats
from scipy.stats import normaltest, shapiro, anderson, jarque_bera
import json
from datetime import datetime

warnings.filterwarnings('ignore')


class EDAAnalyzer:
    """数据探索分析器 - 自动化EDA核心引擎"""

    def __init__(self, config: Optional[Dict] = None):
        """
        初始化EDA分析器

        Parameters:
        - config: 配置参数字典
        """
        self.config = config or {}
        self.data = None
        self.results = {}
        self.report = None

        # 默认配置
        self.default_config = {
            'missing_threshold': 0.05,  # 缺失值阈值
            'outlier_method': 'iqr',    # 异常值检测方法
            'correlation_method': 'pearson',  # 相关性计算方法
            'significance_level': 0.05,  # 显著性水平
            'sample_size_threshold': 10000,  # 大数据集阈值
            'encoding_detection': True,  # 自动编码检测
            'chinese_support': True     # 中文支持
        }

        # 合并配置
        self.config = {**self.default_config, **self.config}

    def load_data(self, data_path: str, **kwargs) -> pd.DataFrame:
        """
        加载数据文件

        Parameters:
        - data_path: 数据文件路径
        - **kwargs: pandas.read_csv的额外参数

        Returns:
        - 加载的DataFrame
        """
        try:
            # 自动检测文件格式
            if data_path.endswith('.csv'):
                # 自动检测编码
                if self.config['encoding_detection']:
                    encodings = ['utf-8', 'gbk', 'gb2312', 'utf-8-sig', 'latin1']
                    for encoding in encodings:
                        try:
                            data = pd.read_csv(data_path, encoding=encoding, **kwargs)
                            print(f"✅ 数据加载成功，使用编码: {encoding}")
                            break
                        except UnicodeDecodeError:
                            continue
                    else:
                        raise ValueError("无法检测到正确的文件编码")
                else:
                    data = pd.read_csv(data_path, **kwargs)

            elif data_path.endswith(('.xlsx', '.xls')):
                data = pd.read_excel(data_path, **kwargs)
                print("✅ Excel文件加载成功")

            elif data_path.endswith('.json'):
                data = pd.read_json(data_path, **kwargs)
                print("✅ JSON文件加载成功")

            else:
                raise ValueError("不支持的文件格式，请使用CSV、Excel或JSON格式")

            self.data = data
            self._log_basic_info()

            return data

        except Exception as e:
            print(f"❌ 数据加载失败: {str(e)}")
            raise

    def _log_basic_info(self):
        """记录数据基本信息"""
        if self.data is not None:
            print(f"📊 数据基本信息:")
            print(f"   - 数据形状: {self.data.shape}")
            print(f"   - 内存使用: {self.data.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
            print(f"   - 列数: {len(self.data.columns)}")
            print(f"   - 数据类型分布:\n{self.data.dtypes.value_counts()}")

    def data_quality_check(self, data: Optional[pd.DataFrame] = None) -> Dict:
        """
        数据质量检查

        Parameters:
        - data: 待检查的数据

        Returns:
        - 数据质量报告
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        quality_report = {
            'basic_info': {
                'shape': data.shape,
                'memory_usage_mb': data.memory_usage(deep=True).sum() / 1024**2,
                'total_cells': data.shape[0] * data.shape[1]
            },
            'missing_values': {},
            'data_types': {},
            'duplicates': {},
            'outliers': {},
            'data_quality_score': 0,
            'issues': [],
            'recommendations': []
        }

        # 1. 缺失值分析
        missing_analysis = self._analyze_missing_values(data)
        quality_report['missing_values'] = missing_analysis

        # 2. 数据类型分析
        dtype_analysis = self._analyze_data_types(data)
        quality_report['data_types'] = dtype_analysis

        # 3. 重复值分析
        duplicate_analysis = self._analyze_duplicates(data)
        quality_report['duplicates'] = duplicate_analysis

        # 4. 异常值分析（仅数值列）
        outlier_analysis = self._analyze_outliers(data)
        quality_report['outliers'] = outlier_analysis

        # 5. 计算数据质量分数
        quality_score = self._calculate_quality_score(quality_report)
        quality_report['data_quality_score'] = quality_score

        # 6. 生成问题和建议
        issues, recommendations = self._generate_quality_recommendations(quality_report)
        quality_report['issues'] = issues
        quality_report['recommendations'] = recommendations

        self.results['data_quality'] = quality_report
        return quality_report

    def _analyze_missing_values(self, data: pd.DataFrame) -> Dict:
        """分析缺失值"""
        missing_info = {}

        # 总体缺失值情况
        total_missing = data.isnull().sum().sum()
        total_cells = data.shape[0] * data.shape[1]
        overall_missing_rate = total_missing / total_cells

        missing_info['overall'] = {
            'total_missing': total_missing,
            'total_cells': total_cells,
            'missing_rate': overall_missing_rate,
            'quality_rating': self._rate_missing_quality(overall_missing_rate)
        }

        # 各列缺失值详情
        missing_columns = {}
        for col in data.columns:
            missing_count = data[col].isnull().sum()
            if missing_count > 0:
                missing_rate = missing_count / len(data)
                missing_columns[col] = {
                    'missing_count': missing_count,
                    'missing_rate': missing_rate,
                    'severity': self._classify_missing_severity(missing_rate)
                }

        missing_info['columns'] = missing_columns
        missing_info['missing_columns_count'] = len(missing_columns)

        # 缺失值模式分析
        missing_patterns = self._analyze_missing_patterns(data)
        missing_info['patterns'] = missing_patterns

        return missing_info

    def _analyze_data_types(self, data: pd.DataFrame) -> Dict:
        """分析数据类型"""
        dtype_info = {
            'type_distribution': data.dtypes.value_counts().to_dict(),
            'numeric_columns': list(data.select_dtypes(include=[np.number]).columns),
            'categorical_columns': list(data.select_dtypes(include=['object', 'category']).columns),
            'datetime_columns': list(data.select_dtypes(include=['datetime64']).columns),
            'type_issues': []
        }

        # 检测可能的类型问题
        for col in data.columns:
            if data[col].dtype == 'object':
                # 检查是否应该是数值类型
                try:
                    pd.to_numeric(data[col], errors='raise')
                    dtype_info['type_issues'].append({
                        'column': col,
                        'issue': '可能应该是数值类型',
                        'suggestion': '尝试转换为数值类型'
                    })
                except:
                    pass

                # 检查是否应该是日期类型
                if data[col].dtype == 'object':
                    sample_values = data[col].dropna().head(5).astype(str)
                    if any('/' in val or '-' in val for val in sample_values if val):
                        try:
                            pd.to_datetime(data[col], errors='raise')
                            dtype_info['type_issues'].append({
                                'column': col,
                                'issue': '可能应该是日期类型',
                                'suggestion': '尝试转换为日期类型'
                            })
                        except:
                            pass

        return dtype_info

    def _analyze_duplicates(self, data: pd.DataFrame) -> Dict:
        """分析重复值"""
        duplicate_info = {}

        # 完全重复行
        duplicate_rows = data.duplicated().sum()
        duplicate_info['exact_duplicates'] = {
            'count': duplicate_rows,
            'rate': duplicate_rows / len(data),
            'severity': self._classify_duplicate_severity(duplicate_rows / len(data))
        }

        # 基于关键字段的重复（如果存在ID列）
        id_columns = [col for col in data.columns if any(keyword in col.lower()
                      for keyword in ['id', '编号', 'code', '标识'])]

        if id_columns:
            for id_col in id_columns[:3]:  # 最多检查3个ID列
                id_duplicates = data[id_col].duplicated().sum()
                duplicate_info[f'{id_col}_duplicates'] = {
                    'count': id_duplicates,
                    'rate': id_duplicates / len(data),
                    'severity': self._classify_duplicate_severity(id_duplicates / len(data))
                }

        return duplicate_info

    def _analyze_outliers(self, data: pd.DataFrame) -> Dict:
        """分析异常值"""
        outlier_info = {}
        numeric_columns = data.select_dtypes(include=[np.number]).columns

        method = self.config['outlier_method']

        for col in numeric_columns:
            outliers = self._detect_outliers(data[col], method)
            if len(outliers) > 0:
                outlier_info[col] = {
                    'outlier_count': len(outliers),
                    'outlier_rate': len(outliers) / len(data),
                    'outlier_indices': outliers.tolist(),
                    'severity': self._classify_outlier_severity(len(outliers) / len(data))
                }

        outlier_info['total_outlier_columns'] = len(outlier_info)
        outlier_info['detection_method'] = method

        return outlier_info

    def _detect_outliers(self, series: pd.Series, method: str = 'iqr') -> np.ndarray:
        """检测异常值"""
        series_clean = series.dropna()

        if method == 'iqr':
            Q1 = series_clean.quantile(0.25)
            Q3 = series_clean.quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            outliers = series_clean[(series_clean < lower_bound) | (series_clean > upper_bound)]

        elif method == 'zscore':
            z_scores = np.abs(stats.zscore(series_clean))
            outliers = series_clean[z_scores > 3]

        elif method == 'modified_zscore':
            median = series_clean.median()
            mad = np.median(np.abs(series_clean - median))
            modified_z_scores = 0.6745 * (series_clean - median) / mad
            outliers = series_clean[np.abs(modified_z_scores) > 3.5]

        else:
            raise ValueError(f"不支持的异常值检测方法: {method}")

        return outliers.index.values

    def generate_statistical_summary(self, data: Optional[pd.DataFrame] = None) -> Dict:
        """
        生成统计描述摘要

        Parameters:
        - data: 待分析的数据

        Returns:
        - 统计摘要
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        summary = {
            'dataset_info': {
                'shape': data.shape,
                'columns_count': len(data.columns),
                'memory_usage_mb': data.memory_usage(deep=True).sum() / 1024**2
            },
            'numeric_summary': {},
            'categorical_summary': {},
            'distribution_tests': {},
            'correlation_analysis': {}
        }

        # 数值列统计摘要
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) > 0:
            numeric_stats = data[numeric_cols].describe()

            # 添加额外的统计指标
            for col in numeric_cols:
                if data[col].dtype in [np.number]:
                    extra_stats = {
                        'skewness': stats.skew(data[col].dropna()),
                        'kurtosis': stats.kurtosis(data[col].dropna()),
                        'cv': stats.variation(data[col].dropna()),  # 变异系数
                        'missing_rate': data[col].isnull().sum() / len(data),
                        'unique_count': data[col].nunique()
                    }

                    # 正态性检验（如果样本量合适）
                    if len(data[col].dropna()) >= 8 and len(data[col].dropna()) <= 5000:
                        try:
                            # Shapiro-Wilk检验（小样本）
                            if len(data[col].dropna()) <= 50:
                                stat, p_value = shapiro(data[col].dropna())
                                test_name = 'Shapiro-Wilk'
                            else:
                                # D'Agostino's K-squared检验（大样本）
                                stat, p_value = normaltest(data[col].dropna())
                                test_name = "D'Agostino's K-squared"

                            summary['distribution_tests'][col] = {
                                'test_name': test_name,
                                'statistic': stat,
                                'p_value': p_value,
                                'is_normal': p_value > self.config['significance_level'],
                                'interpretation': '正态分布' if p_value > self.config['significance_level'] else '非正态分布'
                            }
                        except:
                            summary['distribution_tests'][col] = {
                                'test_name': 'Failed',
                                'reason': '无法执行正态性检验'
                            }

                    summary['numeric_summary'][col] = {
                        **numeric_stats[col].to_dict(),
                        **extra_stats
                    }

        # 分类列统计摘要
        categorical_cols = data.select_dtypes(include=['object', 'category']).columns
        if len(categorical_cols) > 0:
            for col in categorical_cols:
                cat_stats = {
                    'unique_count': data[col].nunique(),
                    'most_frequent': data[col].mode().iloc[0] if not data[col].mode().empty else None,
                    'most_frequent_count': data[col].value_counts().iloc[0] if len(data[col].value_counts()) > 0 else 0,
                    'missing_rate': data[col].isnull().sum() / len(data),
                    'value_counts': data[col].value_counts().head(10).to_dict()  # 前10个值
                }

                # 分类变量的均匀性检验
                if data[col].nunique() <= 10:  # 只对类别数较少的变量进行检验
                    try:
                        observed = data[col].value_counts().values
                        expected = [len(data) / len(observed)] * len(observed)
                        chi2_stat, p_value = stats.chisquare(observed, expected)

                        cat_stats['uniformity_test'] = {
                            'chi2_statistic': chi2_stat,
                            'p_value': p_value,
                            'is_uniform': p_value > self.config['significance_level'],
                            'interpretation': '分布均匀' if p_value > self.config['significance_level'] else '分布不均匀'
                        }
                    except:
                        cat_stats['uniformity_test'] = {
                            'test_name': 'Failed',
                            'reason': '无法执行均匀性检验'
                        }

                summary['categorical_summary'][col] = cat_stats

        # 相关性分析
        if len(numeric_cols) > 1:
            correlation_matrix = data[numeric_cols].corr(method=self.config['correlation_method'])

            # 找出高相关性的变量对
            high_corr_pairs = []
            for i in range(len(correlation_matrix.columns)):
                for j in range(i+1, len(correlation_matrix.columns)):
                    corr_val = correlation_matrix.iloc[i, j]
                    if abs(corr_val) > 0.7:  # 高相关性阈值
                        high_corr_pairs.append({
                            'variable1': correlation_matrix.columns[i],
                            'variable2': correlation_matrix.columns[j],
                            'correlation': corr_val,
                            'strength': self._interpret_correlation_strength(abs(corr_val))
                        })

            summary['correlation_analysis'] = {
                'correlation_matrix': correlation_matrix.to_dict(),
                'method': self.config['correlation_method'],
                'high_correlation_pairs': high_corr_pairs,
                'max_correlation': correlation_matrix.abs().max().max(),
                'avg_correlation': correlation_matrix.abs().mean().mean()
            }

        self.results['statistical_summary'] = summary
        return summary

    def auto_eda(self, data: Optional[pd.DataFrame] = None) -> Dict:
        """
        自动化EDA分析

        Parameters:
        - data: 待分析的数据

        Returns:
        - 完整的EDA分析结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        print("🚀 开始自动化EDA分析...")

        # 1. 数据质量检查
        print("   1. 数据质量检查...")
        quality_report = self.data_quality_check(data)

        # 2. 统计摘要分析
        print("   2. 统计摘要分析...")
        stats_summary = self.generate_statistical_summary(data)

        # 3. 生成洞察
        print("   3. 生成数据洞察...")
        insights = self._generate_insights(quality_report, stats_summary)

        # 4. 数据建议
        print("   4. 生成处理建议...")
        recommendations = self._generate_recommendations(quality_report, stats_summary)

        # 综合结果
        eda_results = {
            'analysis_time': datetime.now().isoformat(),
            'data_quality': quality_report,
            'statistical_summary': stats_summary,
            'insights': insights,
            'recommendations': recommendations,
            'next_steps': self._suggest_next_steps(quality_report, stats_summary)
        }

        self.results['auto_eda'] = eda_results
        self._print_eda_summary(eda_results)

        return eda_results

    def _generate_insights(self, quality_report: Dict, stats_summary: Dict) -> List[str]:
        """生成数据洞察"""
        insights = []

        # 数据质量洞察
        quality_score = quality_report['data_quality_score']
        if quality_score >= 0.8:
            insights.append("数据质量优秀，适合直接进行分析")
        elif quality_score >= 0.6:
            insights.append("数据质量良好，但存在一些需要关注的问题")
        else:
            insights.append("数据质量需要改善，建议进行数据清洗")

        # 缺失值洞察
        missing_rate = quality_report['missing_values']['overall']['missing_rate']
        if missing_rate > 0.1:
            insights.append(f"数据存在较高缺失率({missing_rate:.1%})，需要针对性处理")
        elif missing_rate > 0.05:
            insights.append(f"数据存在少量缺失值({missing_rate:.1%})，可考虑填充或删除")

        # 数据分布洞察
        normal_distributions = sum(1 for test in stats_summary['distribution_tests'].values()
                                 if test.get('is_normal', False))
        total_tests = len(stats_summary['distribution_tests'])

        if total_tests > 0:
            normal_ratio = normal_distributions / total_tests
            if normal_ratio > 0.7:
                insights.append("大部分数值变量呈正态分布，适合参数统计方法")
            elif normal_ratio < 0.3:
                insights.append("大部分数值变量呈非正态分布，建议使用非参数方法")

        # 相关性洞察
        if 'correlation_analysis' in stats_summary:
            high_corr_count = len(stats_summary['correlation_analysis']['high_correlation_pairs'])
            if high_corr_count > 0:
                insights.append(f"发现{high_corr_count}对高度相关的变量，可能存在多重共线性问题")

        return insights

    def _generate_recommendations(self, quality_report: Dict, stats_summary: Dict) -> List[str]:
        """生成处理建议"""
        recommendations = []

        # 基于质量报告的建议
        recommendations.extend(quality_report.get('recommendations', []))

        # 基于统计摘要的建议
        # 异常值处理建议
        outlier_columns = len(quality_report['outliers'])
        if outlier_columns > 0:
            recommendations.append(f"发现{outlier_columns}个变量存在异常值，建议进行异常值处理")

        # 数据类型建议
        type_issues = quality_report['data_types']['type_issues']
        if type_issues:
            recommendations.append("检测到数据类型问题，建议检查并修正数据类型")

        # 正态性建议
        for col, test in stats_summary['distribution_tests'].items():
            if not test.get('is_normal', False) and test.get('p_value') is not None:
                recommendations.append(f"变量'{col}'非正态分布，考虑数据变换或使用非参数方法")

        return recommendations

    def _suggest_next_steps(self, quality_report: Dict, stats_summary: Dict) -> List[str]:
        """建议下一步操作"""
        next_steps = []

        # 基于数据质量确定优先级
        quality_score = quality_report['data_quality_score']

        if quality_score < 0.6:
            next_steps.extend([
                "1. 优先进行数据清洗和质量改善",
                "2. 处理缺失值和异常值",
                "3. 修正数据类型问题"
            ])
        else:
            next_steps.extend([
                "1. 进行数据可视化分析",
                "2. 探索变量间关系",
                "3. 考虑特征工程"
            ])

        # 基于数据特征建议分析方法
        numeric_cols = len(stats_summary['numeric_summary'])
        categorical_cols = len(stats_summary['categorical_summary'])

        if numeric_cols > 5:
            next_steps.append("建议进行降维分析（如PCA）")

        if categorical_cols > 3:
            next_steps.append("建议进行编码转换处理")

        # 相关性分析建议
        if 'correlation_analysis' in stats_summary:
            high_corr_pairs = stats_summary['correlation_analysis']['high_correlation_pairs']
            if high_corr_pairs:
                next_steps.append("处理高相关性变量以避免多重共线性")

        return next_steps

    def _print_eda_summary(self, eda_results: Dict):
        """打印EDA分析摘要"""
        print("\n" + "="*60)
        print("🎉 自动化EDA分析完成!")
        print("="*60)

        quality_score = eda_results['data_quality']['data_quality_score']
        print(f"\n📊 数据质量评分: {quality_score:.2f}/1.00")

        print(f"\n💡 主要洞察:")
        for insight in eda_results['insights']:
            print(f"   • {insight}")

        print(f"\n📋 关键建议:")
        for rec in eda_results['recommendations'][:5]:  # 显示前5个建议
            print(f"   • {rec}")

        print(f"\n🔄 下一步操作:")
        for step in eda_results['next_steps']:
            print(f"   • {step}")

    # 辅助方法
    def _rate_missing_quality(self, missing_rate: float) -> str:
        """评估缺失值质量"""
        if missing_rate == 0:
            return "完美"
        elif missing_rate < 0.01:
            return "优秀"
        elif missing_rate < 0.05:
            return "良好"
        elif missing_rate < 0.15:
            return "一般"
        else:
            return "较差"

    def _classify_missing_severity(self, missing_rate: float) -> str:
        """分类缺失值严重程度"""
        if missing_rate < 0.01:
            return "很低"
        elif missing_rate < 0.05:
            return "低"
        elif missing_rate < 0.15:
            return "中等"
        elif missing_rate < 0.30:
            return "高"
        else:
            return "极高"

    def _classify_duplicate_severity(self, duplicate_rate: float) -> str:
        """分类重复值严重程度"""
        if duplicate_rate < 0.01:
            return "很低"
        elif duplicate_rate < 0.05:
            return "低"
        elif duplicate_rate < 0.10:
            return "中等"
        else:
            return "高"

    def _classify_outlier_severity(self, outlier_rate: float) -> str:
        """分类异常值严重程度"""
        if outlier_rate < 0.01:
            return "很低"
        elif outlier_rate < 0.05:
            return "低"
        elif outlier_rate < 0.10:
            return "中等"
        else:
            return "高"

    def _analyze_missing_patterns(self, data: pd.DataFrame) -> Dict:
        """分析缺失值模式"""
        # 计算缺失值模式
        missing_matrix = data.isnull()

        # 找出常见的缺失值组合
        missing_combinations = {}

        # 如果数据太大，进行采样
        if len(data) > self.config['sample_size_threshold']:
            sample_data = data.sample(min(self.config['sample_size_threshold'], len(data)))
        else:
            sample_data = data

        missing_combinations = sample_data.isnull().value_counts().head(10).to_dict()

        return {
            'common_patterns': {str(k): v for k, v in missing_combinations.items()},
            'total_patterns': len(missing_combinations)
        }

    def _calculate_quality_score(self, quality_report: Dict) -> float:
        """计算数据质量分数"""
        score = 1.0

        # 缺失值扣分
        missing_rate = quality_report['missing_values']['overall']['missing_rate']
        score -= missing_rate * 0.5

        # 重复值扣分
        duplicate_rate = quality_report['duplicates']['exact_duplicates']['rate']
        score -= duplicate_rate * 0.3

        # 异常值扣分
        outlier_cols = quality_report['outliers']['total_outlier_columns']
        if outlier_cols > 0:
            outlier_rates = []
            for key, value in quality_report['outliers'].items():
                if key not in ['total_outlier_columns', 'detection_method']:
                    outlier_rates.append(value['outlier_rate'])
            if outlier_rates:
                avg_outlier_rate = np.mean(outlier_rates)
                score -= avg_outlier_rate * 0.2

        # 数据类型问题扣分
        type_issues = len(quality_report['data_types']['type_issues'])
        score -= type_issues * 0.1

        return max(0, score)

    def _generate_quality_recommendations(self, quality_report: Dict) -> Tuple[List[str], List[str]]:
        """生成质量相关的建议"""
        issues = []
        recommendations = []

        # 缺失值问题
        missing_rate = quality_report['missing_values']['overall']['missing_rate']
        if missing_rate > 0.05:
            issues.append(f"数据缺失率较高({missing_rate:.1%})")
            recommendations.append("建议使用适当的填充策略处理缺失值")

        # 重复值问题
        duplicate_rate = quality_report['duplicates']['exact_duplicates']['rate']
        if duplicate_rate > 0.01:
            issues.append(f"存在重复数据({duplicate_rate:.1%})")
            recommendations.append("建议检查并处理重复数据")

        # 异常值问题
        outlier_cols = quality_report['outliers']['total_outlier_columns']
        if outlier_cols > 0:
            issues.append(f"发现{outlier_cols}个变量存在异常值")
            recommendations.append("建议检查异常值并决定处理策略")

        # 数据类型问题
        type_issues = quality_report['data_types']['type_issues']
        if type_issues:
            issues.append("检测到数据类型问题")
            recommendations.append("建议修正数据类型以获得更好的分析结果")

        return issues, recommendations

    def _interpret_correlation_strength(self, corr_value: float) -> str:
        """解释相关性强度"""
        abs_corr = abs(corr_value)
        if abs_corr >= 0.9:
            return "极强"
        elif abs_corr >= 0.7:
            return "强"
        elif abs_corr >= 0.5:
            return "中等"
        elif abs_corr >= 0.3:
            return "弱"
        else:
            return "极弱"

    def export_results(self, output_path: str, format: str = 'json'):
        """
        导出分析结果

        Parameters:
        - output_path: 输出路径
        - format: 输出格式 ('json', 'csv')
        """
        if not self.results:
            print("❌ 没有可导出的分析结果")
            return

        if format == 'json':
            with open(output_path, 'w', encoding='utf-8') as f:
                json.dump(self.results, f, ensure_ascii=False, indent=2, default=str)
            print(f"✅ 分析结果已导出到 {output_path}")

        elif format == 'csv':
            # 导出为CSV格式（主要用于统计摘要）
            if 'statistical_summary' in self.results:
                # 数值统计
                if self.results['statistical_summary']['numeric_summary']:
                    numeric_df = pd.DataFrame(self.results['statistical_summary']['numeric_summary']).T
                    numeric_df.to_csv(output_path.replace('.csv', '_numeric.csv'), encoding='utf-8-sig')

                # 分类统计
                if self.results['statistical_summary']['categorical_summary']:
                    categorical_df = pd.DataFrame(self.results['statistical_summary']['categorical_summary']).T
                    categorical_df.to_csv(output_path.replace('.csv', '_categorical.csv'), encoding='utf-8-sig')

                print(f"✅ 统计摘要已导出到CSV文件")
        else:
            raise ValueError("不支持的输出格式，请使用 'json' 或 'csv'")

#!/usr/bin/env python3
"""
金融数据分析示例
演示如何使用数据探索可视化技能进行金融数据分析
"""

import sys
import os
import pandas as pd
import numpy as np
from pathlib import Path

# 添加技能路径
skill_path = Path(__file__).parent.parent
sys.path.append(str(skill_path / "scripts"))

def create_financial_sample_data():
    """创建金融数据样本"""
    print("💰 创建金融数据样本...")

    np.random.seed(123)
    n_customers = 1000

    # 基础客户信息
    data = {
        'customer_id': [f'C{i:06d}' for i in range(1, n_customers + 1)],
        'age': np.random.randint(18, 80, n_customers),
        'gender': np.random.choice(['男', '女'], n_customers, p=[0.52, 0.48]),
        'income': np.random.lognormal(10.5, 0.5, n_customers),  # 收入分布
        'credit_score': np.random.normal(650, 100, n_customers),
        'employment_years': np.random.exponential(5, n_customers),
        'home_ownership': np.random.choice(['租房', '按揭', '自有'], n_customers, p=[0.35, 0.45, 0.20]),
        'marital_status': np.random.choice(['单身', '已婚', '离异'], n_customers, p=[0.3, 0.6, 0.1]),
        'education_level': np.random.choice(['高中', '本科', '硕士', '博士'], n_customers, p=[0.3, 0.4, 0.25, 0.05]),
        'debt_to_income_ratio': np.random.beta(2, 5, n_customers),  # 负债收入比
        'savings_amount': np.random.exponential(10000, n_customers),
        'credit_cards_count': np.random.poisson(2, n_customers),
        'late_payments_last_year': np.random.poisson(1, n_customers),
        'bankruptcy_history': np.random.choice([0, 1], n_customers, p=[0.95, 0.05]),
        'loan_amount': np.random.lognormal(9, 1, n_customers),
        'loan_purpose': np.random.choice(
            ['购房', '购车', '教育', '装修', '债务整合', '其他'],
            n_customers, p=[0.25, 0.2, 0.15, 0.15, 0.15, 0.1]
        ),
        'loan_term_months': np.random.choice([12, 24, 36, 48, 60], n_customers, p=[0.1, 0.2, 0.3, 0.25, 0.15]),
        'interest_rate': np.random.uniform(3.5, 15.0, n_customers),
    }

    # 创建相关性
    df = pd.DataFrame(data)

    # 收入与信用分数的相关性
    df['credit_score'] = np.clip(
        df['credit_score'] + (df['income'] - df['income'].mean()) / df['income'].std() * 20,
        300, 850
    )

    # 年龄与工作年限的相关性
    df['employment_years'] = np.minimum(df['employment_years'], df['age'] - 18)

    # 收入与贷款金额的相关性
    df['loan_amount'] = df['loan_amount'] * (0.5 + 0.5 * df['income'] / df['income'].mean())

    # 负债收入比与利率的相关性
    df['interest_rate'] = df['interest_rate'] + df['debt_to_income_ratio'] * 5

    # 信用评分与利率的相关性（负相关）
    df['interest_rate'] = df['interest_rate'] - (df['credit_score'] - 650) / 100

    # 计算违约概率（目标变量）
    default_probability = (
        (df['credit_score'] < 600) * 0.4 +
        (df['debt_to_income_ratio'] > 0.4) * 0.3 +
        (df['late_payments_last_year'] > 3) * 0.2 +
        (df['bankruptcy_history'] == 1) * 0.3 +
        (df['employment_years'] < 1) * 0.2 +
        (df['income'] < 30000) * 0.15 +
        np.random.normal(0, 0.1, n_customers)
    )

    # 转换为二分类
    df['loan_default'] = (default_probability > 0.3).astype(int)

    # 风险等级
    risk_conditions = [
        df['credit_score'] >= 750,
        (df['credit_score'] >= 700) & (df['credit_score'] < 750),
        (df['credit_score'] >= 650) & (df['credit_score'] < 700),
        (df['credit_score'] >= 600) & (df['credit_score'] < 650),
        df['credit_score'] < 600
    ]

    risk_labels = ['AAA', 'AA', 'A', 'BBB', 'BB']
    df['risk_rating'] = np.select(risk_conditions, risk_labels, default='BB')

    # 添加一些缺失值
    missing_indices = np.random.choice(df.index, size=int(0.08 * len(df)), replace=False)
    df.loc[missing_indices, 'savings_amount'] = np.nan

    missing_indices = np.random.choice(df.index, size=int(0.05 * len(df)), replace=False)
    df.loc[missing_indices, 'employment_years'] = np.nan

    # 保存数据
    output_dir = Path(__file__).parent / "data"
    output_dir.mkdir(exist_ok=True)

    data_path = output_dir / "financial_data_sample.csv"
    df.to_csv(data_path, index=False, encoding='utf-8-sig')

    print(f"   ✓ 金融数据样本已保存: {data_path}")
    print(f"   ✓ 数据形状: {df.shape}")
    print(f"   ✓ 违约率: {df['loan_default'].mean():.2%}")
    print(f"   ✓ 平均信用分数: {df['credit_score'].mean():.1f}")

    return df, data_path

def run_financial_eda_analysis(data_path):
    """运行金融数据EDA分析"""
    print("\n🔍 运行金融数据探索性分析...")

    try:
        from eda_analyzer import EDAAnalyzer

        analyzer = EDAAnalyzer()

        # 加载数据
        print("   加载数据...")
        data = analyzer.load_data(data_path)

        # 自动化EDA分析
        print("   执行自动化EDA分析...")
        eda_results = analyzer.auto_eda(data)

        # 保存EDA结果
        output_dir = Path(__file__).parent / "results"
        output_dir.mkdir(exist_ok=True)

        eda_path = output_dir / "financial_eda_results.json"
        analyzer.export_results(eda_results, eda_path)

        print(f"   ✓ EDA分析完成，结果已保存: {eda_path}")

        # 显示关键发现
        print("\n📊 关键发现:")
        if 'insights' in eda_results:
            for insight in eda_results['insights'][:3]:
                print(f"   - {insight}")

        return eda_results, data

    except Exception as e:
        print(f"   ❌ EDA分析失败: {str(e)}")
        return None, None

def run_financial_visualization(data):
    """运行金融数据可视化"""
    print("\n📈 生成金融数据可视化图表...")

    try:
        from visualizer import DataVisualizer

        visualizer = DataVisualizer()

        # 自动可视化
        charts = visualizer.auto_visualize(
            data,
            target_col='loan_default',
            save_charts=True,
            output_dir=str(Path(__file__).parent / "results" / "charts")
        )

        print(f"   ✓ 可视化完成，生成了 {charts['charts_generated']} 个图表")

        # 生成金融专项图表
        financial_charts = {}

        # 1. 信用分数分布
        fig = visualizer.plot_distribution(
            data, 'credit_score',
            interactive=True
        )
        financial_charts['credit_score_distribution'] = fig

        # 2. 收入 vs 贷款金额
        fig = visualizer.plot_scatter(
            data, 'income', 'loan_amount',
            color_col='loan_default',
            interactive=True
        )
        financial_charts['income_loan_scatter'] = fig

        # 3. 风险等级分布
        fig = visualizer.plot_categorical(
            data, 'risk_rating',
            interactive=True
        )
        financial_charts['risk_rating_distribution'] = fig

        # 4. 违约率 vs 特征分析
        default_by_purpose = data.groupby('loan_purpose')['loan_default'].mean().sort_values()
        fig = visualizer.plot_categorical(
            data, 'loan_purpose',
            interactive=True
        )
        financial_charts['default_by_purpose'] = fig

        print(f"   ✓ 金融专项图表生成完成: {len(financial_charts)} 个")

        return financial_charts

    except Exception as e:
        print(f"   ❌ 可视化生成失败: {str(e)}")
        return None

def run_financial_modeling(data, data_path):
    """运行金融数据建模"""
    print("\n🤖 运行金融数据建模...")

    try:
        from data_preprocessor import DataPreprocessor
        from modeling_evaluator import ModelingEvaluator

        # 数据预处理
        print("   数据预处理...")
        preprocessor = DataPreprocessor({
            'missing_threshold': 0.2,
            'feature_selection': True,
            'k_features': 15,
            'balance_data': True,  # 平衡违约样本
            'balance_method': 'smote'
        })

        # 预处理数据
        preprocessing_results = preprocessor.auto_preprocess(
            data,
            target_col='loan_default',
            save_report=True
        )

        # 模型训练
        print("   模型训练...")
        modeler = ModelingEvaluator({
            'cv_folds': 5,
            'enable_hyperparameter_tuning': True,
            'n_iter_search': 15,  # 减少搜索次数加快演示
            'scoring_metric': 'roc_auc'  # 使用ROC AUC作为评估指标
        })

        # 自动建模
        model_results = modeler.auto_modeling(
            data,
            target_col='loan_default',
            model_names=['logistic_regression', 'random_forest', 'xgboost', 'lightgbm']
        )

        print(f"   ✓ 模型训练完成，最佳模型: {model_results['best_model']['name']}")

        # 保存模型
        output_dir = Path(__file__).parent / "results" / "models"
        modeler.save_models(str(output_dir))

        return preprocessing_results, model_results

    except Exception as e:
        print(f"   ❌ 建模失败: {str(e)}")
        return None, None

def calculate_credit_risk_score(data, model_results):
    """计算综合信用风险评分"""
    print("\n📊 计算综合信用风险评分...")

    try:
        # 基于模型结果计算风险评分
        if model_results and 'best_model' in model_results:
            best_model_name = model_results['best_model']['name']
            print(f"   使用模型: {best_model_name}")

            # 创建风险评分表
            risk_scores = []

            for _, row in data.iterrows():
                # 基础信用分数
                base_score = row['credit_score']

                # 调整因子
                adjustments = 0

                # 收入调整
                if row['income'] > 100000:
                    adjustments += 20
                elif row['income'] < 30000:
                    adjustments -= 30

                # 负债收入比调整
                if row['debt_to_income_ratio'] > 0.4:
                    adjustments -= 40
                elif row['debt_to_income_ratio'] < 0.2:
                    adjustments += 15

                # 逾期记录调整
                if row['late_payments_last_year'] > 2:
                    adjustments -= 25

                # 破产历史调整
                if row['bankruptcy_history'] == 1:
                    adjustments -= 100

                # 就业稳定性调整
                if row['employment_years'] > 5:
                    adjustments += 10
                elif row['employment_years'] < 1:
                    adjustments -= 20

                final_score = base_score + adjustments
                final_score = np.clip(final_score, 300, 850)

                risk_scores.append(final_score)

            # 添加到数据中
            data = data.copy()
            data['comprehensive_risk_score'] = risk_scores

            # 风险等级
            score_conditions = [
                data['comprehensive_risk_score'] >= 780,
                (data['comprehensive_risk_score'] >= 740) & (data['comprehensive_risk_score'] < 780),
                (data['comprehensive_risk_score'] >= 700) & (data['comprehensive_risk_score'] < 740),
                (data['comprehensive_risk_score'] >= 660) & (data['comprehensive_risk_score'] < 700),
                (data['comprehensive_risk_score'] >= 620) & (data['comprehensive_risk_score'] < 660),
                data['comprehensive_risk_score'] < 620
            ]

            risk_labels = ['AA+', 'AA', 'A', 'BBB', 'BB', 'B']
            data['final_risk_rating'] = np.select(score_conditions, risk_labels, default='B')

            print(f"   ✓ 风险评分计算完成")
            print(f"   ✓ 平均风险评分: {data['comprehensive_risk_score'].mean():.1f}")

            # 保存风险评分结果
            output_dir = Path(__file__).parent / "results"
            data.to_csv(output_dir / "credit_risk_scores.csv", index=False, encoding='utf-8-sig')

            return data

        else:
            print("   ⚠️ 无法计算风险评分，缺少模型结果")
            return data

    except Exception as e:
        print(f"   ❌ 风险评分计算失败: {str(e)}")
        return data

def generate_financial_report(data, eda_results, model_results):
    """生成金融数据分析报告"""
    print("\n📋 生成金融数据分析报告...")

    try:
        from report_generator import ReportGenerator

        # 配置金融特化报告
        generator = ReportGenerator({
            'report_title': '金融信贷风险分析报告',
            'author': '金融风险分析助手',
            'company': '金融机构',
            'include_recommendations': True
        })

        # 生成报告
        output_dir = Path(__file__).parent / "results"
        output_path = output_dir / "financial_analysis_report.html"

        report_path = generator.generate_comprehensive_report(
            data=data,
            eda_results=eda_results,
            model_results=model_results,
            output_path=str(output_path),
            format="html"
        )

        print(f"   ✓ 金融分析报告已生成: {report_path}")

        # 生成快速报告
        quick_report_path = output_dir / "financial_quick_report.html"
        generator.generate_quick_report(
            data=data,
            target_col='loan_default',
            output_path=str(quick_report_path)
        )

        print(f"   ✓ 快速报告已生成: {quick_report_path}")

        return report_path

    except Exception as e:
        print(f"   ❌ 报告生成失败: {str(e)}")
        return None

def main():
    """主函数"""
    print("💰 金融数据分析示例")
    print("=" * 50)

    # 创建输出目录
    output_dir = Path(__file__).parent / "results"
    output_dir.mkdir(exist_ok=True)

    # 1. 创建样本数据
    data, data_path = create_financial_sample_data()

    # 2. EDA分析
    eda_results, processed_data = run_financial_eda_analysis(data_path)

    if processed_data is None:
        processed_data = data

    # 3. 可视化
    charts = run_financial_visualization(processed_data)

    # 4. 建模
    preprocessing_results, model_results = run_financial_modeling(processed_data, data_path)

    # 5. 计算风险评分
    scored_data = calculate_credit_risk_score(processed_data, model_results)

    # 6. 生成报告
    report_path = generate_financial_report(scored_data, eda_results, model_results)

    # 7. 总结
    print("\n🎉 金融数据分析完成！")
    print("\n📁 生成的文件:")

    results_dir = Path(__file__).parent / "results"
    if results_dir.exists():
        for file_path in results_dir.rglob("*"):
            if file_path.is_file():
                relative_path = file_path.relative_to(results_dir)
                print(f"   📄 {relative_path}")

    print(f"\n📊 主要发现:")
    if eda_results and 'data_quality' in eda_results:
        print(f"   - 数据质量分数: {eda_results['data_quality'].get('overall_score', 'N/A')}")

    if model_results and 'best_model' in model_results:
        best_model = model_results['best_model']
        if best_model and 'metrics' in best_model:
            auc = best_model['metrics'].get('auc', 0)
            print(f"   - 最佳模型AUC: {auc:.3f}")

    default_rate = processed_data['loan_default'].mean()
    print(f"   - 整体违约率: {default_rate:.2%}")

    avg_credit_score = processed_data['credit_score'].mean()
    print(f"   - 平均信用分数: {avg_credit_score:.1f}")

    if 'comprehensive_risk_score' in scored_data.columns:
        avg_risk_score = scored_data['comprehensive_risk_score'].mean()
        print(f"   - 综合风险评分: {avg_risk_score:.1f}")

    if report_path:
        print(f"\n📋 详细分析报告: {report_path}")
        print("   请在浏览器中打开HTML文件查看完整的交互式报告。")

if __name__ == "__main__":
    main()