sklearn-advanced

scikit-learn - 高级架构

要超越简单脚本，你必须掌握 Pipeline API。它允许你将整个预处理和建模流程视为一个对象，确保训练逻辑与生产环境中的推理逻辑完全一致。

适用场景

• 为异构数据构建复杂的特征工程流程。
• 创建可复用的自定义预处理步骤（例如，特定领域的数据清理）。
• 在不发生数据泄漏的情况下进行严格的超参数调优。
• 实现超越标准随机森林的集成方法。
• 监控并解读模型决策（部分依赖、置换重要性）。
• 为高性能生产环境导出模型。

参考文档

• Pipeline 指南：https://scikit-learn.org/stable/modules/compose.html
• 自定义估算器：https://scikit-learn.org/stable/developers/develop.html
• 模型评估：https://scikit-learn.org/stable/modules/model_evaluation.html
• 搜索模式：sklearn.base.BaseEstimator、sklearn.compose.make_column_selector、sklearn.model_selection.GridSearchCV

核心原则

一切皆对象

工作流中的每个步骤都应该是一个估算器。如果你发现自己在训练集和测试集之间手动执行 pandas 操作，就有发生数据泄漏的风险。

Pipeline 契约

Pipeline 确保 .fit() 只会在训练数据上调用，并且 .transform() 会一致地应用于训练集和测试集。

异构数据处理

使用 ColumnTransformer 并行地对数值、类别和文本数据应用不同的逻辑，然后自动合并结果。

快速参考

标准导入

import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import FunctionTransformer, StandardScaler, OneHotEncoder
from sklearn.model_selection import cross_validate, StratifiedKFold

基本模式 - 专业 Pipeline

# 1. Define Preprocessor
preprocessor = ColumnTransformer(
    transformers=[
        ('num', StandardScaler(), make_column_selector(dtype_include=np.number)),
        ('cat', OneHotEncoder(handle_unknown='ignore'), make_column_selector(dtype_include=object))
    ])

# 2. Create the Full Pipeline
clf = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier())
])

# 3. Fit and Tune (The entire pipeline is tuned together)
# Use 'classifier__' prefix to access parameters inside the pipeline
param_grid = {'classifier__n_estimators': [100, 200]}
grid = GridSearchCV(clf, param_grid, cv=5).fit(X_train, y_train)

关键规则

✅ 应该做

• 继承 BaseEstimator 和 TransformerMixin - 这样你就能免费获得 .fit_transform() 和 .get_params()。
• 使用 check_is_fitted - 在自定义转换器中，始终先确认模型已训练，再允许调用 .transform()。
• 设置 handle_unknown='ignore' - 在 OneHotEncoder 中，如果生产环境中出现新类别，这可以避免程序崩溃。
• 使用 TransformedTargetRegressor - 如果需要对目标变量（Y）进行对数变换，请使用它自动对预测结果执行逆变换。
• 优先使用 cross_validate，而不是 cross_val_score - 它支持多个指标，并会返回训练分数以检测过拟合。
• 设置 n_jobs=-1 - 在 GridSearch 和交叉验证期间最大化 CPU 利用率。

❌ 不应该做

• 不要对测试数据使用 fit_transform - 这是导致结果过于乐观的头号原因。
• 不需要时不要实现 fit - 对于无状态变换（例如对数变换），请使用 FunctionTransformer。
• 不要硬编码列名 - 使用 make_column_selector，让你的 Pipeline 能适应新增列。
• 不要忽略 Pipeline 索引 - 如果 Pipeline 运行失败，请使用 pipe.named_steps['step_name'] 检查内部状态。

自定义估算器开发

创建自定义特征选择器

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.utils.validation import check_is_fitted

class VarianceSelector(BaseEstimator, TransformerMixin):
    def __init__(self, threshold=0.01):
        self.threshold = threshold

    def fit(self, X, y=None):
        X = pd.DataFrame(X)
        self.variances_ = X.var()
        self.columns_to_keep_ = self.variances_[self.variances_ > self.threshold].index
        self.n_features_in_ = X.shape[1]
        return self

    def transform(self, X):
        check_is_fitted(self)
        X = pd.DataFrame(X)
        return X[self.columns_to_keep_]

高级预处理

目标编码（处理高基数类别）

from sklearn.preprocessing import TargetEncoder

# Efficiently encodes categories like 'City' or 'ZipCode' 
# based on the average target value, with internal cross-validation
encoder = TargetEncoder(smooth="auto")
X_encoded = encoder.fit_transform(X_cat, y)

堆叠与投票集成

from sklearn.ensemble import StackingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC

estimators = [
    ('rf', RandomForestClassifier()),
    ('svc', Pipeline([('scaler', StandardScaler()), ('svr', SVC())]))
]

# Use a meta-learner (LogisticRegression) to combine base model predictions
stack_clf = StackingClassifier(
    estimators=estimators, final_estimator=LogisticRegression()
)

模型评估与诊断

严格的交叉验证

from sklearn.model_selection import cross_validate

scoring = ['accuracy', 'precision_macro', 'recall_macro', 'f1_macro']
results = cross_validate(clf, X, y, cv=5, scoring=scoring, return_train_score=True)

print(f"Test F1: {results['test_f1_macro'].mean():.4f}")
print(f"Train F1: {results['train_f1_macro'].mean():.4f}") # Check for gap (overfitting)

校准曲线（确保概率准确）

from sklearn.calibration import CalibrationDisplay

# A well-calibrated model's predicted probability matches the actual frequency
CalibrationDisplay.from_estimator(clf, X_test, y_test, n_bins=10)

生产与持久化

使用 Joblib 处理大型模型

import joblib

# Save model
joblib.dump(clf, 'final_model.joblib', compress=3)

# Load model
loaded_model = joblib.load('final_model.joblib')

导出为 ONNX（高速推理）

# requires skl2onnx
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

initial_type = [('float_input', FloatTensorType([None, X.shape[1]]))]
onx = convert_sklearn(clf, initial_types=initial_type)
with open("model.onnx", "wb") as f:
    f.write(onx.SerializeToString())

实用工作流

1. 自动处理缺失数据和异常值

from sklearn.impute import KNNImputer
from sklearn.ensemble import IsolationForest

def build_robust_pipe():
    return Pipeline([
        ('imputer', KNNImputer(n_neighbors=5)),
        # FunctionTransformer for outlier removal is tricky because 
        # it changes row count. IsolationForest is better used for filtering.
        ('scaler', StandardScaler()),
        ('model', GradientBoostingClassifier())
    ])

2. 时间序列拆分（避免未来数据泄漏）

from sklearn.model_selection import TimeSeriesSplit

tscv = TimeSeriesSplit(n_splits=5)
# Use this cv object in GridSearchCV
grid = GridSearchCV(model, params, cv=tscv)

3. 特征联合（并行特征提取）

from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest

# Extract PCA features AND SelectKBest features in parallel
combined_features = FeatureUnion([
    ("pca", PCA(n_components=2)),
    ("univ_select", SelectKBest(k=5))
])

pipe = Pipeline([
    ("features", combined_features),
    ("clf", RandomForestClassifier())
])

性能优化

缓存 Pipeline 结果

如果预处理（例如 KNNImputer）速度很慢，而且你正在执行 GridSearch，请使用 memory 缓存转换器的输出。

from tempfile import mkdtemp
from shutil import rmtree

cachedir = mkdtemp()
pipe = Pipeline(steps=[...], memory=cachedir)
# Clean up after
# rmtree(cachedir)

常见陷阱与解决方案

“X 的 LabelEncoder”错误

LabelEncoder 仅适用于标签（y）。对于特征（X），始终使用 OrdinalEncoder 或 OneHotEncoder。

生产环境中的列不匹配

Pipeline 会存储训练时的列顺序。如果你在生产环境中传入列顺序不同的 DataFrame，可能会失败或产生错误结果。

# ✅ Solution: Ensure your pipeline is the first point of entry 
# for raw data, or use a custom transformer that sorts columns.

超参数调优期间的数据泄漏

在 Pipeline 中进行标准交叉验证是安全的。但如果你在 Pipeline 之前进行特征选择（如 SelectKBest），就会将整个数据集中的信息泄漏到模型中。

# ✅ Solution: Always include feature selection AS A STEP in the Pipeline.

高级 scikit-learn 的核心在于严谨。通过将所有数据转换都纳入 Pipeline/Transformer 架构，你可以创建不仅准确，而且稳健、易于维护并可用于真实场景部署的模型。