数据预处理与特征工程从原始数据到模型输入AI工程的隐藏战场一、数据预处理的隐秘代价80%的时间在洗数据每个AI工程师都知道Garbage In, Garbage Out但很少有人意识到数据预处理占了整个项目80%的时间。一个标注数据集10%的标签是错的5%的样本是重复的3%的特征值缺失2%的数值超出合理范围。这些脏数据如果不清洗模型学到的就是噪声而非信号。更隐蔽的问题是特征工程的艺术性。哪些特征该保留哪些该丢弃哪些该组合哪些该变换——这些决策对模型性能的影响可能超过模型架构的选择。一个精心设计的特征可以让简单模型的性能超过复杂模型。但特征工程高度依赖领域知识难以自动化。数据预处理和特征工程是AI工程中最不起眼却最关键的环节。二、数据预处理与特征工程体系flowchart TD A[原始数据] -- B[数据清洗层] B -- B1[缺失值处理: 填充/删除/标记] B -- B2[异常值检测: 统计/规则/ML] B -- B3[重复值去除: 精确/模糊匹配] B1 -- C[数据变换层] B2 -- C B3 -- C C -- C1[数值标准化: Z-Score/Min-Max] C -- C2[类别编码: OneHot/Target/Embedding] C -- C3[文本向量化: TF-IDF/BERT] C1 -- D[特征工程层] C2 -- D C3 -- D D -- D1[特征构造: 交叉/多项式/统计] D -- D2[特征选择: 过滤/包裹/嵌入] D -- D3[特征降维: PCA/AutoEncoder]2.1 数据清洗管道# data_cleaning.py — 数据清洗管道 # 设计意图系统化处理缺失值、异常值和重复值 # 确保输入数据的质量 import numpy as np import pandas as pd from dataclasses import dataclass, field from typing import Optional, Union from enum import Enum class MissingStrategy(Enum): DROP drop # 删除缺失行 FILL_MEAN fill_mean # 均值填充 FILL_MEDIAN fill_median # 中位数填充 FILL_MODE fill_mode # 众数填充 FILL_CONSTANT fill_constant # 常数填充 FORWARD_FILL forward_fill # 前向填充 MARK mark # 标记缺失新增二值列 class OutlierStrategy(Enum): DROP drop # 删除异常行 CLIP clip # 截断到边界 REPLACE_MEDIAN replace_median # 替换为中位数 MARK mark # 标记异常 dataclass class CleaningConfig: missing_strategy: MissingStrategy MissingStrategy.FILL_MEDIAN fill_constant: float 0 outlier_strategy: OutlierStrategy OutlierStrategy.CLIP outlier_method: str iqr # iqr / zscore outlier_threshold: float 3.0 # Z-Score阈值或IQR倍数 dedup_columns: Optional[list[str]] None dedup_keep: str first # first / last dataclass class CleaningReport: original_rows: int 0 final_rows: int 0 missing_filled: int 0 missing_dropped: int 0 outliers_detected: int 0 outliers_handled: int 0 duplicates_removed: int 0 column_stats: dict field(default_factorydict) class DataCleaner: def __init__(self, config: CleaningConfig None): self.config config or CleaningConfig() self.report CleaningReport() def clean(self, df: pd.DataFrame) - pd.DataFrame: 执行完整清洗流程 self.report CleaningReport(original_rowslen(df)) # 第一步缺失值处理 df self._handle_missing(df) # 第二步异常值处理 df self._handle_outliers(df) # 第三步重复值去除 df self._handle_duplicates(df) self.report.final_rows len(df) return df def _handle_missing(self, df: pd.DataFrame) - pd.DataFrame: 缺失值处理 strategy self.config.missing_strategy total_missing df.isnull().sum().sum() self.report.missing_filled 0 self.report.missing_dropped 0 for col in df.columns: if df[col].isnull().sum() 0: continue if strategy MissingStrategy.DROP: before len(df) df df.dropna(subset[col]) self.report.missing_dropped before - len(df) elif strategy MissingStrategy.FILL_MEAN: if df[col].dtype in [np.float64, np.int64]: fill_val df[col].mean() df[col] df[col].fillna(fill_val) self.report.missing_filled df[col].isnull().sum() elif strategy MissingStrategy.FILL_MEDIAN: if df[col].dtype in [np.float64, np.int64]: fill_val df[col].median() df[col] df[col].fillna(fill_val) self.report.missing_filled 1 elif strategy MissingStrategy.FILL_MODE: mode_val df[col].mode().iloc[0] if not df[col].mode().empty else df[col] df[col].fillna(mode_val) self.report.missing_filled 1 elif strategy MissingStrategy.FILL_CONSTANT: df[col] df[col].fillna(self.config.fill_constant) self.report.missing_filled 1 elif strategy MissingStrategy.FORWARD_FILL: df[col] df[col].ffill() self.report.missing_filled 1 elif strategy MissingStrategy.MARK: df[f{col}_missing] df[col].isnull().astype(int) if df[col].dtype in [np.float64, np.int64]: df[col] df[col].fillna(df[col].median()) else: df[col] df[col].fillna(MISSING) self.report.missing_filled 1 return df def _handle_outliers(self, df: pd.DataFrame) - pd.DataFrame: 异常值处理 strategy self.config.outlier_strategy numeric_cols df.select_dtypes(include[np.number]).columns total_outliers 0 for col in numeric_cols: outlier_mask self._detect_outliers(df[col]) n_outliers outlier_mask.sum() total_outliers n_outliers if n_outliers 0: continue if strategy OutlierStrategy.DROP: df df[~outlier_mask] elif strategy OutlierStrategy.CLIP: lower, upper self._get_bounds(df[col]) df.loc[outlier_mask, col] df.loc[outlier_mask, col].clip( lower, upper ) elif strategy OutlierStrategy.REPLACE_MEDIAN: median df.loc[~outlier_mask, col].median() df.loc[outlier_mask, col] median elif strategy OutlierStrategy.MARK: df[f{col}_outlier] outlier_mask.astype(int) self.report.outliers_detected total_outliers self.report.outliers_handled total_outliers return df def _detect_outliers(self, series: pd.Series) - pd.Series: 检测异常值 if self.config.outlier_method iqr: q1 series.quantile(0.25) q3 series.quantile(0.75) iqr q3 - q1 lower q1 - self.config.outlier_threshold * iqr upper q3 self.config.outlier_threshold * iqr return (series lower) | (series upper) elif self.config.outlier_method zscore: z_scores (series - series.mean()) / series.std() return z_scores.abs() self.config.outlier_threshold return pd.Series(False, indexseries.index) def _get_bounds(self, series: pd.Series) - tuple: 获取截断边界 if self.config.outlier_method iqr: q1 series.quantile(0.25) q3 series.quantile(0.75) iqr q3 - q1 return ( q1 - self.config.outlier_threshold * iqr, q3 self.config.outlier_threshold * iqr, ) mean, std series.mean(), series.std() return ( mean - self.config.outlier_threshold * std, mean self.config.outlier_threshold * std, ) def _handle_duplicates(self, df: pd.DataFrame) - pd.DataFrame: 重复值去除 dedup_cols self.config.dedup_columns or df.columns.tolist() before len(df) df df.drop_duplicates( subsetdedup_cols, keepself.config.dedup_keep ) self.report.duplicates_removed before - len(df) return df2.2 特征工程管道# feature_engineering.py — 特征工程管道 # 设计意图系统化执行特征构造、编码和选择 # 最大化特征的信息量 import numpy as np import pandas as pd from dataclasses import dataclass, field from typing import Optional from enum import Enum class EncodeMethod(Enum): ONE_HOT one_hot # 独热编码 LABEL label # 标签编码 TARGET target # 目标编码 FREQUENCY frequency # 频率编码 BINARY binary # 二值编码 dataclass class FeatureConfig: # 数值特征 scale_method: str standard # standard / minmax / robust # 类别特征 encode_method: EncodeMethod EncodeMethod.ONE_HOT max_one_hot_categories: int 10 # 超过此数量不用OneHot # 特征选择 selection_method: str mutual_info # mutual_info / chi2 / l1 max_features: Optional[int] None # 特征构造 create_interactions: bool True create_polynomials: bool False polynomial_degree: int 2 class FeatureEngineer: def __init__(self, config: FeatureConfig None): self.config config or FeatureConfig() self.encoders: dict {} self.scalers: dict {} self.selected_features: list[str] [] def fit_transform( self, df: pd.DataFrame, target: Optional[pd.Series] None ) - pd.DataFrame: 拟合并转换特征 # 第一步数值特征标准化 df self._scale_numeric(df) # 第二步类别特征编码 df self._encode_categorical(df, target) # 第三步特征构造 df self._construct_features(df) # 第四步特征选择 if target is not None: df self._select_features(df, target) return df def transform(self, df: pd.DataFrame) - pd.DataFrame: 使用已拟合的参数转换新数据 # 数值特征用已拟合的scaler for col, scaler in self.scalers.items(): if col in df.columns: df[col] scaler.transform(df[[col]]) # 类别特征用已拟合的encoder for col, encoder in self.encoders.items(): if col in df.columns: if self.config.encode_method EncodeMethod.TARGET: df[col] df[col].map(encoder).fillna(encoder.get(__default__, 0)) elif self.config.encode_method EncodeMethod.FREQUENCY: df[col] df[col].map(encoder).fillna(0) # 只保留选中的特征 if self.selected_features: available [f for f in self.selected_features if f in df.columns] df df[available] return df def _scale_numeric(self, df: pd.DataFrame) - pd.DataFrame: 数值特征标准化 from sklearn.preprocessing import ( StandardScaler, MinMaxScaler, RobustScaler, ) numeric_cols df.select_dtypes(include[np.number]).columns for col in numeric_cols: if self.config.scale_method standard: scaler StandardScaler() elif self.config.scale_method minmax: scaler MinMaxScaler() elif self.config.scale_method robust: scaler RobustScaler() else: continue df[col] scaler.fit_transform(df[[col]]) self.scalers[col] scaler return df def _encode_categorical( self, df: pd.DataFrame, target: Optional[pd.Series] ) - pd.DataFrame: 类别特征编码 cat_cols df.select_dtypes(include[object, category]).columns for col in cat_cols: n_unique df[col].nunique() method self.config.encode_method # 高基数类别不用OneHot if method EncodeMethod.ONE_HOT and n_unique self.config.max_one_hot_categories: method EncodeMethod.TARGET if target is not None else EncodeMethod.FREQUENCY if method EncodeMethod.ONE_HOT: dummies pd.get_dummies(df[col], prefixcol, dtypeint) df pd.concat([df.drop(col, axis1), dummies], axis1) elif method EncodeMethod.LABEL: codes, uniques pd.factorize(df[col]) df[col] codes self.encoders[col] dict(zip(uniques, range(len(uniques)))) elif method EncodeMethod.TARGET and target is not None: means df.groupby(col)[target.name].mean() means_dict means.to_dict() means_dict[__default__] target.mean() df[col] df[col].map(means_dict).fillna(target.mean()) self.encoders[col] means_dict elif method EncodeMethod.FREQUENCY: freq df[col].value_counts(normalizeTrue) freq_dict freq.to_dict() df[col] df[col].map(freq_dict).fillna(0) self.encoders[col] freq_dict return df def _construct_features(self, df: pd.DataFrame) - pd.DataFrame: 特征构造 if self.config.create_interactions: # 数值特征两两交叉 numeric_cols df.select_dtypes(include[np.number]).columns.tolist() if len(numeric_cols) 10: # 避免特征爆炸 for i in range(len(numeric_cols)): for j in range(i 1, len(numeric_cols)): col_a numeric_cols[i] col_b numeric_cols[j] df[f{col_a}_x_{col_b}] ( df[col_a] * df[col_b] ) if self.config.create_polynomials: from sklearn.preprocessing import PolynomialFeatures numeric_cols df.select_dtypes(include[np.number]).columns if len(numeric_cols) 5: # 多项式特征增长极快 poly PolynomialFeatures( degreeself.config.polynomial_degree, include_biasFalse, interaction_onlyFalse, ) poly_features poly.fit_transform(df[numeric_cols]) poly_df pd.DataFrame( poly_features[:, len(numeric_cols):], columnspoly.get_feature_names_out()[len(numeric_cols):], indexdf.index, ) df pd.concat([df, poly_df], axis1) return df def _select_features( self, df: pd.DataFrame, target: pd.Series ) - pd.DataFrame: 特征选择 from sklearn.feature_selection import ( mutual_info_classif, SelectKBest, ) k self.config.max_features or min(50, df.shape[1]) selector SelectKBest( score_funcmutual_info_classif, kk ) selector.fit(df, target) self.selected_features df.columns[ selector.get_support() ].tolist() return df[self.selected_features]四、边界分析与架构权衡缺失值填充的偏差均值填充会降低特征方差中位数填充会改变分布形态。更严重的是缺失本身可能包含信息如用户故意不填收入字段简单填充会丢失这个信号。标记缺失新增二值列是更安全的做法但增加了特征维度。目标编码的过拟合目标编码用目标变量的均值替换类别值高基数类别容易过拟合某个类别只有2个样本均值波动大。需要加平滑如贝叶斯平均或交叉验证编码但增加了计算复杂度。特征交叉的维度爆炸10个数值特征两两交叉产生45个新特征20个产生190个。多项式特征更恐怖——5个特征3次多项式产生56个新特征。需要严格限制交叉数量或用特征选择过滤无效交叉。特征选择的信息泄露如果特征选择用了全量数据包括测试集选出的特征可能偏向测试集导致线上效果下降。特征选择必须在训练集上执行验证集和测试集只做transform。四、边界分析与架构权衡围绕“数据预处理与特征工程从原始数据到模型输入AI工程的隐藏战场”做生产级落地时不能只看主流程是否成立还要把失败路径提前纳入设计。第一类风险来自输入不稳定真实业务数据往往存在缺字段、格式漂移和异常峰值如果缺少校验层后续模块会把脏数据放大成排障成本。第二类风险来自系统复杂度过多自动化能力会提高维护门槛团队需要明确哪些逻辑可以自动决策哪些节点必须保留人工确认。性能与可靠性也存在取舍。缓存、并行和批处理能提升吞吐但会引入一致性、重试风暴和资源抢占问题。更稳妥的做法是先定义可观测指标再逐步放开优化开关。每个优化项都应配套回滚条件例如错误率超过阈值、延迟超过基线或资源占用持续升高时系统可以退回到保守策略。这样即使收益不如预期也不会把风险扩散到整条链路。五、总结数据预处理与特征工程通过清洗、变换、构造和选择四层管道将原始数据转化为模型可用的优质特征。清洗处理缺失值、异常值和重复值变换标准化数值和编码类别构造交叉和多项式特征增加信息量选择过滤低价值特征降低维度。但填充偏差、目标编码过拟合、维度爆炸和信息泄露是需要权衡的边界条件。落地建议缺失值优先用标记法高基数类别用目标编码平滑特征交叉限制在10个以内特征选择严格在训练集上执行。补充落地建议围绕“数据预处理与特征工程从原始数据到模型输入AI工程的隐藏战场”继续推进时应把验证标准写成可执行清单而不是停留在经验判断。性能类方案要给出基准数据架构类方案要给出故障隔离方式AI 类方案要给出输出质量和人工兜底策略。每一次迭代都应回答三个问题收益是否可量化失败是否可回滚维护成本是否被团队接受。如果短期资源有限可以先保留最关键的观测指标包括处理耗时、失败率、资源占用和人工介入次数。等这些指标稳定后再扩展自动化能力。这样的节奏更慢但风险更低也更符合生产级技术文章强调的工程可验证性。