Advanced Feature Engineering Framework: From Theory to Practice

Abstract

This article presents a comprehensive methodology for advanced feature engineering, covering automation, time series techniques, feature interactions, feature selection, and domain-specific features for finance and risk control. Practical code frameworks help data scientists and machine learning engineers apply these techniques in real projects.

Keywords

Feature engineering, automated feature extraction, time series, feature interactions, feature selection, financial risk control

1. Introduction and Theoretical Foundations

Feature engineering is central to machine learning. Although advances such as deep learning reduce the need for feature engineering in some domains, high-quality features remain crucial for tabular and structured data.

Traditional feature engineering relies on domain knowledge and experience. Modern approaches combine automation, time series analysis, complex interaction discovery, and optimized feature selection.

1.1 Key Challenges in Feature Engineering

High-dimensional sparsity: Identify useful features among a large number of candidates.
Temporal dependence: Account for evolving patterns in time series.
Interaction effects: Discover and model higher-order feature interactions.
Domain adaptation: Tailor feature construction to different business contexts.

1.2 Method Categories

We group feature engineering methods into the following categories:

Automated feature engineering
Time series feature processing
Feature interaction generation and selection
Feature selection techniques
Domain-specific feature engineering

2. Automated Feature Engineering Frameworks

2.1 A Practical FeatureTools Framework

FeatureTools is an open-source library for automated feature engineering. It uses Deep Feature Synthesis to combine transformations and aggregations into new feature sets.

  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

2.2 A Practical TSFresh Framework

Tsfresh is an automated tool for time series feature extraction, offering more than 700 feature calculation methods.

  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

2.3 A Practical AutoFeat Framework

AutoFeat combines symbolic regression and linear models for automated feature engineering.

 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

3. Time Series Feature Engineering

3.1 Lag Features and Rolling Statistics

  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

3.2 Cyclical Encoding Explained

 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

4. Feature Interaction Strategies

4.1 Generating Lower- and Higher-Order Feature Interactions

  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

4.2 Pitfall Detection and Quality Validation

 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

5. Feature Selection Frameworks

5.1 Permutation Importance

  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

5.2 SHAP-Based Importance

  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

5.3 Feature Selection with Genetic Algorithms

  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
from geneticalgorithm import geneticalgorithm as ga
import numpy as np
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.base import BaseEstimator, TransformerMixin

class GeneticFeatureSelector:
    """
    基于遗传算法的特征选择器
    """
    
    def __init__(self, estimator, X, y, cv=5, scoring='accuracy', 
                 n_population=50, n_generations=100, 
                 crossover_probability=0.5, mutation_probability=0.2):
        """
        Args:
            estimator: 指标估计器
            X: 输入特征
            y: 目标变量
            cv: 交叉验证折数
            scoring: 评分指标
            n_population: 种群大小
            n_generations: 代数
            crossover_probability: 交叉概率
            mutation_probability: 变异概率
        """
        self.estimator = estimator
        self.X = X
        self.y = y
        self.cv = cv
        self.scoring = scoring
        self.n_population = n_population
        self.n_generations = n_generations
        self.crossover_probability = crossover_probability
        self.mutation_probability = mutation_probability
        self.selected_features = []
        self.feature_support = None
        self.best_fitness = None
        
    def fitness_function(self, solution):
        """
        适应度函数 - 使用交叉验证分数作为适应度
        """
        # 将二进制解转换为特征掩码
        selected_indices = np.where(solution == 1)[0]
        
        if len(selected_indices) == 0:
            # 如果没有选择任何特征，返回最小值
            return 0.001
        
        X_subset = self.X[:, selected_indices].copy() if isinstance(self.X, np.ndarray) else self.X.iloc[:, selected_indices]
        
        try:
            cv_scores = cross_val_score(
                self.estimator, 
                X_subset, 
                self.y,
                cv=self.cv, 
                scoring=self.scoring
            )
            fitness = cv_scores.mean()
        except Exception as e:
            print(f"计算交叉验证分数错误: {e}")
            fitness = 0.001  # 错误情况下给予惩罚分数
        
        return fitness
    
    def optimize(self):
        """
        运行遗传算法优化
        """
        dim = self.X.shape[1]
        
        # 遗传算法参数
        algorithm_param = {
            'max_num_iteration': self.n_generations,
            'population_size': self.n_population,
            'mutation_probability': self.mutation_probability,
            'elit_ratio': 0.05,
            'crossover_probability': self.crossover_probability,
            'parents_portion': 0.2,
            'crossover_type': 'single_point',
            'max_iteration_without_improv': 20
        }
        
        model = ga(
            function=self.fitness_function,
            dimension=dim,
            variable_type='bool',  # 二进制变量
            algorithm_parameters=algorithm_param
        )
        
        model.run()
        
        self.feature_support = model.output_dict['variable'].astype(bool)
        self.best_fitness = model.output_dict['function']
        self.selected_features = [i for i, is_selected in enumerate(self.feature_support) if is_selected]
        
        print(f"最优适应度: {self.best_fitness}")
        print(f"选中特征数量: {len(self.selected_features)} / {dim}")
        
        return self.selected_features, self.feature_support, self.best_fitness
    
    def transform(self, X):
        """
        应用特征选择变换到新数据
        """
        if self.feature_support is None:
            raise ValueError("未运行优化，请先调用optimize函数")
        
        selected_indices = np.where(self.feature_support)[0]
        return X[:, selected_indices] if isinstance(X, np.ndarray) else X.iloc[:, selected_indices]


class FeatureSelectorHybrid:
    """
    混合特征选择方法 - 结合多种技术和验证方法
    """
    def __init__(self, primary_method='permutation', secondary_methods=None):
        """
        Args:
            primary_method: 主特征选择方法
            secondary_methods: 次要验证方法列表
        """
        self.primary_method = primary_method
        self.secondary_methods = secondary_methods or ['variance_threshold', 'correlation']
        self.selected_by_primary = []
        self.selected_final = []
        
    def fit_selection(self, X, y, model=None):
        """
        同步执行多重选择方法
        """
        method_outputs = {}
        
        # 1. 方差阈值
        from sklearn.feature_selection import VarianceThreshold
        var_selector = VarianceThreshold(threshold=0.01)
        X_var_filtered = var_selector.fit_transform(X)
        method_outputs['variance_threshold'] = var_selector.get_support(indices=True)
        
        # 2. 相关性过滤
        from sklearn.feature_selection import SelectKBest, f_classif
        if len(np.unique(y)) < 20:  # 分类问题
            selector = SelectKBest(score_func=f_classif, k=min(50, X.shape[1]))
        else:  # 回归问题
            from sklearn.feature_selection import f_regression
            selector = SelectKBest(score_func=f_regression, k=min(50, X.shape[1]))
            
        X_corr_filtered = selector.fit_transform(X, y)
        method_outputs['correlation'] = selector.get_support(indices=True)
        
        # 3. 主选择方法
        if self.primary_method == 'permutation' and model is not None:
            # 分割数据用于排列测试
            from sklearn.model_selection import train_test_split
            X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
            
            model.fit(X_train, y_train)  # 重新训练模型
            perm_selector = PermutationImportanceSelector(model, X_val, y_val)
            importances = perm_selector.calculate_permutation_importance()
            
            # 选择重要性较高的特征
            threshold_val = importances['importance_mean'].quantile(0.25)
            selected_indices = importances[importances['importance_mean'] >= threshold_val].index
            method_outputs['permutation'] = selected_indices.values
        
        # 4. 使用交集作为最终选择
        final_indices = set(method_outputs['variance_threshold']).intersection(
            set(method_outputs['correlation'])
        )
        
        # 进一步与主方法取交集（如果有）
        if 'permutation' in method_outputs:
            final_indices = final_indices.intersection(set(method_outputs['permutation']))
        
        self.selected_final = list(final_indices)
        self.selected_by_primary = method_outputs.get(self.primary_method, [])
        
        return self.selected_final if self.selected_final else self.selected_by_primary
    
    def get_consensus_features(self, feature_lists, consensus_ratio=0.6):
        """
        获取多个特征列表的共识特征
        
        Args:
            feature_lists: 特征列表的列表
            consensus_ratio: 达成共识的比例
        """
        if not feature_lists:
            return []
        
        from collections import Counter
        all_features = [item for sublist in feature_lists for item in sublist]
        feature_counts = Counter(all_features)
        total_methods = len(feature_lists)
        
        consensus_features = [
            feat for feat, count in feature_counts.items()
            if count / total_methods >= consensus_ratio
        ]
        
        return consensus_features

6. Domain-Specific Features for Financial Risk Control

6.1 Risk Ratios and Indicators

  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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
import pandas as pd
import numpy as np
from typing import List, Dict, Union
from datetime import datetime, timedelta

class FintechRiskFeatures:
    "金融风控领域特定特征工程类"
    
    def __init__(self):
        self.user_profiles = {}
        self.applications = {}
        self.transactions = {}
        
    def create_risk_ratios(self, df: pd.DataFrame, 
                          balance_cols: List[str], 
                          limit_cols: List[str],
                          income_col: str = None):
        """
        创建风险比率特征
        
        Args:
            df: 包含账户信息的数据
            balance_cols: 余额列名列表
            limit_cols: 授信额度列名列表
            income_col: 收入列名（可选）
        """
        risk_df = df.copy()
        
        # 信用利用率 - 最重要的风控指标之一
        for limit_col in limit_cols:
            for bal_col in balance_cols:
                usage_ratio = f"{bal_col}_utilization_over_{limit_col}"
                risk_df[usage_ratio] = risk_df[bal_col] / (risk_df[limit_col] + 1e-8)
        
        # 总负债/总收入比（如果收入数据可用）
        if income_col and income_col in df.columns:
            total_debt = sum([df[col].fillna(0) for col in balance_cols if col != income_col])
            risk_df['total_debt_over_income'] = total_debt / (risk_df[income_col] + 1e-8)
        
        # 多头借贷比例
        credit_types = [col for col in df.columns if 'credit' in col.lower()]
        total_active_credit_lines = df[credit_types].apply(
            lambda row: (row > 0).sum(), axis=1
        )
        risk_df['multiple_credit_lines_ratio'] = total_active_credit_lines / len(credit_types)
        
        return risk_df
    
    def create_payment_behavior_indicators(self, payment_history_df: pd.DataFrame):
        """
        从还款记录创建行为指标
        
        Args:
            payment_history_df: 包含还款记录的数据框
                需要至少包含：user_id, pay_date, due_date, actual_payment_amount, scheduled_payment_amount
        """
        pay_behavior_df = payment_history_df.copy()
        
        # 计算延期天数
        pd.options.mode.chained_assignment = None  # disable warning
        pay_behavior_df['days_past_due'] = (
            pd.to_datetime(pay_behavior_df['actual_payment_date']) - 
            pd.to_datetime(pay_behavior_df['due_date'])
        ).dt.days.apply(lambda x: max(0, x))  # 确保非负
        
        # 付款充足度
        pay_behavior_df['pay_sufficiency'] = (
            pay_behavior_df['actual_payment_amount'] / 
            (pay_behavior_df['scheduled_payment_amount'] + 1e-8)
        )
        
        # 违约标识
        pay_behavior_df['is_delinquency'] = np.where(
            pay_behavior_df['days_past_due'] > 0, 1, 0
        )
        
        # 对数据按客户分组并创建聚合指标
        user_summary = pay_behavior_df.groupby('user_id').agg({
            'days_past_due': ['mean', 'max', 'std', 'count'],
            'pay_sufficiency': ['min', 'max', 'mean'],
            'is_delinquency': ['sum', 'mean']  # 历史违约次数/违约频率
        }).fillna(0)  # 填充NaN，对于从未违约的用户
        
        # 扁平化列名
        user_summary.columns = ['_'.join(col).strip() for col in user_summary.columns.values]
        user_summary = user_summary.add_prefix('behavior_')
        
        return user_summary.reset_index()
    
    def create_transaction_patterns(self, trans_df: pd.DataFrame):
        """
        从交易数据创建行为模式
        
        Args:
            trans_df: 包含交易数据的数据框
                需要包含：user_id, trans_date, transaction_amount, transaction_type
        """
        trans_patterns_df = trans_df.copy()
        
        # 转换日期列
        trans_patterns_df['trans_date'] = pd.to_datetime(trans_patterns_df['trans_date'])
        
        # 按时间计算交易频率
        trans_patterns_df['time_since_last_trans'] = trans_patterns_df.groupby('user_id')[
            'trans_date'
        ].diff().dt.days.fillna(0)
        
        # 交易金额统计
        user_trans_stats = trans_patterns_df.groupby('user_id').agg({
            'transaction_amount': [
                'mean', 'std', 'min', 'max', 'sum'
            ],
            'trans_date': [
                'count', lambda x: x.nunique()  # 交易总次数, 不同日期数
            ],
            # 转换为数值特征的交易类型计数
            'transaction_type': lambda x: x.value_counts().to_dict()  # 类型分布
        }).fillna(0)
        
        # 扁平化列名
        user_trans_stats.columns = [
            '_'.join(col).strip().rstrip('_') if col[1] != '<lambda>' else f'{col[0]}_unqiued_days'
            for col in user_trans_stats.columns.values
        ]
        user_trans_stats.rename(columns={'trans_date_<lambda>': 'unique_transaction_days'}, 
                               inplace=True)
        
        # 添加一些复合特征
        # 平均每日交易额
        user_trans_stats['avg_daily_spend'] = user_trans_stats['transaction_amount_sum'] / (
            user_trans_stats['unique_transaction_days'] + 1e-8
        )
        # 交易额变异系数（衡量支出稳定性）
        user_trans_stats['transaction_amount_cv'] = user_trans_stats['transaction_amount_std'] / (
            user_trans_stats['transaction_amount_mean'] + 1e-8
        )
        
        return user_trans_stats.reset_index()
    
    def create_time_window_features(self, df: pd.DataFrame, 
                                   date_col: str, 
                                   value_col: str,
                                   windows: List[int] = [7, 14, 30]):
        """
        创建时间窗内滚动聚合特征
        
        Args:
            df: 时间序列格式的交易或事件数据
            date_col: 日期列
            value_col: 数值列
            windows: 周期天数组
        """
        df = df.copy()
        df[date_col] = pd.to_datetime(df[date_col])
        df = df.sort_values([date_col])
        
        for window in windows:
            df[f'{value_col}_last_{window}d_sum'] = df[value_col].rolling(
                window=f'{window}D', 
                min_periods=1
            ).sum()
            df[f'{value_col}_last_{window}d_mean'] = df[value_col].rolling(
                window=f'{window}D', 
                min_periods=1
            ).mean()
            df[f'{value_col}_last_{window}d_max'] = df[value_col].rolling(
                window=f'{window}D', 
                min_periods=1
            ).max()
            df[f'{value_col}_last_{window}d_count'] = df[value_col].rolling(
                window=f'{window}D', 
                min_periods=1
            ).count()
        
        return df.fillna(0)
    
    def create_aggregated_bureau_features(self, bureau_df: pd.DataFrame):
        """
        创建征信数据聚类特征
        
        Args:
            bureau_df: 征信记录数据框
                应包含：person_id, loan_type, amount, status, start_date, close_date
        """
        df = bureau_df.copy()
        df['start_date'] = pd.to_datetime(df['start_date'])
        df['close_date'] = pd.to_datetime(df['close_date'])
        
        # 计算信用历史长度
        df['credit_history_length_days'] = (
            df['close_date'] - df['start_date']
        ).dt.days.fillna(-1)  # 未结清贷款
        
        # 分组聚合
        bureau_agg = df.groupby('person_id').agg({
            # 贷款数量
            'amount': ['count', 'sum', 'mean', 'max', 'std'],
            # 当前活跃贷款数
            'status': lambda x: (x == 'Active').sum(),
            # 信用历史长度统计
            'credit_history_length_days':  ['mean', 'sum', 'max'],
            # 不同类型贷款的数量
            'loan_type': lambda x: x.nunique()
        }).fillna(0)
        
        # 扁平化列名
        bureau_agg.columns = [
            'bureau_' + '_'.join(col).strip().rstrip('<lambda>') 
            for col in bureau_agg.columns.values
        ]
        # 手动修正列名
        bureau_agg.rename(columns={
            'bureau_status_<lambda>': 'bureau_active_loans',
            'bureau_loan_type_<lambda>': 'bureau_unique_loan_types'
        }, inplace=True)
        
        return bureau_agg.reset_index()
    
    def create_scorecard_features(self, raw_data: pd.DataFrame):
        """
        创建评分卡专用特征
        
        Args:
            raw_data: 原始申请者数据
                包括基本信息和申请信息
        """
        features = raw_data.copy()
        
        # 职业风险等级 - 基于职业种类
        occupation_mapping = {
            'Manager': 1, 'Director': 1, 'Senior Manager': 1,           # 低风险
            'Engineer': 2, 'Doctor': 2, 'Lawyer': 2, 'Scientist': 2,   # 中低风险
            'Officer': 3, 'Analyst': 3, 'Teacher': 3, 'Nurse': 3,     # 中等风险
            'Worker': 4, 'Clerk': 4, 'Driver': 4,                     # 中高风险
            'Student': 5, 'Unemployed': 5                             # 高风险
        }
        features['occupation_risk_level'] = features.get('job_title', pd.Series([5]*len(features))).map(
            occupation_mapping
        ).fillna(5).astype(int)
        
        # 意图特征 - 申请行为分析
        features['intent_indicator'] = features['requested_loan_amount'] / (
            features.get('income', pd.Series([1e5]*len(features))) + 1e-8
        )
        
        # 社会经济地位指标
        features['socioeconomic_index'] = (
            (features.get('income', pd.Series([1e5]*len(features))) / 10000) +
            ((features.get('age', pd.Series([30]*len(features))) - 18) / 50) +
            (features.get('education_level', pd.Series([2]*len(features))) - 1) * 0.5
        ) / 3
        
        # 时间趋势特征
        today = pd.Timestamp.now()
        features['years_with_current_employer'] = (
            today.year - pd.to_datetime(features.get('employment_start_date', 
                                          pd.Series(today)]  # 默认使用当前日期
        if 'employment_start_date' in features.columns:
            features['years_with_current_employer'] = (
                pd.to_datetime(features['employment_start_date']) - today
            ).dt.days / 365.25
        
        # 创建离散化特征 - 对评分卡建模有用
        for col in ['age', 'income', 'requested_loan_amount']:
            if col in features.columns:
                bin_labels = [f'{col}_bin_{i}' for i in range(5)]
                features[f'{col}_category'], bins = pd.qcut(
                    features[col], q=5, 
                    labels=bin_labels, 
                    duplicates='drop'
                ).factorize() # 将类别转换为数字

6.2 Behavioral Sequence 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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
    def create_behavioral_sequence_features(self, user_activities_df: pd.DataFrame):
        """
        从用户活动数据创建行为序列特征
        
        Args:
            user_activities_df: 包含用户活动数据的数据框
                应包含：user_id, activity_type, activity_date, activity_details
        """
        activities = user_activities_df.copy()
        activities['activity_date'] = pd.to_datetime(activities['activity_date'])
        activities = activities.sort_values(['user_id', 'activity_date'])
        
        # 计算活动频率
        user_activity_freq = activities.groupby(['user_id', 'activity_type']).size().reset_index(name='freq')
        user_activity_pivot = user_activity_freq.pivot(
            index='user_id', 
            columns='activity_type', 
            values='freq'
        ).fillna(0).add_prefix('act_')
        
        # 活跃度指标
        time_diffs = activities.groupby('user_id')['activity_date'].diff().dt.days.fillna(0)
        activities.loc[:, 'days_since_last_activity'] = time_diffs
        
        # 计算用户持续活跃天数统计
        user_engagement_metrics = activities.groupby('user_id').agg({
            'days_since_last_activity': ['mean', 'min', 'max', 'std'],
            'activity_date': ['nunique', 'count'],  # 独一日期和总活动数
        }).fillna(0)
        
        # 扁平化列名
        user_engagement_metrics.columns = [
            'engagement_' + '_'.join(col).strip() 
            for col in user_engagement_metrics.columns.values
        ]
        
        # 合并所有行为特征
        result = pd.concat([
            user_engagement_metrics,
            user_activity_pivot
        ], axis=1).fillna(0)
        
        return result.reset_index()
    
    def create_risk_segmentation_features(self, df: pd.DataFrame):
        """
        创建风险细分特征
        
        Args:
            df: 包含客户信息的数据框
        """
        risk_features = df.copy()
        
        # 基于信用评分的风险分级
        if 'credit_score' in df.columns:
            risk_features['credit_grade'] = pd.cut(
                df['credit_score'], 
                bins=[0, 550, 650, 750, 850, np.inf],
                labels=['Very Poor', 'Poor', 'Fair', 'Good', 'Excellent']
            ).cat.codes
            
        # 收入等级划分
        if 'income' in df.columns:
            risk_features['income_bracket'] = pd.qcut(
                df['income'], 
                q=5, 
                labels=['Bottom 20%', '20-40%', 'Mid 20%', '40-80%', 'Top 20%']
            ).cat.codes
        
        # 综合风险评分
        if 'credit_score' in df.columns and 'income' in df.columns:
            risk_features['composite_risk_score'] = (
                0.4 * (df['credit_score'] / df['credit_score'].max()) +
                0.3 * (df.get('income', 0) / df.get('income', 0).max()) + 
                0.3 * (df.get('age', 0) / df.get('age', 0).max()).fillna(0)
            )
        
        return risk_features


# 7. 高级特征工程流水线

class AdvancedFeatureEngineeringPipeline:
    """
    高级特征工程流水线整合上述所有方法
    """
    
    def __init__(self):
        self.feature_processors = {
            'tsfresh': AutomatedTSFreshFramework(),
            'featuretools': AutomatedFeatureToolsFramework(),
            'autofeat': AutomatedAutoFeatFramework(),
            'time_series': TimeSeriesFeatureEngineering(),
            'cross_features': CrossFeatureEngineering(),
            'selector': FeatureSelectorHybrid(),
            'fintech_risk': FintechRiskFeatures()
        }
        
    def run_complete_pipeline(self, 
                            raw_data: Dict[str, pd.DataFrame],
                            target_col: str = None,
                            task_type: str = 'classification',
                            enable_ts_features: bool = True,
                            enable_cross_features: bool = True,
                            enable_selection: bool = True):
        """
        运行完整特征工程流水线
        
        Args:
            raw_data: 原始数据字典
            target_col: 目标列名
            task_type: 任务类型 ('classification', 'regression')
            enable_ts_features: 是否启用时间序列特征
            enable_cross_features: 是否启用交叉特征
            enable_selection: 是否启用特征选择
        """
        processed_data = {}
        
        # 1. 自动化特征工程
        print("步骤 1: 运行自动化特征工程...")
        if 'main' in raw_data:
            try:
                feat_tools_framework = AutomatedFeatureToolsFramework()
                feature_matrix, feature_defs = (
                    feat_tools_framework
                    .setup_entityset(raw_data)
                    .generate_features('main')
                )
                
                if target_col and target_col in raw_data['main'].columns:
                    y = raw_data['main'][target_col]
                    X = feature_matrix.drop(columns=[target_col])
                else:
                    print("警告: 未找到目标列，跳过监督特征选择")
                    X = feature_matrix
                    y = None
            except Exception as e:
                print(f"特征工具出错: {e}")
                # 备选方案 - 简单的特征合并
                X = pd.concat(raw_data.values(), axis=1).fillna(0)
                y = raw_data.get('main', pd.DataFrame()).get(target_col)
        else:
            # 如果没有'main'实体，使用第一个数据框
            first_key = list(raw_data.keys())[0]
            X = raw_data[first_key].copy()
            if target_col in X.columns:
                y = X.pop(target_col)  # 从X中移除目标列并保存
            else:
                y = None
        
        # 2. 时间序列特征（如果适用）
        if enable_ts_features and 'timeseries' in raw_data:
            print("步骤 2: 生产时间序列特征...")
            ts_feat_eng = TimeSeriesFeatureEngineering()
            ts_features = ts_feat_eng.create_lag_features(
                raw_data['timeseries'], 
                value_column=raw_data['timeseries'].select_dtypes(
                    include=[np.number]).columns[0]
            )
            # 合并时间序列特征与主特征矩阵
            X = pd.concat([X, ts_features.drop(columns=raw_data['timeseries'].columns)], axis=1)
            
    
        # 3. 交叉特征
        if enable_cross_features and len(X.select_dtypes(include=[np.number]).columns) > 1:
            print("步骤 3: 创建交叉特征...")
            cross_engineer = CrossFeatureEngineering()
            numeric_cols = X.select_dtypes(include=[np.number]).columns[:10]  # 只使用前10个特征，避免过多交互
            X_cross = cross_engineer.create_manual_interactions(X, numeric_cols.tolist())
            X = X_cross
        
        # 4. 风控特征（如果适用）
        if 'risk_data' in raw_data:  
            print("步骤 4: 创建风险领域专用特征...")
            risk_engineer = FintechRiskFeatures()
            if 'bureau' in raw_data:
                bureau_risk_features = risk_engineer.create_aggregated_bureau_features(raw_data['bureau'])
                X = X.merge(bureau_risk_features, left_on='user_id', right_on='person_id', how='left')
        
        # 5. 特征选择（如果提供目标变量）
        if enable_selection and y is not None:
            print("步骤 5: 执行特征选择...")
            
            # 将数据分割为训练集和测试集用于选择
            from sklearn.model_selection import train_test_split
            X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
            
            # 拟合简单模型用于选择
            if task_type == 'classification':
                from sklearn.ensemble import RandomForestClassifier
                model = RandomForestClassifier(n_estimators=50, random_state=42)
            else:
                from sklearn.ensemble import RandomForestRegressor
                model = RandomForestRegressor(n_estimators=50, random_state=42)
            
            model.fit(X_train, y_train)
            
            # 使用混合选择方法
            hybrid_selector = FeatureSelectorHybrid(primary_method='permutation')
            selected_indices = hybrid_selector.fit_selection(X_train, y_train, model=model)
            
            # 裁剪特征集合
            X = X.iloc[:, selected_indices] if len(selected_indices) > 0 else X
        
        print(f"最终特征矩阵形状: {X.shape}")
        
        return X, y


# 8. 决策流程图说明

def print_decision_flow():
    """
    打印特征工程决策流程
    """
    flow_chart = """
                    高级特征工程决策流程图
                   ========================
                             
    开始分析数据结构
           |
           v
    ┌─────────────────┐
    │ 评估数据特征    │
    │ - 数据量级      │
    │ - 数据结构      │
    │ - 时间序列性    │
    │ - 业务领域      │
    └─────────┬───────┘
             |
             v
    ┌─────────────────┐
    │ 自动化特征提取  │
    │ ?              │
    │ 适合复杂关系数据│
    └─────┬─────────┬─┘
          │ 是       │ 否
          v         v
    ┌─────────┐   ┌─────────────┐
    │ TSFresh │   │ 手工设计    │
    │FeatTools│   │ 或领域特征  │
    │AutoFeat │   └─────────────┘
    └─────────┘
          |
          v
    ┌─────────────────┐
    │ 时间序列特征    │
    │ ?              │
    └─────┬───────────┘
          │ 是
          v
    ┌─────────────────┐
    │ 滞后特征        │
    │ 滚动统计        │
    │ 循环编码        │
    │ 变化特征        │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 交叉特征工程    │
    │ ?              │
    └─────┬───────────┘
          │ 是
          v
    ┌─────────────────┐
    │ 低到高阶交叉    │
    │ 比率特征        │
    │ 逻辑交互        │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 域特定特征      │
    │ ?              │
    │ (金融/医疗/等) │
    └─────┬───────────┘
          │ 是
          v
    ┌─────────────────┐
    │ 预定义模板      │
    │ 风险评分        │
    │ 行为指标        │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 特征选择        │
    │ ?              │
    └─────┬───────────┘
          │ 是
          v
    ┌─────────────────┐
    │ - 置换重要性    │
    │ - SHAP分析      │
    │ - 遗传算法      │
    │ - 统计测试      │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 特征质量验证    │
    │ - 重复性检测    │
    │ - 多重共线检测  │
    │ - 有效性评估    │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 清理最终特征集  │
    └─────────────────┘
          |
          v
    ┌─────────────────┐
    │ 构建模型管道     │
    └─────────────────┘
    """
    
    print(flow_chart)


# 9. 使用示例

def example_usage():
    """
    演示如何使用完整的特征工程框架
    """
    print("=== 高级特征工程框架演示 ===\n")
    
    # 示例数据创建
    np.random.seed(42)
    n_samples = 1000
    
    # 创建示例主表数据
    demo_main_data = pd.DataFrame({
        'user_id': range(n_samples),
        'age': np.random.randint(18, 80, n_samples),
        'income': np.random.lognormal(10, 1, n_samples),
        'credit_score': np.random.normal(650, 100, n_samples),
        'requested_amount': np.random.lognormal(10, 0.8, n_samples),
        'target': np.random.binomial(1, 0.1, n_samples)
    })
    
    # 创建示例历史数据
    n_records_per_user = 5
    demo_history_data = pd.DataFrame({
        'user_id': np.repeat(range(n_samples), n_records_per_user),
        'transaction_amount': np.random.normal(1000, 200, n_samples * n_records_per_user),
        'days_ago': np.random.randint(1, 365, n_samples * n_records_per_user)
    })
    
    # 准备数据字典
    sample_data = {
        'main': demo_main_data,
        'history': demo_history_data
    }
    
    print(f"输入数据结构:")
    for name, df in sample_data.items():
        print(f"{name}: 形状={df.shape}, 列名={df.columns.tolist()}")
    
    print(f"\ntarget列值分布:\n{demo_main_data['target'].value_counts()}")
    
    # 实例化流水线
    pipeline = AdvancedFeatureEngineeringPipeline()
    
    # 运行完整流水线
    processed_X, processed_y = pipeline.run_complete_pipeline(
        raw_data=sample_data,
        target_col='target',
        task_type='classification',
        enable_ts_features=True,
        enable_cross_features=True,
        enable_selection=True
    )
    
    print(f"\n处理后的数据:")
    print(f"特征矩阵形状: {processed_X.shape}")
    print(f"目标向量形状: {processed_y.shape if processed_y is not None else 'None'}")
    
    print("\n特征工程完成! 演示结束.")


# 在适当位置添加主执行块
if __name__ == "__main__":
    print("运行高级特征工程框架演示...")
    print_decision_flow()
    print("\n" + "="*60)
    example_usage()

Conclusion

This article has covered the following methods and techniques for advanced feature engineering:

Automated feature engineering: Use FeatureTools, TSFresh, and AutoFeat to generate features and reduce manual work.
Time series features: Capture temporal patterns through lag features, rolling statistics, and cyclical encoding.
Feature interactions: Systematically explore interactions and construct informative composite features.
Feature selection: Combine permutation importance, SHAP values, and genetic algorithms to reduce the feature space.
Domain-specific features: Apply specialized feature engineering techniques to financial risk control.

The framework provides a comprehensive, practical approach spanning data preprocessing through feature selection, with particular attention to financial risk control applications.

With this framework, data science teams can apply advanced feature engineering efficiently and improve predictive modeling, particularly in business applications involving tabular and time series data.

Abstract#

Keywords#

1. Introduction and Theoretical Foundations#

1.1 Key Challenges in Feature Engineering#

1.2 Method Categories#

2. Automated Feature Engineering Frameworks#

2.1 A Practical FeatureTools Framework#

2.2 A Practical TSFresh Framework#

2.3 A Practical AutoFeat Framework#

3. Time Series Feature Engineering#

3.1 Lag Features and Rolling Statistics#

3.2 Cyclical Encoding Explained#

4. Feature Interaction Strategies#

4.1 Generating Lower- and Higher-Order Feature Interactions#

4.2 Pitfall Detection and Quality Validation#

5. Feature Selection Frameworks#

5.1 Permutation Importance#

5.2 SHAP-Based Importance#

5.3 Feature Selection with Genetic Algorithms#

6. Domain-Specific Features for Financial Risk Control#

6.1 Risk Ratios and Indicators#

6.2 Behavioral Sequence Features#

Conclusion#