机器学习特征预处理之特征选择 一、核心作用1.降维Dimensionality Reduction减少特征数量从原始特征中筛选出最重要的子集缓解维数灾难高维数据特征数样本数时特别有效加速模型训练减少计算复杂度2.提升模型性能去除噪声特征消除不相关或冗余特征带来的干扰防止过拟合减少模型复杂度提高泛化能力改善可解释性聚焦关键特征便于理解业务逻辑二、两种方法的区别与作用方法1随机森林RFpython# 作用原理 特征重要性 所有决策树中该特征带来的不纯度减少 / 树的数量适用场景✅ 特征与目标存在非线性关系✅ 特征之间存在交互作用✅ 数据包含分类特征经编码后❌ 高维稀疏数据效果不如Lasso优势自动捕捉复杂模式对异常值不敏感天然处理混合型数据示例应用python# 房价预测 - 非线性特征重要性排序 feature_importance { 地理位置: 0.35, # 非线性影响 面积: 0.28, 房龄: 0.15, # 非线性衰减 房间数: 0.12, 楼层: 0.08, 装修程度: 0.02 # 噪声特征被剔除 }方法2Lasso回归python# 作用原理 损失函数 MSE α × Σ|系数| # L1正则化将不重要的特征系数压缩为0适用场景✅高维数据特征数 样本数✅ 特征与目标存在线性关系✅ 需要稀疏解只保留少数特征✅特征筛选回归一体化优势产生稀疏模型自动选择特征系数大小直接表示重要性数学性质良好理论支持强示例应用python# 基因表达数据分析10000个基因100个样本 lasso_coefficients { 基因_A: 2.3, # 重要特征 基因_B: 1.8, 基因_C: 0.0, # 被剔除 ... # 9900多个系数变为0 } # 最终只保留几个关键基因三、调参功能的作用自动调参auto_tuneTruepython# 对随机森林的调参 param_grid { n_estimators: [50, 100, 200], # 树的数量 max_depth: [None, 10, 20, 30], # 树的深度 min_samples_split: [2, 5, 10] # 分裂所需最小样本数 }作用避免手动试错找到最优模型配置平衡偏差-方差实际效果对比python# 未调参保留12个特征R²0.78 # 调参后保留15个特征R²0.85 # 性能提升 ~9%四、阈值参数的作用threshold 控制筛选强度阈值类型作用结果median保留重要性高于中位数的特征通常保留50%特征mean保留高于平均值的特征保留50%特征0.01(数值)保留重要性0.01的特征自定义筛选强度大数值如10只保留极重要的特征特征数很少强正则化小数值如0.001保留几乎所有特征弱筛选几乎无降维示例def feature_selection(X_train, y_train, X_test, methodrf, thresholdmedian): 特征选择 if method rf: selector SelectFromModel(RandomForestRegressor(n_estimators100, random_state42), thresholdthreshold) elif method lasso: selector SelectFromModel(Lasso(alpha0.01), thresholdthreshold) else: raise ValueError(method must be rf or lasso) selector.fit(X_train, y_train) cols_kept X_train.columns[selector.get_support()].tolist() return X_train[cols_kept], X_test[cols_kept]复杂import numpy as np import pandas as pd from sklearn.feature_selection import SelectFromModel from sklearn.ensemble import RandomForestRegressor from sklearn.linear_model import Lasso from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline from sklearn.model_selection import GridSearchCV, cross_val_score from sklearn.base import BaseEstimator, TransformerMixin from typing import Optional, Union, List, Tuple, Dict, Literal import warnings warnings.filterwarnings(ignore) class FeatureSelector(BaseEstimator, TransformerMixin): 完善的特性选择器支持多种方法和自动调参 Parameters ---------- method : str, defaultrf 特征选择方法rf随机森林或 lassoLasso回归 threshold : str or float, defaultmedian 选择阈值median、mean或浮点数 cv_folds : int, default5 交叉验证折数用于自动调参 auto_tune : bool, defaultFalse 是否自动调参GridSearchCV handle_nan : bool, defaultTrue 是否自动处理缺失值填充中位数/众数 random_state : int, default42 随机种子 def __init__( self, method: Literal[rf, lasso] rf, threshold: Union[str, float] median, cv_folds: int 5, auto_tune: bool False, handle_nan: bool True, random_state: int 42 ): self.method method self.threshold threshold self.cv_folds cv_folds self.auto_tune auto_tune self.handle_nan handle_nan self.random_state random_state self.selector_ None self.scaler_ None self.kept_features_ None self.feature_importances_ None self.performance_metrics_ {} def _handle_missing_values(self, X: pd.DataFrame) - pd.DataFrame: 处理缺失值 if not self.handle_nan: if X.isnull().any().any(): raise ValueError(数据包含缺失值请先处理或设置 handle_nanTrue) return X X_copy X.copy() for col in X_copy.columns: if X_copy[col].dtype in [float64, int64]: # 数值型填充中位数 median_val X_copy[col].median() X_copy[col].fillna(median_val, inplaceTrue) elif X_copy[col].dtype object or X_copy[col].dtype.name category: # 类别型填充众数 mode_val X_copy[col].mode()[0] if not X_copy[col].mode().empty else unknown X_copy[col].fillna(mode_val, inplaceTrue) # 转换为数值编码 X_copy[col] X_copy[col].astype(category).cat.codes else: # 其他类型尝试转换为数值 try: X_copy[col] pd.to_numeric(X_copy[col], errorscoerce) X_copy[col].fillna(X_copy[col].median(), inplaceTrue) except: raise ValueError(f无法处理列 {col} 的数据类型: {X_copy[col].dtype}) return X_copy def _get_estimator(self): 获取基础估算器 if self.method rf: base_params { n_estimators: 100, random_state: self.random_state, n_jobs: -1 } if self.auto_tune: # 定义RF的超参数搜索空间 param_grid { n_estimators: [50, 100, 200], max_depth: [None, 10, 20, 30], min_samples_split: [2, 5, 10], min_samples_leaf: [1, 2, 4] } base_estimator RandomForestRegressor(**base_params) return GridSearchCV( base_estimator, param_grid, cvself.cv_folds, scoringneg_mean_squared_error, n_jobs-1 ) else: return RandomForestRegressor(**base_params) elif self.method lasso: base_params { alpha: 0.01, random_state: self.random_state, max_iter: 1000 } if self.auto_tune: # 定义Lasso的超参数搜索空间 param_grid { alpha: np.logspace(-4, 1, 20) # 0.0001 到 10 } base_estimator Lasso(**base_params) return GridSearchCV( base_estimator, param_grid, cvself.cv_folds, scoringneg_mean_squared_error, n_jobs-1 ) else: return Lasso(**base_params) else: raise ValueError(method must be rf or lasso) def fit(self, X: pd.DataFrame, y: pd.Series): 拟合特征选择器 Returns ------- self : FeatureSelector # 保存原始列名 self.original_features_ X.columns.tolist() # 处理缺失值 X_clean self._handle_missing_values(X) # Lasso需要标准化 if self.method lasso: self.scaler_ StandardScaler() X_scaled pd.DataFrame( self.scaler_.fit_transform(X_clean), columnsX_clean.columns, indexX_clean.index ) else: X_scaled X_clean # 创建选择器 estimator self._get_estimator() # 添加阈值验证 if isinstance(self.threshold, str) and self.threshold not in [median, mean]: raise ValueError(threshold must be median, mean, or a numeric value) self.selector_ SelectFromModel( estimator, thresholdself.threshold, prefitFalse ) # 拟合 self.selector_.fit(X_scaled, y) # 获取选中的特征 support_mask self.selector_.get_support() self.kept_features_ [col for col, keep in zip(X.columns, support_mask) if keep] # 获取特征重要性 if hasattr(estimator, feature_importances_): self.feature_importances_ dict(zip(X.columns, estimator.feature_importances_)) elif hasattr(estimator, coef_): self.feature_importances_ dict(zip(X.columns, np.abs(estimator.coef_))) elif hasattr(estimator, best_estimator_): best_est estimator.best_estimator_ if hasattr(best_est, feature_importances_): self.feature_importances_ dict(zip(X.columns, best_est.feature_importances_)) elif hasattr(best_est, coef_): self.feature_importances_ dict(zip(X.columns, np.abs(best_est.coef_))) # 记录性能指标 if self.auto_tune: self.performance_metrics_[best_params] estimator.best_params_ self.performance_metrics_[best_score] estimator.best_score_ # 计算交叉验证性能 cv_scores cross_val_score( estimator, X_scaled, y, cvself.cv_folds, scoringr2 ) self.performance_metrics_[cv_mean_r2] cv_scores.mean() self.performance_metrics_[cv_std_r2] cv_scores.std() self.performance_metrics_[cv_scores] cv_scores return self def transform(self, X: pd.DataFrame) - pd.DataFrame: 转换数据只保留选中的特征 if self.kept_features_ is None: raise RuntimeError(必须先调用 fit()) # 处理缺失值 X_clean self._handle_missing_values(X) # 确保列存在 missing_cols set(self.kept_features_) - set(X_clean.columns) if missing_cols: raise ValueError(f数据中缺少选中的特征: {missing_cols}) return X_clean[self.kept_features_] def fit_transform(self, X: pd.DataFrame, y: pd.Series) - pd.DataFrame: 同时进行拟合和转换 self.fit(X, y) return self.transform(X) def get_feature_importance_df(self) - pd.DataFrame: 获取特征重要性DataFrame if self.feature_importances_ is None: return pd.DataFrame() df_importance pd.DataFrame( list(self.feature_importances_.items()), columns[feature, importance] ) df_importance df_importance.sort_values(importance, ascendingFalse) df_importance[selected] df_importance[feature].isin(self.kept_features_) return df_importance def get_summary(self) - Dict: 获取选择摘要 return { method: self.method, threshold: self.threshold, total_features: len(self.original_features_), selected_features: len(self.kept_features_), removed_features: len(self.original_features_) - len(self.kept_features_), kept_features: self.kept_features_, performance: self.performance_metrics_ } # 便捷函数版本向后兼容 def feature_selection( X_train: pd.DataFrame, y_train: pd.Series, X_test: pd.DataFrame, method: Literal[rf, lasso] rf, threshold: Union[str, float] median, auto_tune: bool False, handle_nan: bool True, return_selector: bool False, **kwargs ) - Union[ Tuple[pd.DataFrame, pd.DataFrame], Tuple[pd.DataFrame, pd.DataFrame, FeatureSelector] ]: 特征选择便捷函数 Parameters ---------- X_train, y_train : 训练数据 X_test : 测试数据 method : rf 或 lasso threshold : median, mean 或数值 auto_tune : 是否自动调参 handle_nan : 是否处理缺失值 return_selector : 是否返回选择器对象 **kwargs : 传递给 FeatureSelector 的其他参数 Returns ------- X_train_selected, X_test_selected : 选择后的数据 [optional] selector : FeatureSelector 实例 selector FeatureSelector( methodmethod, thresholdthreshold, auto_tuneauto_tune, handle_nanhandle_nan, **kwargs ) X_train_selected selector.fit_transform(X_train, y_train) X_test_selected selector.transform(X_test) if return_selector: return X_train_selected, X_test_selected, selector return X_train_selected, X_test_selected # 使用示例 if __name__ __main__: # 生成示例数据 np.random.seed(42) n_samples 200 n_features 20 X pd.DataFrame( np.random.randn(n_samples, n_features), columns[ffeature_{i} for i in range(n_features)] ) y pd.Series( 3 * X[feature_0] 2 * X[feature_5] - X[feature_10] 0.5 * np.random.randn(n_samples) ) # 模拟测试集 X_test pd.DataFrame( np.random.randn(50, n_features), columns[ffeature_{i} for i in range(n_features)] ) # 1. 基础用法 print( 基础用法 ) X_train_sel, X_test_sel feature_selection(X, y, X_test, methodrf) print(f原始特征数: {X.shape[1]}, 选择后特征数: {X_train_sel.shape[1]}) print(f选中的特征: {X_train_sel.columns.tolist()}) # 2. 自动调参 print(\n 自动调参 ) X_train_sel2, X_test_sel2, selector feature_selection( X, y, X_test, methodrf, auto_tuneTrue, return_selectorTrue ) print(f最佳参数: {selector.performance_metrics_[best_params]}) print(fCV R²: {selector.performance_metrics_[cv_mean_r2]:.4f} ± {selector.performance_metrics_[cv_std_r2]:.4f}) # 3. 查看特征重要性 print(\n 特征重要性 ) importance_df selector.get_feature_importance_df() print(importance_df.head(10)) # 4. 获取摘要 print(\n 选择摘要 ) summary selector.get_summary() for key, value in summary.items(): if key not in [kept_features, performance]: print(f{key}: {value})使用建议场景推荐配置快速原型methodrf, auto_tuneFalse高维数据methodlasso, auto_tuneTrue追求精度methodrf, auto_tuneTrue, cv_folds10解释性需求methodlasso查看系二、为什么Lasso必须标准化问题演示import numpy as np import pandas as pd from sklearn.linear_model import Lasso from sklearn.preprocessing import StandardScaler # 创建数据特征尺度差异巨大 np.random.seed(42) X pd.DataFrame({ feature_A: np.random.randn(100) * 1000, # 大尺度 feature_B: np.random.randn(100) * 0.001 # 小尺度 }) y 2 * X[feature_A] 3 * X[feature_B] np.random.randn(100) * 0.1 # 未标准化时 lasso_unscaled Lasso(alpha0.1) lasso_unscaled.fit(X, y) print(未标准化系数:, lasso_unscaled.coef_) # 输出: [1.99, 0.00] → feature_B被错误剔除 # 标准化后 scaler StandardScaler() X_scaled scaler.fit_transform(X) lasso_scaled Lasso(alpha0.1) lasso_scaled.fit(X_scaled, y) print(标准化后系数:, lasso_scaled.coef_) # 输出: [0.82, 0.75] → 两个特征都保留系数比例正确核心原因python# Lasso的正则化项α × Σ|β_i| # 如果feature_A范围是[0, 1000]feature_B范围是[0, 1] # 为了预测yfeature_A的系数β_A会很小feature_B的系数β_B会很大 # 但正则化对所有系数平等惩罚 # → 系数大的β_B受惩罚更重被错误压缩为0 # 标准化后所有特征均值为0标准差为1 # → 系数大小直接反映重要性 # → 正则化公平对待所有特征三、随机森林不需要标准化的原因from sklearn.ensemble import RandomForestRegressor # 同一数据不标准化 rf_unscaled RandomForestRegressor(n_estimators100, random_state42) rf_unscaled.fit(X, y) print(未标准化重要性:, rf_unscaled.feature_importances_) # 输出: [0.51, 0.49] → 两个特征都保留 # RF的特征重要性计算 # 1. 每个节点选择分裂特征时只比较数值大小 # 2. 不涉及距离计算或梯度下降 # 3. 基于信息增益/基尼不纯度不受尺度影响 # 理论证明 # 特征A: [0, 1000]特征B: [0, 1] # 虽然数值范围不同但分裂时只看排序和分裂点 # 两个特征的区分能力不变 ✓四、是否需要标准化的完整矩阵方法需要标准化原因替代方案Lasso✅ 必须L1惩罚-Ridge✅ 必须L2惩罚-ElasticNet✅ 必须两种惩罚-逻辑回归✅ 必须梯度下降-SVM✅ 必须距离度量-KNN✅ 必须距离度量-神经网络✅ 建议梯度下降BatchNormPCA✅ 必须方差解释-随机森林❌ 不需要分裂规则-XGBoost❌ 不需要分裂规则-LightGBM❌ 不需要分裂规则-决策树❌ 不需要分裂规则-相关性分析❌ 不需要相关系数-卡方检验❌ 不需要频数统计-互信息❌ 不需要概率-方差阈值❌ 不需要原始方差可用标准化