高级特征工程框架：从理论到实践

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

import featuretools as ft
import pandas as pd
import numpy as np
from featuretools.primitives import (
    Count, 
    Mean, 
    Std, 
    Max, 
    Min, 
    Sum, 
    TimeSinceLast,
    TimeSinceFirst,
    NMostCommon
)

class AutomatedFeatureToolsFramework:
    """
    自动化特征工程框架 - 基于 FeatureTools
    """
    
    def __init__(self, target_entity='main'):
        self.target_entity = target_entity
        self.es = None  # entityset
        self.feature_matrix = None
        self.features = None
        
    def setup_entityset(self, data_dict):
        """
        设置实体集结构
        
        Args:
            data_dict: 包含数据框的字典 {entity_name: DataFrame}
        """
        es = ft.EntitySet("automated_framework")
        
        # 添加所有实体
        for entity_name, df in data_dict.items():
            es.add_dataframe(
                dataframe_name=entity_name,
                dataframe=df,
                index=df.columns[0]  # 默认第一列作为索引
            )
        
        # 建立实体关系（根据实际需求调整）
        # 示例关系建立逻辑
        entity_list = list(data_dict.keys())
        for i, entity1 in enumerate(entity_list):
            for j in range(i+1, len(entity_list)):
                entity2 = entity_list[j]
                # 检查是否可以建立关系（通过共享列）
                shared_cols = set(data_dict[entity1].columns) & set(data_dict[entity2].columns)
                
                if shared_cols:
                    # 这里需要根据实际数据结构调整关系定义
                    pass
        
        self.es = es
        return self
        
    def generate_features(self, target_entity, max_depth=2):
        """
        生成特征矩阵和特征定义
        """
        if self.es is None:
            raise ValueError("EntitySet 未初始化，请先调用 setup_entityset")
            
        # 定义要使用的基元
        agg_primitives = [
            'count', 'sum', 'mean', 'std', 'min', 'max',
            'n_most_common'  # 取最常见的前N个值
        ]
        
        trans_primitives = [
            'absolute', 'negate', 'add_numeric', 'subtract_numeric',
            'multiply_numeric', 'divide_numeric', 'modulo_numeric',
            'and', 'or', 'equal', 'not_equal', 'less_than',
            'greater_than', 'less_than_equal_to', 'greater_than_equal_to'
        ]
        
        # 深度特征合成
        self.feature_matrix, self.features = ft.dfs(
            entityset=self.es,
            target_dataframe_name=target_entity,
            agg_primitives=agg_primitives,
            trans_primitives=trans_primitives,
            max_depth=max_depth,
            n_jobs=-1,  # 并行化处理
            verbose=True
        )
        
        return self.feature_matrix, self.features
    
    def handle_relationships(self, relationships):
        """
        手动添加关系
        
        Args:
            relationships: 关系列表 [(parent_variable, child_variable)]
        """
        for parent_var, child_var in relationships:
            self.es.add_relationship(parent_var, child_var)
        return self

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

from tsfresh import extract_features, select_features
from tsfresh.utilities.dataframe_functions import impute
from tsfresh.feature_extraction import ComprehensiveFCParameters
from sklearn.feature_selection import VarianceThreshold

class AutomatedTSFreshFramework:
    """
    自动时间序列特征工程框架 - 基于 TSFresh
    """
    
    def __init__(self):
        self.extract_settings = None
        self.selected_features = None
        
    def set_extraction_settings(self, kind_to_fc_parameters=None):
        """
        设置提取参数
        """
        if kind_to_fc_parameters is None:
            # 使用全面的特征参数
            self.extract_settings = ComprehensiveFCParameters()
        else:
            self.extract_settings = kind_to_fc_parameters
        return self
    
    def extract_time_series_features(self, df, column_id, column_sort, column_kind=None, 
                                   extra_fbursts_features=None, workers=1):
        """
        提取时间序列特征
        
        Args:
            df: 时间序列数据框 (必须包含 id, time, value 列)
            column_id: ID列名
            column_sort: 时间排序列名
            column_kind: 可选，多变量时间序列的类型列名
            extra_fbursts_features: 额外的特征参数
            workers: 并行处理线程数
        """
        # 提取特征
        extracted_features = extract_features(
            df,
            column_id=column_id,
            column_sort=column_sort,
            column_kind=column_kind,
            default_fc_parameters=self.extract_settings,
            n_jobs=workers
        )
        
        # 处理缺失值
        extracted_features = impute(extracted_features)
        
        return extracted_features
    
    def select_relevant_features(self, X, y, test_for_binary_target_cat_correlation=True, 
                               ml_task='classification'):
        """
        特征选择
        
        Args:
            X: 特征矩阵，由tsfresh输出获得
            y: 目标向量
            test_for_binary_target_cat_correlation: 是否测试二分类目标
            ml_task: 机器学习任务类型 ('classification' 或 'regression')
        """
        # 特征选择
        self.selected_features = select_features(
            X,
            y,
            test_for_binary_target_cat_correlation=test_for_binary_target_cat_correlation,
            ml_task=ml_task
        )
        
        return self.selected_features
    
    def extract_advanced_timeseries_features(self, df, ids, columns):
        """
        高级时间序列特征提取
        
        Args:
            df: 包含时间序列数据的数据框
            ids: 要提取特征的时间序列IDs列表
            columns: 要分析的列
            
        Returns:
            包含高级特征的DataFrame
        """
        result_features = {}
        
        for col in columns:
            for i in ids:
                series = df[df.index == i][col].dropna()
                features = self._calculate_advanced_features(series)
                for feat_name, feat_value in features.items():
                    result_features[f'{col}_{i}_{feat_name}'] = feat_value
                    
        return pd.DataFrame([result_features])
    
    def _calculate_advanced_features(self, series):
        """
        计算高级时间序列特征
        """
        features = {}
        
        # 统计特征
        features['mean'] = series.mean()
        features['std'] = series.std()
        features['skewness'] = series.skew()
        features['kurtosis'] = series.kurtosis()
        
        # 极值特征
        features['min'] = series.min()
        features['max'] = series.max()
        features['median'] = series.median()
        
        # 变化特性
        if len(series) > 1:
            diff_series = series.diff().dropna()
            features['diff_mean'] = diff_series.mean()
            features['diff_std'] = diff_series.std()
            features['change_percentage'] = (diff_series.abs() > 0).sum() / len(diff_series)
        
        # 趋势强度
        x = np.arange(len(series))
        if len(set(x)) > 1 and len(set(series.values)) > 1:
            trend_coeffs = np.polyfit(x, series.values, 1)
            features['trend_slope'] = trend_coeffs[0]
            features['trend_strength'] = abs(trend_coeffs[0]) / series.std()
        
        return features

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

try:
    from autofeat import AutoFeatRegressor, AutoFeatClassifier
except ImportError:
    # 提供Mock实现或跳过
    pass

class AutomatedAutoFeatFramework:
    """
    自动化特征工程框架 - 基于 AutoFeat
    """
    
    def __init__(self, task_type='regression', max_opts=50, n_bins=100):
        """
        Args:
            task_type: 任务类型 ('regression' 或 'classification')
            max_opts: 最大函数组合操作数
            n_bins: 计算特征重要性的分箱数
        """
        self.task_type = task_type
        self.max_opts = max_opts
        self.n_bins = n_bins
        self.model = None
        self.feature_names = None
        
    def fit_and_transform(self, X, y):
        """
        拟合AutoFeat模型并转换特征
        """
        if self.task_type == 'regression':
            self.model = AutoFeatRegressor(
                max_opts=self.max_opts,
                n_bins=self.n_bins,
                verbose=1
            )
        else:
            self.model = AutoFeatClassifier(
                max_opts=self.max_opts,
                n_bins=self.n_bins,
                verbose=1
            )
        
        # 在这里我们需要模拟fit_transform的行为
        # 因为AutoFeat通常只提供fit/predict分离的方法
        X_transformed = self.model.fit(X, y).transform(X) if hasattr(self.model, 'transform') else X
        
        return X_transformed, self.model
    
    def select_best_features(self, af_model, X_original, threshold=0.01):
        """
        基于AutoFeat的选择函数，选择最佳特征
        
        Args:
            af_model: 已训练的AutoFeat模型
            X_original: 原始特征
            threshold: 特征重要性阈值
        """
        # 如果存在特征重要性属性，则根据其进行筛选
        if hasattr(af_model, 'featureimps_') and af_model.featureimps_ is not None:
            feature_mask = af_model.featureimps_ > threshold
            n_features_before = X_original.shape[1] 
            X_selected = X_original[:, feature_mask] if isinstance(X_original, np.ndarray) else X_original.loc[:, feature_mask]
            n_features_after = X_selected.shape[1] 
            
            print(f"特征数量从 {n_features_before} 减少到 {n_features_after}")
            return X_selected
        else:
            print("模型不支持特征重要性评估")
            return X_original

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241

import pandas as pd
import numpy as np
from typing import Dict, List, Union
from sklearn.preprocessing import StandardScaler, MinMaxScaler
import warnings
warnings.filterwarnings('ignore')

class TimeSeriesFeatureEngineering:
    """
    时间序列特征工程框架
    """
    
    def __init__(self):
        self.scalers = {}
        self.lag_features = []
        self.rolling_features = []
        
    def create_lag_features(self, df: pd.DataFrame, 
                           value_column: str, 
                           group_column: str = None,
                           lags: List[int] = [1, 2, 3, 5, 7, 14, 30]):
        """
        创建滞后特征
        
        Args:
            df: 输入数据框
            value_column: 值列名
            group_column: 分组列名（如果有的话）
            lags: 滞后期数列表
        """
        lag_df = df.copy()
        
        if group_column:
            # 按组计算滞后（适用于面板数据）
            for lag in lags:
                lag_df[f'{value_column}_lag_{lag}'] = lag_df.groupby(group_column)[value_column].shift(lag)
        else:
            # 全局滞后
            for lag in lags:
                lag_df[f'{value_column}_lag_{lag}'] = lag_df[value_column].shift(lag)
        
        self.lag_features = [f'{value_column}_lag_{lag}' for lag in lags]
        return lag_df
    
    def create_rolling_features(self, df: pd.DataFrame,
                               value_column: str,
                               group_column: str = None,
                               windows: List[int] = [3, 7, 14, 30],
                               functions: List[str] = ['mean', 'std', 'min', 'max', 'sum']):
        """
        创建滚动窗口特征
        
        Args:
            df: 输入数据框
            value_column: 值列名
            group_column: 分组列名
            windows: 窗口大小列表
            functions: 聚合函数列表
        """
        roll_df = df.copy()
        
        if group_column:
            # 按组进行滚动计算
            grouped = df.groupby(group_column)
        else:
            # 整体滚动计算
            grouped = df.groupby(lambda x: '')
        
        for window in windows:
            for func in functions:
                if group_column:
                    roll_feature = grouped[value_column].rolling(window=window).agg(func).reset_index(level=0, drop=True)
                    roll_df[f'{value_column}_roll_{window}_{func}'] = roll_feature
                else:
                    roll_feature = df[value_column].rolling(window=window).agg(func)
                    roll_df[f'{value_column}_roll_{window}_{func}'] = roll_feature
                
                self.rolling_features.append(f'{value_column}_roll_{window}_{func}')
        
        # 为了保持索引一致，将groupby的结果重新索引
        if group_column and not roll_df.index.equals(df.index):
            roll_df = roll_df.reindex(df.index)
            
        return roll_df
    
    def create_expanding_features(self, df: pd.DataFrame,
                                 value_column: str,
                                 group_column: str = None):
        """
        创建扩展窗口特征（从开始到当前时间点）
        """
        expand_df = df.copy()
        
        if group_column:
            expanding_mean = df.groupby(group_column)[value_column].expanding().mean().reset_index(level=0, drop=True)
            expanding_std = df.groupby(group_column)[value_column].expanding().std().reset_index(level=0, drop=True)
            expanding_min = df.groupby(group_column)[value_column].expanding().min().reset_index(level=0, drop=True)
            expanding_max = df.groupby(group_column)[value_column].expanding().max().reset_index(level=0, drop=True)
        else:
            expanding_mean = df[value_column].expanding().mean()
            expanding_std = df[value_column].expanding().std()
            expanding_min = df[value_column].expanding().min()
            expanding_max = df[value_column].expanding().max()
        
        expand_df[f'{value_column}_expand_mean'] = expanding_mean
        expand_df[f'{value_column}_expand_std'] = expanding_std
        expand_df[f'{value_column}_expand_min'] = expanding_min
        expand_df[f'{value_column}_expand_max'] = expanding_max
        
        return expand_df
    
    def create_cyclical_encoding(self, dates: Union[pd.Series, pd.DatetimeIndex], 
                                encode_cos: bool = True):
        """
        创建周期性编码（如月份、日、小时等的sin/cos编码）
        """
        # 将输入转换为日期时间类型
        if isinstance(dates, pd.Series):
            dt = pd.to_datetime(dates)
        else:
            dt = dates
            
        # 获取时间特性进行周期编码
        features = {}
        
        # 天内小时的循环编码
        hours_in_day = 24
        features['hour_sin'] = np.sin(2 * np.pi * dt.hour / hours_in_day)
        features['hour_cos'] = np.cos(2 * np.pi * dt.hour / hours_in_day)
        
        # 每天（月的天数）的循环编码
        days_in_month = 31  # 最大28-31，这里使用最大值
        features['day_sin'] = np.sin(2 * np.pi * dt.day / days_in_month)
        features['day_cos'] = np.cos(2 * np.pi * dt.day / days_in_month)
        
        # 月的循环编码
        months_in_year = 12
        features['month_sin'] = np.sin(2 * np.pi * dt.month / months_in_year)
        features['month_cos'] = np.cos(2 * np.pi * dt.month / months_in_year)
        
        # 年内的天数循环编码
        day_of_year = dt.dt.dayofyear
        days_in_year = 365  # 平年，闰年需特殊处理
        features['dayofyear_sin'] = np.sin(2 * np.pi * day_of_year / days_in_year)
        features['dayofyear_cos'] = np.cos(2 * np.pi * day_of_year / days_in_year)
        
        # 如果需要cosine编码，则也返回它们
        if not encode_cos:
            # 只保留每个维度的一个部分（如正弦）
            for key in ['hour', 'day', 'month', 'dayofyear']:
                del features[key + '_cos']
        
        # 构建DataFrame
        result_df = pd.DataFrame(features)
        return result_df
    
    def create_time_based_features(self, df: pd.DataFrame, datetime_column: str):
        """
        从datetime列创建时间相关的特征
        """
        df = df.copy()
        df[datetime_column] = pd.to_datetime(df[datetime_column])
        
        # 一周的天数
        df['weekday'] = df[datetime_column].dt.weekday
        df['is_weekend'] = (df[datetime_column].dt.weekday >= 5).astype(int)
        
        # 一月的天数和季度
        df['day_of_month'] = df[datetime_column].dt.day
        df['quarter'] = df[datetime_column].dt.quarter
        
        # 周数和工作周标志
        df['week_of_year'] = df[datetime_column].dt.isocalendar().week.astype(int)
        df['is_month_start'] = df[datetime_column].dt.is_month_start.astype(int)
        df['is_month_end'] = df[datetime_column].dt.is_month_end.astype(int)
        
        return df
    
    def create_change_features(self, df: pd.DataFrame, value_column: str):
        """
        创建变化相关特征
        """
        df = df.copy()
        
        # 前后差分变化
        df[f'{value_column}_diff_1'] = df[value_column].diff()
        df[f'{value_column}_pct_change_1'] = df[value_column].pct_change()
        
        # 移动平均的变化率
        df[f'{value_column}_ma_3'] = df[value_column].rolling(window=3).mean()
        df[f'{value_column}_ma_7'] = df[value_column].rolling(window=7).mean()
        
        df[f'{value_column}_vs_ma_7'] = df[value_column] / df[f'{value_column}_ma_7']
        
        return df
    
    def create_volatility_features(self, df: pd.DataFrame, value_column: str, 
                                  periods: List[int] = [5, 10, 20]):
        """
        创建波动率特征
        """
        df = df.copy()
        
        for period in periods:
            # 计算百分比回报
            returns = df[value_column].pct_change()
            # 滚动波动率
            volatility = returns.rolling(window=period).std()
            df[f'{value_column}_volatility_{period}'] = volatility
        
        return df
    
    def scale_features(self, df: pd.DataFrame, columns: List[str], method: str = 'standard'):
        """
        标准化特征
        
        Args:
            df: 数据框
            columns: 需要标准化的列名列表
            method: 缩放方法 ('standard', 'min-max')
        """
        scaler_map = {
            'standard': StandardScaler(),
            'min-max': MinMaxScaler()
        }
        
        if method not in scaler_map:
            raise ValueError(f"不支持的缩放方法: {method}. 支持的方法: {list(scaler_map.keys())}")
        
        df_scaled = df.copy()
        
        for col in columns:
            if col not in self.scalers:
                self.scalers[col] = scaler_map[method]
                scaled_value = self.scalers[col].fit_transform(df[[col]])
            else:
                scaled_value = self.scalers[col].transform(df[[col]])
            
            df_scaled[f'{col}_scaled'] = scaled_value
        
        return df_scaled

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

def demonstrate_cyclical_encoding():
    """
    演示循环编码的工作原理
    """
    import matplotlib.pyplot as plt
    
    # 创建一年的数据
    dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D')
    ts_fe = TimeSeriesFeatureEngineering()
    
    # 创建循环特征
    cyclical_features = ts_fe.create_cyclical_encoding(dates)
    
    # 可视化
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 4))
    
    # 月的编码可视化
    ax1.plot(month_sin := cyclical_features['month_sin'], label='Month Sin')
    ax1.plot(month_cos := cyclical_features['month_cos'], label='Month Cos')
    ax1.set_title('Cyclical Encoding: Month')
    ax1.legend()
    
    # 天的编码可视化
    ax2.plot(cyclical_features['day_sin'], label='Day Sin')
    ax2.plot(cyclical_features['day_cos'], label='Day Cos')
    ax2.set_title('Cyclical Encoding: Day of Month')
    ax2.legend()
    
    plt.tight_layout()
    plt.show()
    
    return cyclical_features

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218

from itertools import combinations, combinations_with_replacement
from sklearn.preprocessing import PolynomialFeatures
from sklearn.base import BaseEstimator, TransformerMixin
import pandas as pd
import numpy as np

class CrossFeatureEngineering:
    """
    交叉特征工程框架
    """
    
    def __init__(self):
        self.interaction_terms = []
        self.candidate_pairs = []
        
    def create_polynomial_interactions(self, X: pd.DataFrame, degree: int = 2, 
                                     interaction_only: bool = True):
        """
        创建多项式交叉特征
        
        Args:
            X: 输入特征矩阵
            degree: 多项式的度数
            interaction_only: 是否仅创建交互项（排除平方项）
        """
        poly = PolynomialFeatures(
            degree=degree, 
            interaction_only=interaction_only, 
            include_bias=False
        )
        
        X_poly = poly.fit_transform(X.select_dtypes(include=[np.number]))
        
        # 获取特征名
        feature_names = [f'interaction_{i}' for i in range(X_poly.shape[1])]
        X_poly_df = pd.DataFrame(X_poly, columns=feature_names[2:])  
        # 排除常数项（第一个），因为PolynomialFeatures总是包含它（即使include_bias=False，也会有原始特征）
        X_poly_df.columns = list(X.columns) + [ f'poly_x{i}' for i in range(degree, X_poly.shape[1])]
        
        return X_poly, feature_names, poly
    
    def create_manual_interactions(self, df: pd.DataFrame, numeric_columns: List[str],
                                  operators: List[str] = ['*', '/', '+', '-']):
        """
        创建手动指定的交叉特征
        
        Args:
            df: 输入数据框
            numeric_columns: 数值列名列表
            operators: 运算符列表
        """
        df_manual = df.copy()
        
        for i, col1 in enumerate(numeric_columns):
            for j, col2 in enumerate(numeric_columns):
                if i >= j:  # 避免重复和自乘
                    continue
                    
                for op in operators:
                    new_col = f'{col1}_{op}_{col2}'
                    
                    if op == '*':
                        df_manual[new_col] = df_manual[col1] * df_manual[col2]
                    elif op == '/':
                        # 防止除以0
                        df_manual[new_col] = df_manual[col1] / (df_manual[col2] + 1e-8)
                    elif op == '+':
                        df_manual[new_col] = df_manual[col1] + df_manual[col2]
                    elif op == '-':
                        df_manual[new_col] = df_manual[col1] - df_manual[col2]
                
                    self.interaction_terms.append(new_col)
        
        return df_manual
    
    def create_ratios(self, df: pd.DataFrame, numerator_cols: List[str], 
                     denominator_cols: List[str], safe_division: bool = True):
        """
        创建比率特征
        
        Args:
            df: 输入数据框
            numerator_cols: 分子列名列表
            denominator_cols: 分母列名列表
            safe_division: 是否安全除法（避免除零）
        """
        df_ratios = df.copy()
        
        for num_col in numerator_cols:
            for den_col in denominator_cols:
                if num_col != den_col:
                    new_col = f'{num_col}_over_{den_col}'
                    
                    if safe_division:
                        df_ratios[new_col] = df_ratios[num_col] / (df_ratios[den_col] + 1e-8)
                    else:
                        df_ratios[new_col] = df_ratios[num_col] / df_ratios[den_col]
                
                    self.interaction_terms.append(new_col)
        
        return df_ratios
    
    def create_threshold_features(self, df: pd.DataFrame, columns: List[str],
                                 thresholds: List[float] = [0.1, 0.25, 0.5, 0.75]):
        """
        基于给定阈值创建二进制特征
        
        Args:
            df: 输入数据框
            columns: 列名列表
            thresholds: 阈值列表
        """
        df_thresh = df.copy()
        
        for col in columns:
            for thr in thresholds:
                new_col = f'{col}_above_{thr}'
                df_thresh[new_col] = (df_thresh[col] > thr).astype(int)
                
                self.interaction_terms.append(new_col)
        
        return df_thresh
    
    def detect_potential_interactions(self, df: pd.DataFrame, target_col: str,
                                     method: str = 'correlation', top_n: int = 20):
        """
        检测潜在的交叉特征对
        
        Args:
            df: 输入数据框
            target_col: 目标列
            method: 检测方法 ('correlation', 'mutual_info')
            top_n: 返回前n个最可能的配对
        """
        if method == 'correlation':
            correlations = df.corr()[target_col].abs().sort_values(ascending=False)
            return correlations.head(top_n)
        else:
            # Mutual Information或其他
            from sklearn.feature_selection import mutual_info_regression
            X = df.drop(columns=[target_col]).select_dtypes(include=[np.number])
            y = df[target_col]
            
            mi_scores = mutual_info_regression(X, y)
            mi_result = pd.Series(mi_scores, index=X.columns).sort_values(ascending=False)
            return mi_result.head(top_n)
    
    def high_order_interactions_selector(self, df: pd.DataFrame, 
                                        interaction_candidates: List[str],
                                        target: pd.Series,
                                        top_k: int = 50,
                                        selection_method: str = 'mi'):
        """
        高阶交叉特征选择器
        
        Args:
            df: 数据框
            interaction_candidates: 候选交叉特征列表
            target: 目标系列
            top_k: 选择的特征数量
            selection_method: 选择方法 ('mi', 'anova', 'rf')
        """
        X = df[interaction_candidates]
        X_cleaned = X.fillna(X.mean())  # 清理缺失值
        
        from sklearn.feature_selection import mutual_info_regression, f_regression
        from sklearn.ensemble import RandomForestRegressor

        if selection_method == 'mi':
            scores = mutual_info_regression(X_cleaned, target)
        elif selection_method == 'anova':
            scores, _ = f_regression(X_cleaned, target)
        elif selection_method == 'rf':
            rf = RandomForestRegressor(n_estimators=100, random_state=42)
            rf.fit(X_cleaned, target)
            scores = rf.feature_importances_
        else:
            raise ValueError(f"不支持的选择方法: {selection_method}")
        
        # 获取得分最高的k个特征
        sorted_idx = np.argsort(scores)[::-1]
        top_idx = sorted_idx[:top_k]
        selected_features = [interaction_candidates[i] for i in top_idx]
        
        return selected_features, scores[top_idx]
    
    def create_logic_interactions(self, df: pd.DataFrame, 
                                bool_features: List[str],
                                operators: List[str] = ['&', '|', '^']):
        """
        创建布尔逻辑操作交互
        
        Args:
            df: 输入数据框
            bool_features: 布尔特征列表
            operators: 逻辑运算符 (&, |, ^)
        """
        df_logic = df.copy()
        
        for i, col1 in enumerate(bool_features):
            for j, col2 in enumerate(bool_features):
                # 跳过自身比较
                if i >= j:
                    continue
                for op in operators:
                    new_col = f'{col1}_{op}_{col2}'
                    if op == '&':
                        df_logic[new_col] = df_logic[col1] & df_logic[col2]
                    elif op == '|':
                        df_logic[new_col] = df_logic[col1] | df_logic[col2]
                    elif op == '^':
                        df_logic[new_col] = df_logic[col1] ^ df_logic[col2]
                    
                    df_logic[new_col] = df_logic[new_col].astype(int)
                    
                    self.interaction_terms.append(new_col)
        
        return df_logic

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

class InteractionValidator:
    """
    交叉特征质量与验证工具
    """
    
    def __init__(self):
        pass
    
    def detect_duplicate_features(self, df: pd.DataFrame, tol: float = 1e-8):
        """
        检测重复特征
        """
        duplicates = []
        cols = df.select_dtypes(include=[np.number]).columns
        
        for i in range(len(cols)):
            for j in range(i+1, len(cols)):
                if abs(df[cols[i]] - df[cols[j]]).mean() < tol:
                    duplicates.append((cols[i], cols[j]))
                    
        return duplicates
    
    def detect_multicollinearity(self, df: pd.DataFrame, threshold: float = 0.9):
        """
        检测多重共线性
        """
        corr_matrix = df.select_dtypes(include=[np.number]).corr().abs()
        high_corr_pairs = []
        
        for i in range(len(corr_matrix.columns)):
            for j in range(i+1, len(corr_matrix.columns)):
                if corr_matrix.iloc[i, j] > threshold:
                    high_corr_pairs.append((
                        corr_matrix.columns[i], 
                        corr_matrix.columns[j], 
                        corr_matrix.iloc[i, j]
                    ))
                    
        return high_corr_pairs
    
    def evaluate_feature_quality(self, X: pd.DataFrame, y: pd.Series, 
                               selected_features: List[str] = None):
        """
        评估特征质量
        """
        if selected_features:
            X_use = X[selected_features].copy()
        else:
            X_use = X.copy()
        
        from sklearn.feature_selection import mutual_info_regression
        from sklearn.metrics import mean_squared_error
        
        # 特征统计信息
        results = {
            'std_zero': [],
            'corr_with_target_high': [],
            'unique_values_low': []  # 唯一值很少的特征
        }
        
        for col in X_use.columns:
            if X_use[col].std() < 1e-8:
                results['std_zero'].append(col)
            if len(X_use[col].unique()) <= 2:
                results['unique_values_low'].append(col)
        
        # 目标相关性
        mi_scores = mutual_info_regression(X_use, y)
        target_corr_pairs = X_use.corrwith(y).abs()
        results['corr_with_target_high'] = target_corr_pairs[target_corr_pairs > 0.9].index.tolist()
        
        return results

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108

from sklearn.metrics import accuracy_score, roc_auc_score, mean_squared_error
from sklearn.model_selection import cross_val_score
import numpy as np
from tqdm import tqdm
import warnings
warnings.filterwarnings('ignore')

class PermutationImportanceSelector:
    """
    基于置换重要性的特征选择器
    """
    
    def __init__(self, model, X_test, y_test, scoring_func=None, n_repeats=10):
        """
        Args:
            model: 训练好的模型
            X_test: 测试特征矩阵
            y_test: 测试标签
            scoring_func: 评分函数，如果为None则使用默认的准确率
            n_repeats: 重复次数（用于获取重要性的置信区间）
        """
        self.model = model
        self.X_test = X_test.copy()
        self.y_test = y_test
        self.scoring_func = scoring_func
        self.n_repeats = n_repeats
        self.baseline_score = None
        self.importances = None
        
    def calculate_permutation_importance(self):
        """
        计算置换重要性
        """
        if self.scoring_func is None:
            # 根据目标变量判断是分类还是回归
            if len(np.unique(self.y_test)) <= 20 and self.y_test.dtype in [int, 'int64', 'int32']:  # 假设整数标签是分类问题
                self.scoring_func = accuracy_score if len(np.unique(self.y_test)) <= 2 else \
                                    (lambda y_true, y_pred: roc_auc_score(y_true, y_pred, multi_class='ovr'))
            else:  # 回归问题
                self.scoring_func = mean_squared_error
        
        # 获取基准分数（原性能）
        y_pred = self.model.predict(self.X_test)
        
        if callable(self.scoring_func):
            self.baseline_score = self.scoring_func(self.y_test, y_pred)
        else:
            self.baseline_score = self.model.score(self.X_test, self.y_test) 
        
        importances = []
        
        for col in self.X_test.columns:
            col_scores = []
            
            for _ in range(self.n_repeats):
                # 复制测试数据
                X_permuted = self.X_test.copy()
                
                # 打乱该列
                X_permuted[col] = np.random.permutation(X_permuted[col])
                
                # 获取预测并评估
                y_pred_permuted = self.model.predict(X_permuted)
                
                if callable(self.scoring_func):
                    permuted_score = self.scoring_func(self.y_test, y_pred_permuted)
                else:
                    permuted_score = self.model.score(X_permuted, self.y_test)
                    
                # 重要性 = (baseline - permuted_score)，越大表明该特征越重要
                col_scores.append(self.baseline_score - permuted_score)
            
            importances.append({
                'feature': col,
                'importance_mean': np.mean(col_scores),
                'importance_std': np.std(col_scores),
                'importance_scores': col_scores
            })
        
        self.importances = pd.DataFrame(importances).sort_values('importance_mean', axis=0, 
                                                               ascending=False).reset_index(drop=True)
            
        return self.importances
    
    def select_features_by_importance(self, threshold=None, n_features=None):
        """
        根据重要性选择特征
        
        Args:
            threshold: 重要性阈值
            n_features: 选择的特征数量
        """
        if self.importances is None:
            self.calculate_permutation_importance()
            
        if threshold is not None:
            selected_features = self.importances[
                self.importances['importance_mean'] >= threshold
            ]['feature'].tolist()
        elif n_features is not None:
            selected_features = self.importances.head(n_features)['feature'].tolist()
        else:
            # 默认情况下选择重要性大于0的特征
            selected_features = self.importances[
                self.importances['importance_mean'] > 0
            ]['feature'].tolist()
            
        return selected_features

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

import shap
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

class SHAPFeatureSelector:
    """
    基于SHAP值的特征选择器
    """
    
    def __init__(self, model, X_train, feature_names=None):
        """
        Args:
            model: 训练好的模型
            X_train: 训练数据（用于构建背景数据）
            feature_names: 特征名称
        """
        self.model = model
        self.X_train = X_train
        if feature_names is None:
            if hasattr(X_train, 'columns'):
                self.feature_names = X_train.columns.tolist()
            else:
                self.feature_names = [f'feature_{i}' for i in range(X_train.shape[1])]
        else:
            self.feature_names = feature_names
        self.explainer = None
        self.shap_values = None
        self.feature_importances = None
        self.summary_plot_ready = False
    
    def calculate_shap_values(self, X_explain=None, method='auto'):
        """
        计算SHAP值
        
        Args:
            X_explain: 需要解释的数据，如果不提供则使用部分训练数据
            method: 解释方法 ('auto', 'permutation', 'tree')
        """
        if X_explain is None:
            # 如果没有提供解释数据，默认使用训练数据的一小部分
            if len(self.X_train) > 100:
                X_explain = self.X_train.sample(100)
            else:
                X_explain = self.X_train
        
        # 根据模型类型选择合适的explainer
        try:
            if method == 'tree' or hasattr(self.model, 'tree_') or 'lightgbm' in str(type(self.model)):
                self.explainer = shap.TreeExplainer(self.model)
            elif method == 'permutation':
                self.explainer = shap.PermutationExplainer(self.model.predict, X_explain)
            else:  # auto/default
                self.explainer = shap.Explainer(self.model.predict, X_explain)
            
            self.shap_values = self.explainer(X_explain)
            
            # 计算特征重要性（平均SHAP绝对值）
            self.feature_importances = np.mean(np.abs(self.shap_values.values), axis=0)
            
            feature_importance_df = pd.DataFrame({
                'feature': self.feature_names,
                'shap_importance': self.feature_importances
            }).sort_values('shap_importance', ascending=False).reset_index(drop=True)
            
            self.summary_plot_ready = True
            return feature_importance_df, self.shap_values
            
        except Exception as e:
            print(f"SHAP计算出错: {e}")
            print("可能是模型不兼容或缺少shap包，请检查安装和模型类型")
            return None
    
    def select_features_by_shap(self, threshold=None, n_features=None):
        """
        根据SHAP值选择特征
        
        Args:
            threshold: SHAP阈值
            n_features: 选择的特征数量
        """
        if self.feature_importances is None:
            feature_importance_df, _ = self.calculate_shap_values()
            
            if feature_importance_df is None:
                return []  # 如果SHAP失败，返回空列表
        else:
            feature_importance_df = pd.DataFrame({
                'feature': self.feature_names,
                'shap_importance': self.feature_importances
            }).sort_values('shap_importance', ascending=False).reset_index(drop=True)
            
        if threshold is not None:
            selected_features = feature_importance_df[
                feature_importance_df['shap_importance'] >= threshold
            ]['feature'].tolist()
        elif n_features is not None:
            selected_features = feature_importance_df.head(n_features)['feature'].tolist()
        else:
            # 默认情况：选择前20个特征或大于最小值的特征
            selected_features = feature_importance_df.head(20)['feature'].tolist()
            
        return selected_features
    
    def powershap_selection(self, X_train, y_train, alpha=0.01, power=2):
        """
        PowerShap方法 - 结合SHAP和统计检验的混合特征选择方法
        """
        try:
            from powershap import PowerShap
            
            selector = PowerShap(
                model=self.model,
                alpha=alpha,
                power=power,
                automatic=True
            )
            
            X_selected = selector.fit_transform(X_train, y_train)
            selected_features = X_train.columns[selector.support_].tolist()
            
            return selected_features, selector
        except ImportError:
            print("PowerShap未安装，跳过此功能")
            return [], None