# 06 · 模型评估、调参与端到端实战

> AI生成｜2026年10月26日｜codebuddy

**本篇目标**：把前五篇的技能串成一套能交付的工作流——**怎么划分数据不泄漏、怎么按业务选指标、怎么高效调超参、怎么诊断模型、怎么把模型存成可上线的产物**。最后用一个端到端项目把全部环节跑一遍。

**导航**：[栏目首页](./README.md) ｜ 上一篇：[05 · 神经网络与深度学习入门](./05-神经网络与深度学习入门.md)

---

## 一、核心概念

同一个模型、同一份数据，不同的评估方式能给出相差 20% 的分数。所以评估不是收尾工作，而是**整个流程里最容易被低估、也最容易翻车的一环**。

三条铁律：

1. **训练集/验证集/测试集职责分明**：参数从训练集学，超参数用验证集选，测试集只在最终结果里碰一次。
2. **指标要对齐业务**：没有一个万能指标，"准确率 92%"在没有上下文时毫无信息量。
3. **任何数据相关的变换都必须在交叉验证的每一折内部重新开始算**，否则就是数据泄漏。

---

## 二、方法与工具

### 2.1 数据怎么划分

| 策略 | API | 适用 |
| --- | --- | --- |
| 留出法 | `train_test_split(test_size=0.2, stratify=y)` | 样本量大、只想快速看结果 |
| K 折交叉验证 | `KFold` / `StratifiedKFold` | **默认选择**，分类任务一定用分层版 |
| 重复 K 折 | `RepeatedStratifiedKFold` | 样本少、想压随机性 |
| 留一法 | `LeaveOneOut` | 样本极小，代价是 n 次训练 |
| 时间序列划分 | `TimeSeriesSplit` | **顺序敏感**，不能随机打乱，只能用过去预测未来 |
| 分组划分 | `GroupKFold` | 同一用户/同一设备的样本必须在同一折，否则就是"背答案" |

**泄漏的三种常见形态**：

```text
1. 特征泄漏：特征里混进了只有事后才知道的信息（如"理赔金额"预测"是否骗保"）
2. 预处理泄漏：先在全量数据上 fit 了 StandardScaler / Imputer / 目标编码，再划分
   → 正确做法：放进 Pipeline，让它在 CV 的每一折里独立 fit
3. 样本泄漏：同一条样本（或同一用户的近重复样本）同时出现在 train 和 val
```

### 2.2 指标怎么选

| 场景 | 首选指标 | 备选/补充 |
| --- | --- | --- |
| 回归 | RMSE、MAE、R² | MAPE（相对误差，业务汇报友好） |
| 二分类（均衡） | accuracy、F1 | ROC-AUC（看排序能力） |
| 二分类（不平衡） | **PR-AUC（average_precision）**、recall | MCC、平衡准确率、KS |
| 多分类 | macro-F1（各类平等）或 weighted-F1 | 混淆矩阵、per-class recall |
| 概率要用 | Brier score、log loss、校准曲线 | `CalibratedClassifierCV` 校准 |
| 排序/推荐 | NDCG@K、MAP@K | Hit Rate |

**sklearn 里所有 scorer 都能用 `cross_validate(scoring=[...])` 一次算多个**，自定义业务指标用 `make_scorer`：

```python
from sklearn.metrics import make_scorer
def profit(y_true, y_pred):        # 举例：每挽留一个流失客户净赚 450
    from sklearn.metrics import confusion_matrix
    tn, fp, fn, tp = confusion_matrix(y_true, y_pred, labels=[0, 1]).ravel()
    return tp * 450 - fp * 50 - fn * 1000
profit_scorer = make_scorer(profit, greater_is_better=True)
```

### 2.3 调超参的四种武器

| 方法 | 适用 | 备注 |
| --- | --- | --- |
| `GridSearchCV` | 参数少于 3 个、候选值少 | 组合数指数爆炸，很容易算爆 |
| `RandomizedSearchCV` | **默认首选** | 同样预算下通常优于网格（大部分参数互相不敏感） |
| `HalvingGridSearchCV` | 训练代价高 | 先粗筛再精筛（连续减半） |
| Optuna / 贝叶斯优化 | 参数多、评估昂贵 | 用历史结果指导下一组采样 |

**嵌套交叉验证（Nested CV）**是唯一"诚实"的做法：内层 CV 负责选超参，外层 CV 负责估计泛化性能。反过来「在同一份 CV 上选完最优超参数、再把那个最好分数当成最终结果」会得到**乐观偏高的分数**。

### 2.4 四个诊断工具

| 工具 | 回答什么问题 |
| --- | --- |
| `learning_curve` | 该**加数据**还是该**换/简化模型**？（两条曲线是否收敛、间隙多大） |
| `validation_curve` | 这个超参数到底敏不敏感、最优点在哪 |
| `permutation_importance` | 哪个特征真正重要（在验证集上打乱列看分数掉多少） |
| `calibration_curve` | 输出的 0.8 是不是真的有 80% 的概率发生 |

### 2.5 上线与监控

```text
1. 用 Pipeline 包住全部预处理 + 模型 → joblib.dump(pipeline, "model.joblib")
   （千万不要只存模型：上线时把预处理弄丢是最低级的翻车）
2. 离线/在线的特征计算必须一致（特征平台 / Training-Serving skew）
3. 监控三件事：输入特征分布漂移（PSI）、预测分布变化、延迟与失败率
4. 效果监控要滞后指标（用真实标签回流计算）+ 代理指标（当日可用）
5. 重训节奏：数据漂移或效果跌破阈值时触发；上线先影子模式（shadow）再小流量 A/B
```

---

## 三、实例：端到端「客户流失预测」项目

完全离线构造一份带缺失值和类别特征的数据集，然后走一遍真实项目的全流程。

```python
import numpy as np
import pandas as pd
import joblib
from sklearn.datasets import make_classification
from sklearn.model_selection import (train_test_split, StratifiedKFold, cross_validate,
                                     cross_val_predict, RandomizedSearchCV, cross_val_score)
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.dummy import DummyClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, HistGradientBoostingClassifier
from sklearn.inspection import permutation_importance
from sklearn.metrics import (classification_report, confusion_matrix, roc_auc_score,
                             average_precision_score, precision_recall_curve)

rng = np.random.default_rng(42)

# ---------- 0) 造一个逼真的客户表 ----------
Xn, y = make_classification(
    n_samples=6000, n_features=10, n_informative=5, n_redundant=2,
    weights=[0.85, 0.15], flip_y=0.03, class_sep=0.8, random_state=42)
num_cols = ["tenure", "monthly_charges", "total_charges", "support_tickets",
            "login_days", "avg_session_min", "discount_rate", "nps_score",
            "payment_delay", "plan_level"]
df = pd.DataFrame(Xn, columns=num_cols)
df["contract_type"] = rng.choice(["monthly", "yearly", "two_year"], len(df))
df["payment_method"] = rng.choice(["card", "transfer", "echeck"], len(df))
df["region"] = rng.choice(["north", "south", "east", "west"], len(df))

# 注入缺失值（模拟真实数据），注意：nan 只注入到特征，标签不受影响
for col in ["total_charges", "nps_score", "payment_delay"]:
    idx = rng.choice(len(df), size=int(len(df) * 0.05), replace=False)
    df.loc[idx, col] = np.nan

cat_cols = ["contract_type", "payment_method", "region"]
X = df[num_cols + cat_cols]
print("缺失率 TOP3:\n", X.isna().mean().sort_values(ascending=False).head(3))
print("流失率 = {:.2%}".format(y.mean()))

# ---------- 1) 先划分，再考虑其他（永远是第一步） ----------
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, stratify=y, random_state=42)

# ---------- 2) 预处理放进 ColumnTransformer（防泄漏的关键） ----------
preprocess = ColumnTransformer([
    ("num", Pipeline([("impute", SimpleImputer(strategy="median")),
                      ("scale", StandardScaler())]), num_cols),
    ("cat", Pipeline([("impute", SimpleImputer(strategy="most_frequent")),
                      ("ohe", OneHotEncoder(handle_unknown="ignore"))]), cat_cols),
], sparse_threshold=0.0)      # 输出 dense，便于 HistGradientBoosting 使用

cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

# ---------- 3) 基线：什么都不做能到多少 ----------
for name, model in [("Dummy(most_frequent)", DummyClassifier(strategy="most_frequent")),
                    ("LogisticRegression", LogisticRegression(max_iter=2000,
                                                              class_weight="balanced"))]:
    pipe = Pipeline([("prep", preprocess), ("clf", model)])
    s = cross_validate(pipe, X_train, y_train, cv=cv,
                       scoring=["average_precision", "roc_auc", "f1"])
    print("{:22s} PR-AUC={:.4f}  ROC-AUC={:.4f}  F1={:.4f}".format(
        name, s["test_average_precision"].mean(), s["test_roc_auc"].mean(), s["test_f1"].mean()))

# ---------- 4) 多模型对比 ----------
for name, model in [("RandomForest", RandomForestClassifier(n_estimators=400, random_state=42, n_jobs=-1)),
                    ("HistGradientBoosting", HistGradientBoostingClassifier(random_state=42))]:
    pipe = Pipeline([("prep", preprocess), ("clf", model)])
    s = cross_validate(pipe, X_train, y_train, cv=cv,
                       scoring=["average_precision", "roc_auc", "f1"], n_jobs=-1)
    print("{:22s} PR-AUC={:.4f}  ROC-AUC={:.4f}  F1={:.4f}".format(
        name, s["test_average_precision"].mean(), s["test_roc_auc"].mean(), s["test_f1"].mean()))

# ---------- 5) 随机搜索调参 ----------
pipe = Pipeline([("prep", preprocess),
                 ("clf", HistGradientBoostingClassifier(random_state=42))])
space = {
    "clf__learning_rate":     [0.03, 0.05, 0.1, 0.2],
    "clf__max_leaf_nodes":    [15, 31, 63],
    "clf__min_samples_leaf":  [20, 40, 80],
    "clf__l2_regularization": [0.0, 1.0, 10.0],
    "clf__max_iter":          [200, 400],
}
search = RandomizedSearchCV(pipe, space, n_iter=30, cv=cv,
                            scoring="average_precision", random_state=42, n_jobs=-1)
search.fit(X_train, y_train)
print("\n最优参数:", search.best_params_)
print("内层 CV PR-AUC = {:.4f}".format(search.best_score_))

best = search.best_estimator_

# ---------- 6) 用折外预测挑阈值（收益导向，不是默认 0.5） ----------
oof = cross_val_predict(best, X_train, y_train, cv=cv, method="predict_proba")[:, 1]
GAIN, COST, LOSS = 450, 50, 1000          # 挽留成功净收益 / 打扰成本 / 流失损失

grid = np.linspace(0.05, 0.95, 91)
profits = []
for t in grid:
    tn, fp, fn, tp = confusion_matrix(y_train, (oof >= t).astype(int), labels=[0, 1]).ravel()
    profits.append(tp * GAIN - fp * COST - fn * LOSS)
best_thr = grid[int(np.argmax(profits))]
print("收益最优阈值 = {:.2f}，训练集期望收益 = {:,.0f}".format(best_thr, max(profits)))

# ---------- 7) 测试集只碰这一次 ----------
prob = best.predict_proba(X_test)[:, 1]
print("PR-AUC = {:.4f}   ROC-AUC = {:.4f}".format(
    average_precision_score(y_test, prob), roc_auc_score(y_test, prob)))
for tag, thr in [("0.50", 0.5), ("业务阈值", best_thr)]:
    y_pred = (prob >= thr).astype(int)
    tn, fp, fn, tp = confusion_matrix(y_test, y_pred, labels=[0, 1]).ravel()
    print("\n--- 阈值 {} ---".format(tag))
    print("混淆矩阵 TN={} FP={} FN={} TP={}   收益={:,.0f}".format(
        tn, fp, fn, tp, tp * GAIN - fp * COST - fn * LOSS))
    print(classification_report(y_test, y_pred, digits=3))

# ---------- 8) 谁在推动预测 ----------
r = permutation_importance(best, X_test, y_test, n_repeats=5,
                           random_state=42, scoring="average_precision", n_jobs=-1)
imp = pd.Series(r.importances_mean, index=X.columns).sort_values(ascending=False)
print("\nTop-6 特征重要性:\n", imp.head(6).round(4))

# ---------- 9) 嵌套交叉验证：诚实地报告泛化能力 ----------
inner = StratifiedKFold(n_splits=5, shuffle=True, random_state=1)
outer = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
nested_search = RandomizedSearchCV(pipe, space, n_iter=20, cv=inner,
                                   scoring="average_precision", random_state=42, n_jobs=-1)
nested = cross_val_score(nested_search, X_train, y_train, cv=outer,
                         scoring="average_precision", n_jobs=-1)
print("嵌套 CV PR-AUC = {:.4f} +/- {:.4f}".format(nested.mean(), nested.std()))

# ---------- 10) 持久化并模拟线上推理 ----------
joblib.dump({"model": best, "threshold": best_thr, "features": list(X.columns)}, "churn_model.joblib")
bundle = joblib.load("churn_model.joblib")
new_customer = pd.DataFrame([{
    "tenure": 3.0, "monthly_charges": 89.5, "total_charges": 268.0, "support_tickets": 4.0,
    "login_days": 2.0, "avg_session_min": 4.0, "discount_rate": 0.0, "nps_score": 2.0,
    "payment_delay": 12.0, "plan_level": 1.0,
    "contract_type": "monthly", "payment_method": "echeck", "region": "east",
}])[bundle["features"]]
p_churn = bundle["model"].predict_proba(new_customer)[0, 1]
print("流失概率 = {:.3f} → {}".format(p_churn, "需挽回" if p_churn >= bundle["threshold"] else "暂不干预"))
```

**这段代码里值得抄走的几点**：

- `ColumnTransformer` 把数值与类别两条流水线分开处理，缺失值填充和标准化**只对每一折的训练部分 fit**。
- `DummyClassifier` 是必须跑的第一步：它给出的分数就是"什么都不用做"的水平。
- 阈值是用**训练集的折外预测**挑出来的，绝不能在测试集上挑。
- 最终报告用**嵌套 CV** 的分数，而不是"CV 调完参数拿到的那个最好分数"。
- 存的是**整个 pipeline + 阈值 + 特征顺序**，上线时能把 mis-align 的概率降到 0。

---

## 四、上线前的自检清单

```text
□ 是否先划分数据，再做任何 fit / fit_transform？
□ 类别 features 是否需要 stratify？时间序列是否用了 TimeSeriesSplit？同用户是否用了 GroupKFold？
□ 预处理是否全部放进 Pipeline（含 Imputer、Scaler、编码、降维、目标编码）？
□ 指标是否与业务目标一致（而非默认 accuracy）？
□ 阈值是否单独调优过？
□ 是否报告了 nested CV 的分数，并在多个随机种子上验证稳定性？
□ 测试集是否只被使用了一次？
□ 保存的是完整 Pipeline 吗？输入列的顺序能保证一致吗？
□ 是否规划了线上监控指标（特征漂移 PSI、预测分布、真实标签回流）与重训触发条件？
```

---

## 五、小结与系列总览

**本篇要点**：划分策略决定评估可信度 → 指标要对齐业务 → 随机搜索 + 嵌套 CV 是性价比最高的调参组合 → 用四种诊断工具定位问题 → 存整个 Pipeline 并在上线后持续监控。

**给不同任务的第一选择**（把整个系列收成一张表）：

| 你的任务 | 先试这个 | 然后升级到 |
| --- | --- | --- |
| 二分类，想先看清对错 | 逻辑回归（要概率） | LightGBM / XGBoost |
| 表格数据，追求分数 | 随机森林（基线） | GBDT + 随机搜索 + 特征工程 |
| 高维文本分类 | TF-IDF + LinearSVC / ComplementNB | 微调预训练语言模型 |
| 图像 | 预训练 CNN（ResNet、ConvNeXt） | ViT / 检测分割模型 |
| 没有标签要分群 | K-Means（先 PCA） | DBSCAN / HDBSCAN / GMM |
| 数据太多要压缩/可视化 | PCA | UMAP |
| 序列里有时间顺序 | TimeSeriesSplit + 树模型 | Transformer / 时序大模型 |

**再往前一步可以学什么**：特征存储与 Training-Serving 一致性、模型压缩与蒸馏、AutoML 与超参优化框架（Optuna）、模型卡与可解释性（SHAP、LIME）、以及大模型时代的微调与 RAG。

> 欢迎从任意一篇开始读，每篇都可以单独看懂；动手把每篇的「三、实例」跑一遍，胜过读十遍正文。