【保存版】scikit-learn基礎：回帰・分類・前処理・パイプライン・交差検証を“実務の型”で習得

import numpy as np, pandas as pd
from sklearn.datasets import fetch_california_housing, load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.dummy import DummyRegressor, DummyClassifier
from sklearn.metrics import mean_squared_error, accuracy_score

# 回帰データ

d = fetch\_california\_housing(as\_frame=True)
Xr, yr = d.data, d.target
Xr\_train, Xr\_test, yr\_train, yr\_test = train\_test\_split(Xr, yr, test\_size=0.2, random\_state=42)
base\_r = DummyRegressor(strategy="median").fit(Xr\_train, yr\_train)
rmse\_base = mean\_squared\_error(yr\_test, base\_r.predict(Xr\_test), squared=False)
print(f"回帰ベースライン RMSE={rmse\_base:.3f}")

# 分類データ

d2 = load\_breast\_cancer(as\_frame=True)
Xc, yc = d2.data, d2.target
Xc\_train, Xc\_test, yc\_train, yc\_test = train\_test\_split(Xc, yc, test\_size=0.2, stratify=yc, random\_state=42)
base\_c = DummyClassifier(strategy="most\_frequent").fit(Xc\_train, yc\_train)
acc\_base = accuracy\_score(yc\_test, base\_c.predict(Xc\_test))
print(f"分類ベースライン ACC={acc\_base:.3f}")

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

num\_cols = Xr.columns  # 例：数値のみ

numeric\_proc = Pipeline(\[
("impute", SimpleImputer(strategy="median")),
("scale", StandardScaler())
])

# 例：カテゴリ列がある場合

# cat\_cols = \["gender", "area"]

# categorical\_proc = Pipeline(\[

# ("impute", SimpleImputer(strategy="most\_frequent")),

# ("onehot", OneHotEncoder(handle\_unknown="ignore", sparse\_output=False))

# ])

preprocess = ColumnTransformer(\[
("num", numeric\_proc, num\_cols),
\# ("cat", categorical\_proc, cat\_cols)
], remainder="drop")

from sklearn.linear_model import Ridge
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import r2_score, mean_squared_error
from sklearn.pipeline import Pipeline

ridge = Pipeline(\[
("prep", preprocess),
("model", Ridge(alpha=1.0, random\_state=42))
])
rf = Pipeline(\[
("prep", preprocess),
("model", RandomForestRegressor(n\_estimators=300, random\_state=42, n\_jobs=-1))
])

ridge.fit(Xr\_train, yr\_train)
rf.fit(Xr\_train, yr\_train)

for name, mdl in {"Ridge": ridge, "RF": rf}.items():
pred = mdl.predict(Xr\_test)
print(name, "RMSE=", mean\_squared\_error(yr\_test, pred, squared=False), "R2=", r2\_score(yr\_test, pred))

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score, classification_report, accuracy_score

pre\_num\_c = Pipeline(\[("impute", SimpleImputer()), ("scale", StandardScaler())])
prep\_c = ColumnTransformer(\[("num", pre\_num\_c, Xc.columns)])

logit = Pipeline(\[("prep", prep\_c), ("model", LogisticRegression(max\_iter=1000))])
rf\_c = Pipeline(\[("prep", prep\_c), ("model", RandomForestClassifier(n\_estimators=400, random\_state=42, n\_jobs=-1))])

for name, mdl in {"Logit": logit, "RF": rf\_c}.items():
mdl.fit(Xc\_train, yc\_train)
proba = mdl.predict\_proba(Xc\_test)\[:,1]
pred = (proba >= 0.5).astype(int)
print(name, "ACC=", accuracy\_score(yc\_test, pred), "AUC=", roc\_auc\_score(yc\_test, proba))

print(classification\_report(yc\_test, rf\_c.predict(Xc\_test)))

from sklearn.model_selection import cross_val_score, KFold, StratifiedKFold

cv\_r = KFold(n\_splits=5, shuffle=True, random\_state=42)
cv\_scores = cross\_val\_score(rf, Xr\_train, yr\_train, scoring="neg\_root\_mean\_squared\_error", cv=cv\_r, n\_jobs=-1)
print("CV RMSE(mean±std)=", -cv\_scores.mean(), "+/-", cv\_scores.std())

cv\_c = StratifiedKFold(n\_splits=5, shuffle=True, random\_state=42)
cv\_auc = cross\_val\_score(rf\_c, Xc\_train, yc\_train, scoring="roc\_auc", cv=cv\_c, n\_jobs=-1)
print("CV AUC(mean±std)=", cv\_auc.mean(), "+/-", cv\_auc.std())

from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from scipy.stats import randint

# 例：RF回帰（Grid）

param\_grid = {
"model\_\_n\_estimators": \[200, 400, 800],
"model\_\_max\_depth": \[None, 10, 20],
"model\_\_min\_samples\_split": \[2, 5, 10]
}
rf\_gs = GridSearchCV(rf, param\_grid, scoring="neg\_root\_mean\_squared\_error", cv=3, n\_jobs=-1)
rf\_gs.fit(Xr\_train, yr\_train)
print("best\_params=", rf\_gs.best\_params\_)

# 例：RF分類（Random）

param\_dist = {
"model\_\_n\_estimators": randint(100, 800),
"model\_\_max\_depth": \[None] + list(range(3, 15)),
}
rf\_rs = RandomizedSearchCV(rf\_c, param\_distributions=param\_dist, n\_iter=20, scoring="roc\_auc", cv=3, random\_state=42, n\_jobs=-1)
rf\_rs.fit(Xc\_train, yc\_train)
print("best\_params=", rf\_rs.best\_params\_)

from sklearn.inspection import permutation_importance

rf\_c.fit(Xc\_train, yc\_train)
r = permutation\_importance(rf\_c, Xc\_test, yc\_test, n\_repeats=10, random\_state=42, n\_jobs=-1)
imp = pd.Series(r.importances\_mean, index=Xc.columns).sort\_values(ascending=False)
print(imp.head(10))

import joblib
from sklearn.metrics import roc_auc_score

joblib.dump(rf\_c, "model\_rf\_class.joblib")
mdl = joblib.load("model\_rf\_class.joblib")
print("loaded AUC:", roc\_auc\_score(yc\_test, mdl.predict\_proba(Xc\_test)\[:,1]))

project/
  ├─ data/                 # 入出力（追跡対象外も検討）
  ├─ notebooks/            # 実験ノート
  ├─ src/                  # 前処理/学習の関数
  │   ├─ features.py
  │   ├─ models.py
  │   └─ evaluate.py
  ├─ reports/              # 図/スライド
  ├─ requirements.txt
  └─ README.md             # 再現手順

import numpy as np
from sklearn.metrics import roc_curve

proba = rf\_c.predict\_proba(Xc\_test)\[:,1]
fpr, tpr, thr = roc\_curve(yc\_test, proba)
youden = tpr - fpr
thr\_opt = thr\[np.argmax(youden)]
print("best threshold by Youden:", thr\_opt)

from sklearn.model_selection import TimeSeriesSplit, cross_val_score

ts = TimeSeriesSplit(n\_splits=5)
scores = cross\_val\_score(ridge, Xr\_train.sort\_index(), yr\_train.sort\_index(), cv=ts,
scoring="neg\_root\_mean\_squared\_error")
print("TimeSeries CV RMSE=", -scores.mean())