未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适合有 Python 和机器学习基础的读者学习特征选择、变换、编码、降维及模拟信贷分类，不宜作为可直接复用的生产流水线或真实风控效果报告。开头流水线是尚未实现私有方法的流程骨架；目标编码、显式传入部分 columns 参数和稀疏矩阵输入仍有代码限制。信贷示例先在全量数据上拟合标准化器、后拆分训练与测试数据，所得 AUC 不能视为严格隔离的泛化评估；模拟标签也没有真实的未来12个月观察窗口或原始特征基准实验。当前文末还重复收录了几乎整篇正文，阅读时应以两处相同内容为同一示例，而非两个独立版本。
特征工程：机器学习模型性能的核心驱动
概述
特征工程是机器学习中的重要技术环节，可影响模型性能；实际效果还取决于数据质量、任务定义、学习算法和验证方式。它通过从原始数据中提取、构造、选择和变换特征，将业务知识转化为算法可理解的数值表示，是数据科学家的核心技能。
特征工程的技术定义与重要性
特征工程本质上是一个数据表示学习过程，其核心目标是构建最优的特征空间，使得机器学习算法能够有效学习数据中的潜在模式。
技术价值体现
- 提升模型性能：优质特征可能改善模型表现；10-20%的提升仅可作为待验证的示例假设，实际幅度需通过明确指标和基准对照实验确定
- 减少数据需求：良好的特征表示降低对大规模数据的依赖
- 增强模型可解释性：有意义的特征提供业务洞察
- 降低计算复杂度：精选特征减少训练和推理成本
标准特征工程管道
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder, PolynomialFeatures
from sklearn.feature_selection import SelectKBest, f_classif, RFE
from sklearn.ensemble import RandomForestRegressor
class FeatureEngineeringPipeline:
"""标准化特征工程流水线"""
def __init__(self):
self.scalers = {}
self.encoders = {}
self.selectors = {}
def fit_transform(self, X, y=None):
"""特征工程流程骨架：须先实现下述私有方法才能执行"""
# 1. 特征构造
X = self._create_features(X)
# 2. 特征编码
X = self._encode_features(X)
# 3. 特征变换
X = self._transform_features(X)
# 4. 特征选择
if y is not None:
X = self._select_features(X, y)
return X
特征选择技术
特征选择基于信息论和统计学习理论，通过量化特征与目标变量的关系强度，剔除冗余和无关特征。
过滤法（Filter Methods）
基于统计测试和信息论指标，独立于具体学习算法。
from sklearn.feature_selection import VarianceThreshold, SelectKBest, f_classif, mutual_info_classif
from scipy.stats import chi2_contingency
class FilterSelector:
"""过滤法特征选择器"""
def variance_filter(self, X, threshold=0.01):
"""方差过滤：移除低方差特征"""
selector = VarianceThreshold(threshold=threshold)
X_filtered = selector.fit_transform(X)
return X.columns[selector.get_support()]
def univariate_filter(self, X, y, k=10, score_func=f_classif):
"""单变量统计测试"""
selector = SelectKBest(score_func=score_func, k=k)
selector.fit(X, y)
# 获取特征得分
scores = pd.DataFrame({
'feature': X.columns,
'score': selector.scores_,
'pvalue': selector.pvalues_
}).sort_values('score', ascending=False)
return scores.head(k)['feature'].tolist()
def mutual_info_filter(self, X, y, k=10):
"""互信息特征选择"""
mi_scores = mutual_info_classif(X, y)
mi_df = pd.DataFrame({
'feature': X.columns,
'mi_score': mi_scores
}).sort_values('mi_score', ascending=False)
return mi_df.head(k)['feature'].tolist()
# 使用示例
selector = FilterSelector()
# 假设X是特征矩阵，y是目标变量
top_features = selector.univariate_filter(X, y, k=15)
包装法（Wrapper Methods）
基于模型性能评估，使用启发式搜索在特征子集空间中寻找最优组合。
from sklearn.feature_selection import RFE, SequentialFeatureSelector
from sklearn.model_selection import cross_val_score
from sklearn.linear_model import LogisticRegression
class WrapperSelector:
"""包装法特征选择器"""
def __init__(self, estimator=None):
self.estimator = estimator or LogisticRegression(random_state=42)
def recursive_elimination(self, X, y, n_features=10):
"""递归特征消除（RFE）"""
rfe = RFE(self.estimator, n_features_to_select=n_features)
rfe.fit(X, y)
ranking_df = pd.DataFrame({
'feature': X.columns,
'ranking': rfe.ranking_,
'selected': rfe.support_
}).sort_values('ranking')
return X.columns[rfe.support_].tolist(), ranking_df
def forward_selection(self, X, y, n_features=10, cv=5):
"""前向特征选择"""
sfs = SequentialFeatureSelector(
self.estimator, n_features_to_select=n_features,
direction='forward', cv=cv, scoring='accuracy'
)
sfs.fit(X, y)
return X.columns[sfs.support_].tolist()
def backward_elimination(self, X, y, n_features=10, cv=5):
"""后向特征消除"""
sbs = SequentialFeatureSelector(
self.estimator, n_features_to_select=n_features,
direction='backward', cv=cv, scoring='accuracy'
)
sbs.fit(X, y)
return X.columns[sbs.support_].tolist()
# 使用示例
wrapper = WrapperSelector()
selected_features, ranking = wrapper.recursive_elimination(X, y, n_features=15)
嵌入法（Embedded Methods）
在模型训练过程中集成特征选择，如L1正则化、树模型特征重要性。
from sklearn.linear_model import LassoCV, Lasso
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_selection import SelectFromModel
class EmbeddedSelector:
"""嵌入法特征选择器"""
def lasso_selection(self, X, y, alpha=None):
"""基于Lasso的L1正则化特征选择"""
if alpha is None:
lasso = LassoCV(cv=5, random_state=42)
else:
lasso = Lasso(alpha=alpha, random_state=42)
selector = SelectFromModel(lasso)
selector.fit(X, y)
# 获取特征系数
coef_df = pd.DataFrame({
'feature': X.columns,
'coefficient': selector.estimator_.coef_,
'abs_coefficient': np.abs(selector.estimator_.coef_)
}).sort_values('abs_coefficient', ascending=False)
return X.columns[selector.get_support()].tolist(), coef_df
def tree_importance_selection(self, X, y, threshold='median'):
"""基于树模型重要性的特征选择"""
rf = RandomForestClassifier(n_estimators=100, random_state=42)
selector = SelectFromModel(rf, threshold=threshold)
selector.fit(X, y)
# 获取特征重要性
importance_df = pd.DataFrame({
'feature': X.columns,
'importance': selector.estimator_.feature_importances_
}).sort_values('importance', ascending=False)
return X.columns[selector.get_support()].tolist(), importance_df
# 使用示例
embedded = EmbeddedSelector()
lasso_features, coef_info = embedded.lasso_selection(X, y)
tree_features, importance_info = embedded.tree_importance_selection(X, y)
特征变换技术
特征变换基于数学变换理论，将原始特征映射到更适合学习算法的特征空间。
数值特征标准化
解决不同量纲和分布带来的学习偏差。
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler, QuantileTransformer
from sklearn.preprocessing import PowerTransformer
class NumericalTransformer:
"""数值特征变换器"""
def __init__(self):
self.transformers = {}
def standard_scaling(self, X, columns=None):
"""Z-score标准化: (x - μ) / σ"""
columns = columns or X.select_dtypes(include=[np.number]).columns
scaler = StandardScaler()
X_scaled = X.copy()
X_scaled[columns] = scaler.fit_transform(X[columns])
self.transformers['standard'] = scaler
return X_scaled
def robust_scaling(self, X, columns=None):
"""鲁棒标准化: (x - median) / IQR"""
columns = columns or X.select_dtypes(include=[np.number]).columns
scaler = RobustScaler()
X_scaled = X.copy()
X_scaled[columns] = scaler.fit_transform(X[columns])
self.transformers['robust'] = scaler
return X_scaled
def power_transform(self, X, columns=None, method='yeo-johnson'):
"""幂变换：使数据更接近正态分布"""
columns = columns or X.select_dtypes(include=[np.number]).columns
transformer = PowerTransformer(method=method)
X_transformed = X.copy()
X_transformed[columns] = transformer.fit_transform(X[columns])
self.transformers['power'] = transformer
return X_transformed
def quantile_transform(self, X, columns=None, n_quantiles=1000):
"""分位数变换：映射到均匀分布"""
columns = columns or X.select_dtypes(include=[np.number]).columns
transformer = QuantileTransformer(n_quantiles=n_quantiles, random_state=42)
X_transformed = X.copy()
X_transformed[columns] = transformer.fit_transform(X[columns])
self.transformers['quantile'] = transformer
return X_transformed
# 使用示例
transformer = NumericalTransformer()
X_standardized = transformer.standard_scaling(X)
X_robust = transformer.robust_scaling(X)
类别特征编码
将离散类别信息转换为数值表示。
from sklearn.preprocessing import LabelEncoder, OneHotEncoder, OrdinalEncoder
import category_encoders as ce
class CategoricalEncoder:
"""类别特征编码器"""
def __init__(self):
self.encoders = {}
def label_encoding(self, X, columns=None):
"""逐列标签映射示例：LabelEncoder用于目标标签，不保证特征的业务顺序；有序特征应使用显式指定类别顺序的OrdinalEncoder"""
columns = columns or X.select_dtypes(include=['object']).columns
X_encoded = X.copy()
for col in columns:
le = LabelEncoder()
X_encoded[col] = le.fit_transform(X[col].astype(str))
self.encoders[f'label_{col}'] = le
return X_encoded
def onehot_encoding(self, X, columns=None, drop_first=True):
"""独热编码：适用于无序类别"""
columns = columns or X.select_dtypes(include=['object']).columns
X_encoded = X.copy()
for col in columns:
# 获取独热编码结果
encoder = OneHotEncoder(drop='first' if drop_first else None, sparse_output=False)
encoded = encoder.fit_transform(X[[col]])
# 创建新列名
feature_names = [f'{col}_{cat}' for cat in encoder.categories_[0]]
if drop_first:
feature_names = feature_names[1:]
# 添加到DataFrame
encoded_df = pd.DataFrame(encoded, columns=feature_names, index=X.index)
X_encoded = pd.concat([X_encoded.drop(columns=[col]), encoded_df], axis=1)
self.encoders[f'onehot_{col}'] = encoder
return X_encoded
def target_encoding(self, X, y, columns=None, smoothing=1.0):
"""目标编码：用目标变量的统计量编码类别"""
columns = columns or X.select_dtypes(include=['object']).columns
X_encoded = X.copy()
for col in columns:
encoder = ce.TargetEncoder(cols=[col], smoothing=smoothing)
X_encoded[f'{col}_target'] = encoder.fit_transform(X[col], y)
self.encoders[f'target_{col}'] = encoder
return X_encoded
# 使用示例
encoder = CategoricalEncoder()
X_onehot = encoder.onehot_encoding(X, columns=['category'])
X_target = encoder.target_encoding(X, y, columns=['category'])
特征创造技术
通过数学变换和业务逻辑创造新的表征维度。
多项式特征和交互特征
from sklearn.preprocessing import PolynomialFeatures
from itertools import combinations
class FeatureCreator:
"""特征创造器"""
def polynomial_features(self, X, degree=2, interaction_only=False):
"""生成多项式特征"""
poly = PolynomialFeatures(
degree=degree,
interaction_only=interaction_only,
include_bias=False
)
X_poly = poly.fit_transform(X)
feature_names = poly.get_feature_names_out(X.columns)
return pd.DataFrame(X_poly, columns=feature_names, index=X.index)
def interaction_features(self, X, max_combinations=2):
"""生成交互特征"""
X_interactions = X.copy()
numeric_cols = X.select_dtypes(include=[np.number]).columns
for r in range(2, min(max_combinations + 1, len(numeric_cols) + 1)):
for combo in combinations(numeric_cols, r):
feature_name = '_x_'.join(combo)
X_interactions[feature_name] = X[list(combo)].prod(axis=1)
return X_interactions
def ratio_features(self, X, numerator_cols, denominator_cols):
"""生成比率特征"""
X_ratios = X.copy()
for num_col in numerator_cols:
for den_col in denominator_cols:
if num_col != den_col:
ratio_name = f'{num_col}_div_{den_col}'
# 避免除零
X_ratios[ratio_name] = X[num_col] / (X[den_col] + 1e-8)
return X_ratios
def aggregation_features(self, df, group_cols, agg_cols, agg_funcs=['mean', 'std', 'max', 'min']):
"""生成聚合特征"""
df_agg = df.copy()
for group_col in group_cols:
for agg_col in agg_cols:
if group_col != agg_col:
group_stats = df.groupby(group_col)[agg_col].agg(agg_funcs)
# 重命名列
group_stats.columns = [f'{agg_col}_{func}_by_{group_col}' for func in agg_funcs]
# 合并回原数据
df_agg = df_agg.merge(group_stats, left_on=group_col, right_index=True, how='left')
return df_agg
# 使用示例
creator = FeatureCreator()
X_poly = creator.polynomial_features(X, degree=2, interaction_only=True)
X_ratios = creator.ratio_features(X, ['feature1'], ['feature2'])
降维技术
减少特征维度，保留主要信息，解决维度灾难问题。
PCA主成分分析
from sklearn.decomposition import PCA, TruncatedSVD
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
class DimensionReducer:
"""降维器"""
def pca_reduction(self, X, n_components=0.95, plot=True):
"""主成分分析降维"""
if isinstance(n_components, float) and n_components < 1:
# 保留指定方差比例
pca = PCA(n_components=n_components, random_state=42)
else:
# 保留指定数量的主成分
pca = PCA(n_components=int(n_components), random_state=42)
X_reduced = pca.fit_transform(X)
# 创建结果DataFrame
columns = [f'PC{i+1}' for i in range(X_reduced.shape[1])]
X_pca = pd.DataFrame(X_reduced, columns=columns, index=X.index)
# 打印统计信息
print(f"原始维度: {X.shape[1]}")
print(f"降维后维度: {X_reduced.shape[1]}")
print(f"累积解释方差比: {pca.explained_variance_ratio_.sum():.3f}")
return X_pca, pca
def lda_reduction(self, X, y, n_components=None):
"""线性判别分析降维（监督降维）"""
n_classes = len(np.unique(y))
if n_components is None:
n_components = min(X.shape[1], n_classes - 1)
lda = LinearDiscriminantAnalysis(n_components=n_components)
X_reduced = lda.fit_transform(X, y)
columns = [f'LD{i+1}' for i in range(X_reduced.shape[1])]
X_lda = pd.DataFrame(X_reduced, columns=columns, index=X.index)
print(f"LDA降维: {X.shape[1]} -> {X_reduced.shape[1]}")
return X_lda, lda
def truncated_svd_reduction(self, X, n_components=50):
"""截断SVD降维（适用于稀疏矩阵）"""
svd = TruncatedSVD(n_components=n_components, random_state=42)
X_reduced = svd.fit_transform(X)
columns = [f'SVD{i+1}' for i in range(X_reduced.shape[1])]
X_svd = pd.DataFrame(X_reduced, columns=columns, index=X.index)
print(f"SVD降维: {X.shape[1]} -> {X_reduced.shape[1]}")
print(f"解释方差比: {svd.explained_variance_ratio_.sum():.3f}")
return X_svd, svd
# 使用示例
reducer = DimensionReducer()
X_pca, pca_model = reducer.pca_reduction(X, n_components=0.95)
X_lda, lda_model = reducer.lda_reduction(X, y, n_components=2)
以下以假设的银行信贷风险评估需求为背景，使用模拟数据演示特征工程与分类建模，不代表真实银行项目或经验证的违约预测效果。
业务场景分析
- 目标：预测客户未来12个月的违约概率
- 数据源：客户基本信息、交易流水、征信记录、行为日志
- 挑战：数据维度高、类别特征多、时间序列特征复杂
完整特征工程实现
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 模拟银行客户数据
np.random.seed(42)
n_customers = 10000
# 构造原始数据
data = {
'customer_id': range(1, n_customers + 1),
'age': np.random.normal(40, 15, n_customers).astype(int),
'income': np.random.lognormal(10, 0.5, n_customers),
'loan_amount': np.random.lognormal(9, 0.8, n_customers),
'loan_term_months': np.random.choice([12, 24, 36, 48, 60], n_customers),
'education': np.random.choice(['高中', '本科', '硕士', '博士'], n_customers),
'employment_type': np.random.choice(['全职', '兼职', '自雇', '无业'], n_customers),
'city_tier': np.random.choice([1, 2, 3], n_customers),
'credit_score': np.random.normal(650, 100, n_customers).astype(int),
'previous_loans': np.random.poisson(2, n_customers),
'account_balance': np.random.exponential(5000, n_customers),
'monthly_transactions': np.random.poisson(20, n_customers),
'loan_start_date': pd.date_range('2020-01-01', periods=n_customers, freq='D')[:n_customers]
}
df = pd.DataFrame(data)
# 构造目标变量（违约标签）
def create_default_target(row):
"""基于业务逻辑构造违约概率"""
risk_score = 0
# 收入风险
if row['income'] < 30000:
risk_score += 0.3
elif row['income'] < 50000:
risk_score += 0.1
# 征信风险
if row['credit_score'] < 600:
risk_score += 0.4
elif row['credit_score'] < 650:
risk_score += 0.2
# 负债比例风险
debt_ratio = row['loan_amount'] / (row['income'] + 1)
if debt_ratio > 0.5:
risk_score += 0.3
elif debt_ratio > 0.3:
risk_score += 0.15
return np.random.binomial(1, min(risk_score, 0.8))
df['default'] = df.apply(create_default_target, axis=1)
print(f"数据维度: {df.shape}")
print(f"违约率: {df['default'].mean():.2%}")
class FinancialFeatureEngineer:
"""金融风控特征工程器"""
def __init__(self):
self.transformers = {}
def engineer_features(self, df):
"""执行完整特征工程流程"""
df_features = df.copy()
# 1. 基础特征变换
df_features = self._create_basic_features(df_features)
# 2. 业务特征构造
df_features = self._create_business_features(df_features)
# 3. 时间特征提取
df_features = self._create_time_features(df_features)
# 4. 类别特征编码
df_features = self._encode_categorical(df_features)
# 5. 数值特征标准化
df_features = self._normalize_numerical(df_features)
return df_features
def _create_basic_features(self, df):
"""创建基础派生特征"""
# 收入分箱
df['income_tier'] = pd.cut(df['income'],
bins=[0, 30000, 60000, 100000, float('inf')],
labels=['低收入', '中等收入', '高收入', '高净值'])
# 年龄分组
df['age_group'] = pd.cut(df['age'],
bins=[0, 25, 35, 50, 65, 100],
labels=['青年', '青壮年', '中年', '中老年', '老年'])
# 征信等级
df['credit_grade'] = pd.cut(df['credit_score'],
bins=[0, 580, 620, 680, 740, 850],
labels=['差', '一般', '良好', '优秀', '极佳'])
return df
def _create_business_features(self, df):
"""创建业务相关特征"""
# 负债收入比
df['debt_to_income'] = df['loan_amount'] / (df['income'] + 1)
# 月供收入比（假设利率5%）
monthly_rate = 0.05 / 12
df['monthly_payment'] = (df['loan_amount'] * monthly_rate *
(1 + monthly_rate)**df['loan_term_months']) / \
((1 + monthly_rate)**df['loan_term_months'] - 1)
df['payment_to_income'] = df['monthly_payment'] / (df['income'] / 12)
# 资产流动性
df['liquidity_ratio'] = df['account_balance'] / df['monthly_payment']
# 信贷经验
df['credit_experience'] = np.where(df['previous_loans'] > 0, 'experienced', 'new')
# 客户活跃度
df['transaction_frequency'] = df['monthly_transactions'] / 30 # 日均交易次数
# 风险评分（综合指标）
df['risk_score'] = (df['debt_to_income'] * 0.4 +
df['payment_to_income'] * 0.3 +
(1 - df['credit_score']/850) * 0.3)
return df
def _create_time_features(self, df):
"""创建时间相关特征"""
# 提取日期特征
df['loan_month'] = df['loan_start_date'].dt.month
df['loan_quarter'] = df['loan_start_date'].dt.quarter
df['loan_weekday'] = df['loan_start_date'].dt.weekday
# 季节性特征
df['is_year_end'] = df['loan_month'].isin([11, 12]).astype(int)
df['is_spring_festival'] = df['loan_month'].isin([1, 2]).astype(int)
return df
def _encode_categorical(self, df):
"""编码类别特征"""
from sklearn.preprocessing import LabelEncoder, OneHotEncoder
# 标签映射示例：以下LabelEncoder不保留业务顺序，不能视为正确的有序类别编码
ordinal_features = ['income_tier', 'age_group', 'credit_grade']
for col in ordinal_features:
if col in df.columns:
le = LabelEncoder()
df[f'{col}_encoded'] = le.fit_transform(df[col].astype(str))
self.transformers[f'le_{col}'] = le
# 独热编码（无序类别）
nominal_features = ['education', 'employment_type', 'credit_experience']
for col in nominal_features:
if col in df.columns:
dummies = pd.get_dummies(df[col], prefix=col, drop_first=True)
df = pd.concat([df, dummies], axis=1)
return df
def _normalize_numerical(self, df):
"""标准化数值特征"""
from sklearn.preprocessing import RobustScaler
numerical_features = [
'age', 'income', 'loan_amount', 'credit_score',
'account_balance', 'debt_to_income', 'payment_to_income',
'liquidity_ratio', 'risk_score', 'transaction_frequency'
]
scaler = RobustScaler()
df[numerical_features] = scaler.fit_transform(df[numerical_features])
self.transformers['scaler'] = scaler
return df
# 执行特征工程
feature_engineer = FinancialFeatureEngineer()
df_engineered = feature_engineer.engineer_features(df)
print(f"特征工程后维度: {df_engineered.shape}")
print(f"新增特征数: {df_engineered.shape[1] - df.shape[1]}")
模型验证与特征重要性分析
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, roc_auc_score, roc_curve
import matplotlib.pyplot as plt
# 准备建模数据
feature_cols = df_engineered.select_dtypes(include=[np.number]).columns.tolist()
feature_cols = [col for col in feature_cols if col not in ['customer_id', 'default']]
X = df_engineered[feature_cols]
y = df_engineered['default']
# 分割数据
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 训练多个模型
models = {
'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42),
'Gradient Boosting': GradientBoostingClassifier(random_state=42),
'Logistic Regression': LogisticRegression(random_state=42, max_iter=1000)
}
results = {}
for name, model in models.items():
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
y_proba = model.predict_proba(X_test)[:, 1]
auc_score = roc_auc_score(y_test, y_proba)
results[name] = {
'model': model,
'auc': auc_score,
'predictions': y_proba
}
print(f"\n{name} - AUC: {auc_score:.4f}")
print(classification_report(y_test, y_pred))
# 特征重要性分析
best_model = results['Random Forest']['model']
feature_importance = pd.DataFrame({
'feature': X.columns,
'importance': best_model.feature_importances_
}).sort_values('importance', ascending=False)
print("\n=== TOP 10 重要特征 ===")
print(feature_importance.head(10))
# 业务价值总结
print(f"\n=== 特征工程业务价值 ===")
print(f"1. 模拟数据测试集AUC：{results['Random Forest']['auc']:.4f}；未建立基准对照，不能据此判断特征工程带来的提升")
print(f"2. 关键风险因子：{feature_importance.head(3)['feature'].tolist()}")
print(f"3. 特征构造与筛选：原始表包含{len(df.columns)}列，建模使用{len(feature_cols)}个数值特征；此处并未执行降维")
特征工程最佳实践
核心原则
- 业务驱动：基于业务理解构造有意义的特征
- 避免数据泄露：确保特征不包含未来信息
- 可解释性：优先选择可解释的特征变换
- 计算效率：平衡特征质量与计算成本
常见陷阱及规避策略
时间泄露
# 错误：使用未来信息
df['future_avg'] = df.groupby('user_id')['value'].transform('mean')
# 前提：df已按用户和实际事件时间排序；在各用户内移位并对齐原索引
df['historical_avg'] = df.groupby('user_id')['value'].transform(
lambda s: s.expanding().mean().shift(1)
)
目标泄露
# 错误：直接使用目标相关特征
df['target_mean_by_category'] = df.groupby('category')['target'].transform('mean')
# K折折外目标编码示例：并非留一法；未见类别还需设置仅由训练折计算的回退值
from sklearn.model_selection import KFold
def target_encoding_cv(X, y, column, cv=5):
kf = KFold(n_splits=cv, shuffle=True, random_state=42)
encoded_values = np.zeros(len(X))
for train_idx, val_idx in kf.split(X):
target_mean = y.iloc[train_idx].groupby(X[column].iloc[train_idx]).mean()
encoded_values[val_idx] = X[column].iloc[val_idx].map(target_mean)
return encoded_values
过拟合控制
# 特征选择时使用交叉验证
from sklearn.feature_selection import RFECV
def robust_feature_selection(X, y, estimator):
selector = RFECV(estimator, cv=5, scoring='roc_auc')
X_selected = selector.fit_transform(X, y)
return X.columns[selector.support_], selector.cv_results_['mean_test_score']
技术总结
- 特征选择：过滤法快速筛选，包装法精确选择，嵌入法集成优化
- 特征变换：标准化处理量纲，编码转换类别，降维缓解维度灾难
- 特征创造：多项式捕获非线性，交互发现组合效应，聚合提取统计模式
- 验证策略：交叉验证防过拟合，时间分割避泄露，业务验证保可用
特征工程是机器学习成功的关键，需要技术能力与业务理解的深度结合，持续迭代优化以实现最佳效果。
创建时间：2024-06-29
最后更新：2024-09-02
预计阅读时长：45分钟
相关文档链接：