Day 051 — 交差検証の基礎（K-Fold CV・Stratified K-Fold）

📚 背景知識（読んでから問題へ）

モデルを評価するとき、データを1回だけ「訓練用・テスト用」に分けると 運に左右される評価 になります。たまたまテストデータが簡単なデータなら高スコアに、難しければ低スコアになるだけかもしれません。

交差検証（Cross-Validation）の核心

データを複数のブロック（fold）に分けて、何度も訓練・評価を繰り返す手法。全データが一度ずつテストに使われるため、より信頼性の高い評価が得られます。

🗂️ K-Fold の仕組み（K=5）

1000件のデータを5等分（各200件）。1回ずつfoldを変えてテストに使う。

🗂️ データスキーマ（合成分類データ 1000件）

📝 問題

📋 コピーimport pandas as pd
import numpy as np
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import KFold, StratifiedKFold, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import accuracy_score

np.random.seed(42)
X, y = make_classification(
    n_samples=1000,
    n_features=10,
    n_informative=5,
    n_redundant=2,
    weights=[0.8, 0.2],   # クラス比率 8:2（不均衡）
    random_state=42
)

X = pd.DataFrame(X, columns=[f"feat_{i}" for i in range(10)])
y = pd.Series(y, name="label")

問1 — K-Fold CVの手動実装

• KFold(n_splits=5, shuffle=True, random_state=42) を使って手動でCVを実装する
• 各foldで LogisticRegression を学習し accuracy_score で評価する
• 5つのスコアと、その 平均 ± 標準偏差 を表示する

問2 — Stratified K-Fold との比較

• StratifiedKFold(n_splits=5, shuffle=True, random_state=42) で同じ評価を実施する
• 各foldのクラス1の比率（y[test_idx].mean()）を表示する
• K-Fold と Stratified K-Fold でスコアの 標準偏差 を比較し、どちらが安定しているか確認する

問3 — cross_val_score を使った簡潔な実装

• cross_val_score 関数を使って問1・問2と同じ評価を1行で書く
• scoring="accuracy" と scoring="roc_auc" の両方で評価する
• n_jobs=-1（並列処理）を使って実行する

問4 — データリーク防止: Pipeline との組み合わせ

以下の悪い例（データリーク）と良い例（Pipeline）を実装し、スコアを比較する：

📋 コピー# 悪い例（データリーク）
scaler = StandardScaler()
X_scaled_leak = scaler.fit_transform(X)  # CV の外でスケーリング！
scores_leak = cross_val_score(LogisticRegression(), X_scaled_leak, y, cv=5)

# 良い例（Pipeline）
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression())
])
scores_correct = cross_val_score(pipeline, X, y, cv=StratifiedKFold(5, shuffle=True, random_state=42))

🔍 ヒント（段階的開示）

📋 コピーkf = KFold(n_splits=5, shuffle=True, random_state=42)
scores = []

for fold, (train_idx, test_idx) in enumerate(kf.split(X), 1):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]

    model = LogisticRegression(random_state=42, max_iter=500)
    model.fit(X_train, y_train)
    score = accuracy_score(y_test, model.predict(X_test))
    scores.append(score)

print(f"K-Fold CV: {np.mean(scores):.4f} ± {np.std(scores):.4f}")

📋 コピー# Stratified K-Fold でクラス比率を確認
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
for fold, (train_idx, test_idx) in enumerate(skf.split(X, y), 1):
    ratio = y.iloc[test_idx].mean()
    print(f"Fold {fold}: クラス1比率 = {ratio:.3f}")

# cross_val_score
cv_strategy = StratifiedKFold(5, shuffle=True, random_state=42)
scores_acc = cross_val_score(
    LogisticRegression(max_iter=500), X, y,
    cv=cv_strategy, scoring="accuracy", n_jobs=-1
)
scores_auc = cross_val_score(
    LogisticRegression(max_iter=500), X, y,
    cv=cv_strategy, scoring="roc_auc", n_jobs=-1
)
print(f"Accuracy: {np.mean(scores_acc):.4f} ± {np.std(scores_acc):.4f}")
print(f"ROC-AUC : {np.mean(scores_auc):.4f} ± {np.std(scores_auc):.4f}")

✅ 模範解答

📋 コピーimport pandas as pd
import numpy as np
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import KFold, StratifiedKFold, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import accuracy_score

np.random.seed(42)
X, y = make_classification(
    n_samples=1000, n_features=10, n_informative=5,
    n_redundant=2, weights=[0.8, 0.2], random_state=42
)
X = pd.DataFrame(X, columns=[f"feat_{i}" for i in range(10)])
y = pd.Series(y, name="label")

# ── 問1: K-Fold CV 手動実装 ──
kf = KFold(n_splits=5, shuffle=True, random_state=42)
scores_kf = []
for fold, (train_idx, test_idx) in enumerate(kf.split(X), 1):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
    model = LogisticRegression(random_state=42, max_iter=500)
    model.fit(X_train, y_train)
    score = accuracy_score(y_test, model.predict(X_test))
    scores_kf.append(score)
    print(f"Fold {fold}: {score:.4f}")
print(f"K-Fold 平均: {np.mean(scores_kf):.4f} ± {np.std(scores_kf):.4f}")

# ── 問2: Stratified K-Fold との比較 ──
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores_skf = []
for fold, (train_idx, test_idx) in enumerate(skf.split(X, y), 1):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
    ratio = y_test.mean()
    model = LogisticRegression(random_state=42, max_iter=500)
    model.fit(X_train, y_train)
    score = accuracy_score(y_test, model.predict(X_test))
    scores_skf.append(score)
    print(f"Fold {fold}: クラス1比率={ratio:.3f}, accuracy={score:.4f}")
print(f"Stratified 平均: {np.mean(scores_skf):.4f} ± {np.std(scores_skf):.4f}")
print(f"標準偏差比較 → K-Fold: {np.std(scores_kf):.4f} / Stratified: {np.std(scores_skf):.4f}")

# ── 問3: cross_val_score ──
cv_strategy = StratifiedKFold(5, shuffle=True, random_state=42)
model = LogisticRegression(random_state=42, max_iter=500)
scores_acc = cross_val_score(model, X, y, cv=cv_strategy, scoring="accuracy", n_jobs=-1)
scores_auc = cross_val_score(model, X, y, cv=cv_strategy, scoring="roc_auc", n_jobs=-1)
print(f"Accuracy: {np.mean(scores_acc):.4f} ± {np.std(scores_acc):.4f}")
print(f"ROC-AUC : {np.mean(scores_auc):.4f} ± {np.std(scores_auc):.4f}")

# ── 問4: Pipeline と データリーク ──
cv5 = StratifiedKFold(5, shuffle=True, random_state=42)
scaler = StandardScaler()
X_scaled_leak = scaler.fit_transform(X)
scores_leak = cross_val_score(
    LogisticRegression(max_iter=500), X_scaled_leak, y, cv=cv5, scoring="accuracy"
)
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression(max_iter=500))
])
scores_correct = cross_val_score(pipeline, X, y, cv=cv5, scoring="accuracy")
print(f"[リーク有] {np.mean(scores_leak):.4f} ± {np.std(scores_leak):.4f}")
print(f"[Pipeline] {np.mean(scores_correct):.4f} ± {np.std(scores_correct):.4f}")

📊 K-Fold vs Stratified K-Fold の比較

各手法のスコア安定性（概念図）

K値（分割数）とトレードオフ

🔗 データリーク vs Pipeline（可視化）

🪜 Step-by-Step 解説

1 KFold の基本的な使い方

📋 コピーfrom sklearn.model_selection import KFold

kf = KFold(
    n_splits=5,          # データを5等分
    shuffle=True,        # 分割前にシャッフル（必須）
    random_state=42      # 再現性のためにシードを固定
)

# split(X) は (訓練インデックスの配列, テストインデックスの配列) を返す
for train_idx, test_idx in kf.split(X):
    print(f"train: {len(train_idx)}, test: {len(test_idx)}")
    # → train: 800, test: 200 (5等分なので)

    X_train = X.iloc[train_idx]  # pandas では .iloc でインデックスを使う
    X_test  = X.iloc[test_idx]

shuffle=False（デフォルト）のままだとデータの並び順のまま分割されます。時系列以外では必ず shuffle=True を使いましょう。

2 StratifiedKFold でクラス比率を確認する

📋 コピーfrom sklearn.model_selection import StratifiedKFold

# 分類では split(X, y) に y を渡す（これが Stratified の鍵）
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

for fold, (train_idx, test_idx) in enumerate(skf.split(X, y), 1):
    # テストデータのクラス1の比率を確認
    class1_ratio = y.iloc[test_idx].mean()
    print(f"Fold {fold}: クラス1比率 = {class1_ratio:.3f}")
    # → 全fold で ≈ 0.200 になる（元データの比率を維持）

3 cross_val_score の便利な使い方

📋 コピーfrom sklearn.model_selection import cross_val_score
import numpy as np

# scoring 文字列一覧（よく使うもの）
# "accuracy"              → 正解率
# "roc_auc"               → ROC-AUC（分類）
# "neg_mean_squared_error" → MSE（符号反転、回帰）
# "neg_root_mean_squared_error" → RMSE（符号反転、回帰）
# "r2"                    → R²（回帰）

scores = cross_val_score(
    LogisticRegression(max_iter=500),
    X, y,
    cv=StratifiedKFold(5, shuffle=True, random_state=42),
    scoring="roc_auc",
    n_jobs=-1   # 全CPUコアで並列実行 → 大量データで高速化
)

print(f"mean: {scores.mean():.4f}")
print(f"std:  {scores.std():.4f}")
print(f"all:  {scores.round(4)}")

4 Pipeline でデータリークを確実に防ぐ

📋 コピーfrom sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

# Pipeline のステップは (名前, オブジェクト) のタプルのリスト
pipeline = Pipeline([
    ("scaler", StandardScaler()),       # Step 1: スケーリング
    ("model",  LogisticRegression())    # Step 2: 学習
])

# cross_val_score に渡すだけで OK
# 内部で各fold ごとに: scaler.fit_transform(X_train) → model.fit(X_train_scaled)
#                              scaler.transform(X_test)  → model.predict(X_test_scaled)
# が自動で行われる

scores = cross_val_score(pipeline, X, y, cv=5, scoring="accuracy")
print(f"Pipeline CV: {scores.mean():.4f} ± {scores.std():.4f}")

🔢 数学・統計の補足（文系向け）

「平均 ± 標準偏差」の読み方

K値のトレードオフ（直感的説明）

🏆 Kaggleでの実践的な使い方

📋 コピー# Kaggle 上位の OOF（Out-of-Fold）予測テンプレート
import numpy as np
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import roc_auc_score

def run_cv_with_oof(model, X, y, n_splits=5):
    skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=42)
    oof_preds = np.zeros(len(y))  # 全サンプルの予測を格納
    scores = []

    for fold, (train_idx, val_idx) in enumerate(skf.split(X, y), 1):
        X_tr, X_val = X.iloc[train_idx], X.iloc[val_idx]
        y_tr, y_val = y.iloc[train_idx], y.iloc[val_idx]

        model.fit(X_tr, y_tr)
        oof_preds[val_idx] = model.predict_proba(X_val)[:, 1]

        score = roc_auc_score(y_val, oof_preds[val_idx])
        scores.append(score)
        print(f"Fold {fold}: AUC = {score:.4f}")

    oof_auc = roc_auc_score(y, oof_preds)
    print(f"\nOOF AUC: {oof_auc:.4f} (CV: {np.mean(scores):.4f} ± {np.std(scores):.4f})")
    return oof_preds  # スタッキング・ブレンディングに使用

# 使用例
from sklearn.linear_model import LogisticRegression
oof = run_cv_with_oof(LogisticRegression(max_iter=500), X, y)

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: TimeSeriesSplit — 時系列データでは未来のデータが訓練に混入しないよう、fold を時間順に設計する必要があります。金融・需要予測コンペで必須の知識です
• 次回予告 (Day 052): ロジスティック回帰入門 — 「確率」として予測を出す仕組み。シグモイド関数・決定境界・係数の解釈を学びます

📋 自己評価（解いた後に記入）