scikit-learn Pipeline 實戰：前處理、交叉驗證與避免資料洩漏

Python 學習 - 本文屬於一個選集。

§ 114: 本文

機器學習 Demo 最常見的樣子，是先把整份資料補值、標準化、轉成 one-hot，最後才切 train/test。

程式跑得動，分數甚至漂亮得令人感動——然後上線就摔得很有戲劇張力。

問題通常不是模型不夠強，而是前處理偷看了驗證資料。今天拍拍君要用 scikit-learn 的 Pipeline 與 ColumnTransformer，把這條容易漏水的流程封成一個可以安全訓練、交叉驗證、調參與部署的物件。

一. Pipeline 解決的不是少寫幾行程式
#

假設資料同時有：

數值欄位：年齡、消費金額、近 30 天造訪次數
類別欄位：城市、註冊管道
缺值：使用者沒有填年齡，或來源管道遺失
預測目標：下個月是否續訂

我們需要對數值欄補值與標準化，對類別欄補值與 one-hot encoding，再把結果交給分類器。

如果手動分開呼叫，每次訓練、驗證與推論都要記得完全相同的順序。任何一步漏掉，模型看到的特徵空間就可能不同。

Pipeline 把「依序處理」變成一個 estimator；ColumnTransformer 則負責讓不同欄位走不同分支：

原始 DataFrame
├── 數值欄位 → 補中位數 → StandardScaler ┐
└── 類別欄位 → 補眾數   → OneHotEncoder ├→ LogisticRegression

真正重要的效果是：每一折 cross-validation 都只會用該折的 training subset 去 fit 前處理器。

二. 安裝與建立專案
#

用 uv 建一個小專案：

uv init sklearn-pipeline-demo
cd sklearn-pipeline-demo
uv add scikit-learn pandas joblib

或用既有虛擬環境：

python -m pip install scikit-learn pandas joblib

確認版本與匯入：

import sklearn

print(sklearn.__version__)

本文使用的是穩定公開 API，不需要碰 private method。若團隊會保存模型，仍應把 Python 與套件版本鎖進 lockfile。

三. 準備一份混合型別資料
#

先用合成資料建立可重跑的範例，不必下載外部資料集：

import numpy as np
import pandas as pd
from sklearn.datasets import make_classification

raw, target = make_classification(
    n_samples=240,
    n_features=4,
    n_informative=3,
    n_redundant=0,
    weights=[0.58, 0.42],
    class_sep=0.9,
    random_state=42,
)

rng = np.random.default_rng(42)

customers = pd.DataFrame(
    {
        "age": np.clip(np.rint(42 + raw[:, 0] * 12), 18, 80),
        "annual_spend": np.round(2400 + raw[:, 1] * 750, 2),
        "visits_30d": np.clip(np.rint(8 + raw[:, 2] * 3), 0, None),
        "days_since_last": np.clip(np.rint(20 + raw[:, 3] * 8), 0, None),
        "city": rng.choice(["Taipei", "Taichung", "Kaohsiung"], 240),
        "channel": rng.choice(["organic", "ads", "referral"], 240),
    }
)

# 刻意加入缺值，模擬真實表單資料。
customers.loc[rng.choice(240, 18, replace=False), "age"] = np.nan
customers.loc[rng.choice(240, 14, replace=False), "channel"] = None

labels = pd.Series(target, name="will_renew")

先把欄位角色明確列出，不要讓 dtype 猜測偷偷決定商業語意：

numeric_features = [
    "age",
    "annual_spend",
    "visits_30d",
    "days_since_last",
]

categorical_features = ["city", "channel"]

這份名單也是一個輕量的 input contract。推論資料少了必要欄位時，應在進模型前就失敗，而不是默默產生奇怪預測。

四. 先看資料洩漏怎麼發生
#

下面是常見但危險的寫法：

from sklearn.impute import SimpleImputer
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import StandardScaler

imputer = SimpleImputer(strategy="median")
scaler = StandardScaler()

# 錯誤：兩個 fit 都看過全部樣本。
numeric_ready = imputer.fit_transform(customers[numeric_features])
numeric_ready = scaler.fit_transform(numeric_ready)

# 之後才做 cross-validation，已經太晚了。
scores = cross_val_score(model, numeric_ready, labels, cv=5)

median、平均值、標準差、類別集合與特徵選擇結果，都是從資料學到的統計量。

只要在切分之前對整份資料 fit_transform()，validation fold 的資訊就已經滲進 training fold。模型雖然沒直接看到答案，卻看到了不該知道的分布。

安全規則很好記：

任何會從資料學習狀態的步驟，都要放進 Pipeline，再把整條 Pipeline 交給切分或交叉驗證工具。

五. 數值與類別各自建立小 Pipeline
#

數值欄位先補中位數，再標準化：

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

numeric_pipeline = Pipeline(
    steps=[
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", StandardScaler()),
    ]
)

類別欄位則補最常見值，再做 one-hot encoding：

categorical_pipeline = Pipeline(
    steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        (
            "onehot",
            OneHotEncoder(handle_unknown="ignore"),
        ),
    ]
)

handle_unknown="ignore" 很重要。訓練資料可能只有 Taipei、Taichung，正式環境卻第一次出現 Tainan；沒有這個設定，encoder 會直接丟出錯誤。

它不是叫模型「理解新城市」，而是把未知類別在這組 one-hot 欄位中編碼成全零。是否合理，仍要依產品語意監控。

六. 用 ColumnTransformer 合併兩條分支
#

把欄位清單與 transformer 綁在一起：

from sklearn.compose import ColumnTransformer

preprocessor = ColumnTransformer(
    transformers=[
        ("numeric", numeric_pipeline, numeric_features),
        ("categorical", categorical_pipeline, categorical_features),
    ],
    remainder="drop",
    verbose_feature_names_out=True,
)

每個 tuple 都是：

(步驟名稱, transformer, 要處理的欄位)

remainder="drop" 代表未列出的欄位會被丟棄。這比不小心把 user_id、活動後才產生的欄位，甚至 target proxy 一起送進模型安全得多。

若確定其餘欄位都可以原樣保留，也能用 remainder="passthrough"；但正式專案最好明確 allowlist。

七. 組成一個完整可訓練模型
#

最後加上分類器：

from sklearn.linear_model import LogisticRegression

model = Pipeline(
    steps=[
        ("preprocess", preprocessor),
        (
            "model",
            LogisticRegression(
                max_iter=1000,
                class_weight="balanced",
                random_state=42,
            ),
        ),
    ]
)

現在 model 接受的是原始 DataFrame，而不是手工加工過的矩陣。

切出最後保留的 test set：

from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    customers,
    labels,
    test_size=0.2,
    stratify=labels,
    random_state=42,
)

model.fit(X_train, y_train)

probability = model.predict_proba(X_test)[:, 1]
prediction = model.predict(X_test)

呼叫一次 fit()，scikit-learn 會依序完成補值、縮放、編碼與分類器訓練；呼叫 predict() 時則只做 transform()，不會重新學習統計量。

八. 正確做交叉驗證
#

保留 test set 後，在 training set 裡做 model selection：

from sklearn.model_selection import StratifiedKFold, cross_validate

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

result = cross_validate(
    model,
    X_train,
    y_train,
    cv=cv,
    scoring={
        "accuracy": "accuracy",
        "roc_auc": "roc_auc",
    },
    return_train_score=False,
)

print("accuracy:", result["test_accuracy"].mean())
print("roc_auc:", result["test_roc_auc"].mean())

每一折的順序都是：

clone 一份完整 Pipeline
只用該折 training subset 計算補值、縮放與類別
訓練分類器
transform 該折 validation subset
評分

這才是沒有偷看的 validation score。

如果資料有群組、時間順序或同一使用者的重複事件，StratifiedKFold 仍可能不適合。那是切分策略問題，應改用 GroupKFold、時間序列切分，或按使用者先分組。

九. 用巢狀參數名稱一起調前處理與模型
#

Pipeline 裡的參數使用雙底線 __ 往下走：

preprocess__numeric__imputer__strategy
└─ 外層步驟  └─ ColumnTransformer 分支 └─ 小 Pipeline 步驟 └─ 參數

用 GridSearchCV 同時比較補值策略與分類器正規化強度：

from sklearn.model_selection import GridSearchCV

search = GridSearchCV(
    estimator=model,
    param_grid={
        "preprocess__numeric__imputer__strategy": ["mean", "median"],
        "model__C": [0.1, 1.0, 10.0],
    },
    scoring="roc_auc",
    cv=cv,
    n_jobs=-1,
    refit=True,
)

search.fit(X_train, y_train)

print(search.best_params_)
print(search.best_score_)

注意：搜尋器收到的仍然是原始 X_train。不要先在外面把資料 transform 完，再把固定矩陣送進 search。

refit=True 會用最佳參數在全部 X_train 上重新訓練。最後只用一次 untouched test set 做總驗收：

from sklearn.metrics import classification_report, roc_auc_score

best_model = search.best_estimator_
test_probability = best_model.predict_proba(X_test)[:, 1]
test_prediction = best_model.predict(X_test)

print("test roc_auc:", roc_auc_score(y_test, test_probability))
print(classification_report(y_test, test_prediction))

test set 不是每天偷看一次的排行榜。若看完 test 結果又回頭調參，它就逐漸變成另一份 validation set。

十. 看懂轉換後的特徵名稱
#

Pipeline 不代表模型變成黑盒子。fit 完可以取出已訓練步驟：

fitted_preprocessor = best_model.named_steps["preprocess"]
feature_names = fitted_preprocessor.get_feature_names_out()

print(feature_names)

搭配 logistic regression 係數做基本檢查：

coefficients = pd.Series(
    best_model.named_steps["model"].coef_[0],
    index=feature_names,
).sort_values()

print(coefficients.head())
print(coefficients.tail())

這不是因果解釋，但能幫你抓出明顯異常：例如 ID 被當成數值、某個活動後欄位權重離譜，或類別名稱跟預期不同。

也可以先查看所有可調參數：

for name in sorted(model.get_params()):
    if "imputer" in name or name.endswith("__C"):
        print(name)

猜不到巢狀名稱時，用 get_params() 比盯著錯誤訊息發呆有效率。

十一. 儲存的是整條流程，不只是分類器
#

正式推論最怕訓練端與服務端各自維護一份前處理。請保存完整 Pipeline：

from pathlib import Path
import joblib

artifact_path = Path("artifacts/renewal_pipeline.joblib")
artifact_path.parent.mkdir(parents=True, exist_ok=True)

joblib.dump(best_model, artifact_path)

載入後直接餵原始欄位：

loaded_model = joblib.load(artifact_path)

new_customers = pd.DataFrame(
    [
        {
            "age": 35,
            "annual_spend": 3200.0,
            "visits_30d": 9,
            "days_since_last": 4,
            "city": "Tainan",
            "channel": "organic",
        }
    ]
)

renewal_probability = loaded_model.predict_proba(new_customers)[0, 1]
print(f"續訂機率：{renewal_probability:.1%}")