附录 B · 代码速查

pip install numpy scikit-learn          # 第 1-6 章
pip install torch                        # 第 7 章起
pip install lightgbm xgboost             # 可选，第 4 章

from sklearn.model_selection import (train_test_split, StratifiedKFold,
                                     KFold, GroupKFold, TimeSeriesSplit)

# 单次切分：分类务必加 stratify
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2,
                                          random_state=42, stratify=y)

cv = StratifiedKFold(5, shuffle=True, random_state=42)   # ⭐ 分类默认
cv = TimeSeriesSplit(n_splits=5)                          # ⭐ 时序必须
cv = GroupKFold(n_splits=5)                               # 同用户多条数据

from sklearn.pipeline import Pipeline, make_pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer

# ⭐ 所有会"看到数据统计量"的步骤都要放进来
pre = ColumnTransformer([
    ("num", Pipeline([("imp", SimpleImputer(strategy="median")),
                      ("sc",  StandardScaler())]), num_cols),
    ("cat", Pipeline([("imp", SimpleImputer(strategy="most_frequent")),
                      ("oh",  OneHotEncoder(handle_unknown="ignore"))]), cat_cols),
])
pipe = Pipeline([("pre", pre), ("model", LGBMClassifier())])

from sklearn.model_selection import cross_val_score
scores = cross_val_score(pipe, X, y, cv=cv, scoring="roc_auc")
print(f"{scores.mean():.4f} ± {scores.std():.4f}")     # ⭐ 永远带标准差

from sklearn.metrics import (accuracy_score, precision_score, recall_score,
                             f1_score, roc_auc_score, average_precision_score,
                             confusion_matrix, classification_report)

prob = model.predict_proba(X_te)[:, 1]
print(classification_report(y_te, model.predict(X_te)))
print("AUC   ", roc_auc_score(y_te, prob))
print("PR-AUC", average_precision_score(y_te, prob))    # ⭐ 不平衡时看这个

# 阈值扫描：找最佳 F1
import numpy as np
best = max(((f1_score(y_te, (prob >= t).astype(int)), t)
            for t in np.arange(0.05, 0.95, 0.01)))
print(f"最佳阈值 {best[1]:.2f}  F1={best[0]:.3f}")

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
import lightgbm as lgb

rf = RandomForestClassifier(n_estimators=300, n_jobs=-1, random_state=0)

# LightGBM 常用配置 + 早停
m = lgb.LGBMClassifier(
    n_estimators=2000, learning_rate=0.05, num_leaves=31,
    max_depth=-1, subsample=0.8, colsample_bytree=0.8,
    reg_alpha=0.0, reg_lambda=1.0, random_state=0)
m.fit(X_tr, y_tr, eval_set=[(X_va, y_va)], eval_metric="auc",
      callbacks=[lgb.early_stopping(100), lgb.log_evaluation(200)])

# ⭐ 比自带 importance 可靠
from sklearn.inspection import permutation_importance
r = permutation_importance(m, X_va, y_va, n_repeats=10, random_state=0)
for i in r.importances_mean.argsort()[::-1][:10]:
    print(f"{feature_names[i]:<25}{r.importances_mean[i]:.4f}")

import numpy as np, torch, torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader

def set_seed(s=42):
    import random
    random.seed(s); np.random.seed(s)
    torch.manual_seed(s); torch.cuda.manual_seed_all(s)

set_seed()
device = "cuda" if torch.cuda.is_available() else "cpu"

model = nn.Sequential(
    nn.Linear(30, 64), nn.ReLU(), nn.Dropout(0.2),
    nn.Linear(64, 32), nn.ReLU(),
    nn.Linear(32, 1),                      # ⭐ 二分类不加 Sigmoid
).to(device)

criterion = nn.BCEWithLogitsLoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=0.01)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=60)
loader = DataLoader(TensorDataset(Xtr, ytr), batch_size=32, shuffle=True,
                    num_workers=4, pin_memory=True)

# ===== 第0步：10样本过拟合检查（第11章）⭐ =====
tx, ty = Xtr[:10].to(device), ytr[:10].to(device)
for _ in range(300):
    optimizer.zero_grad(); loss = criterion(model(tx), ty)
    loss.backward(); optimizer.step()
assert loss.item() < 0.01, "❌ 10样本都过拟合不了，有bug，别往下走"

# ===== 正式训练 + 早停 =====
best, wait, patience, best_state = float("inf"), 0, 10, None
for ep in range(60):
    model.train(); total = 0.0
    for xb, yb in loader:
        xb, yb = xb.to(device), yb.to(device)
        optimizer.zero_grad()                                # ⭐
        loss = criterion(model(xb), yb)
        loss.backward()
        gn = torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0).item()
        optimizer.step(); total += loss.item()*len(xb)
    scheduler.step()

    model.eval()                                             # ⭐
    with torch.no_grad():                                    # ⭐
        vl = criterion(model(Xva.to(device)), yva.to(device)).item()
    print(f"ep {ep:<3} train {total/len(Xtr):.4f} | val {vl:.4f} "
          f"| lr {optimizer.param_groups[0]['lr']:.2e} | grad {gn:.2f}")

    if vl < best:
        best, wait = vl, 0
        best_state = {k: v.clone() for k, v in model.state_dict().items()}
    else:
        wait += 1
        if wait >= patience:
            model.load_state_dict(best_state); print(f"早停 ep{ep}"); break

def grad_check(f, param, analytic_grad, idx, eps=1e-5):
    """f: 只依赖 param 的损失函数；idx: 要检查的位置"""
    orig = param[idx]
    param[idx] = orig + eps; lp = f()
    param[idx] = orig - eps; lm = f()
    param[idx] = orig
    num = (lp - lm) / (2*eps)
    rel = abs(num - analytic_grad) / (abs(num) + abs(analytic_grad) + 1e-12)
    print(f"数值 {num:.8f} | 解析 {analytic_grad:.8f} | 相对误差 {rel:.2e} "
          f"{'✅' if rel < 1e-6 else '❌'}")

import torchvision.models as models, torch.nn as nn

model = models.resnet18(weights="IMAGENET1K_V1")
model.fc = nn.Linear(model.fc.in_features, num_classes)

# 数据少：冻结主干
for p in model.parameters():   p.requires_grad = False
for p in model.fc.parameters(): p.requires_grad = True

# 数据多：全微调，lr 小 10 倍
optimizer = torch.optim.AdamW([
    {"params": model.fc.parameters(),       "lr": 1e-3},
    {"params": [p for n,p in model.named_parameters() if not n.startswith("fc")],
     "lr": 1e-4},
])

import numpy as np, pandas as pd

df["log_amt"]   = np.log1p(df["amount"])                  # 长尾
df["age_bin"]   = pd.qcut(df["age"], 10, labels=False, duplicates="drop")
df["amt_missing"] = df["amount"].isna().astype(int)       # ⭐ 缺失本身是信号
df["hour_sin"]  = np.sin(2*np.pi*df["hour"]/24)           # ⭐ 周期性
df["hour_cos"]  = np.cos(2*np.pi*df["hour"]/24)

# 聚合统计（提分最猛）
agg = df.groupby("user_id")["amount"].agg(["count","mean","max","std"])
agg.columns = [f"user_amt_{c}" for c in agg.columns]
df = df.merge(agg, on="user_id", how="left")

# OOF 目标编码（防泄漏）
def target_encode_oof(df, col, target, n_splits=5, smooth=20):
    from sklearn.model_selection import KFold
    gm = df[target].mean(); oof = np.full(len(df), np.nan)
    for tr, va in KFold(n_splits, shuffle=True, random_state=0).split(df):
        s = df.iloc[tr].groupby(col)[target].agg(["mean","count"])
        sm = (s["mean"]*s["count"] + gm*smooth) / (s["count"]+smooth)
        oof[va] = df.iloc[va][col].map(sm).fillna(gm).values
    return oof

# 混合精度
import torch
scaler = torch.amp.GradScaler()
with torch.amp.autocast("cuda"):
    loss = criterion(model(xb), yb)
scaler.scale(loss).backward(); scaler.step(optimizer); scaler.update()

# 编译（PyTorch 2.0+）
model = torch.compile(model)

# 并行 CV
cross_val_score(pipe, X, y, cv=cv, n_jobs=-1)