【保存版】特徴量エンジニアリング入門：設計→実装→リーク対策→評価→自動化まで“精度が上がる”実務の型

予測日：YYYY-MM-DD
学習窓：予測日の過去N日（例：90日）
評価窓：直近L日（例：30日）
ターゲット：評価窓内の離反=1

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score, StratifiedKFold

num\_cols = \["qty","price","days\_since\_last","amount"]
cat\_cols = \["store","category","area"]

num\_pipe = Pipeline(\[
("impute", SimpleImputer(strategy="median")),
("scale", StandardScaler()),
])
cat\_pipe = Pipeline(\[
("impute", SimpleImputer(strategy="most\_frequent")),
("onehot", OneHotEncoder(handle\_unknown="ignore", min\_frequency=5))
])

preprocess = ColumnTransformer(\[
("num", num\_pipe, num\_cols),
("cat", cat\_pipe, cat\_cols)
])

clf = Pipeline(\[
("prep", preprocess),
("model", LogisticRegression(max\_iter=1000, n\_jobs=-1))
])

cv = StratifiedKFold(n\_splits=5, shuffle=True, random\_state=42)

# PR-AUC推奨（不均衡向け）

scores = cross\_val\_score(clf, X, y, scoring="average\_precision", cv=cv, n\_jobs=-1)
print(scores.mean(), scores.std())

df["date"] = pd.to_datetime(df["date"])
df["ym"] = df["date"].dt.to_period("M").astype(str)
df["dow"] = df["date"].dt.dayofweek

df = df.sort\_values("date")
df\["amt"] = df\["qty"]\*df\["price"]

# ラグ/移動平均（未来参照はNG）

df\["amt\_lag1"] = df.groupby("store")\["amt"].shift(1)
df\["amt\_ma4"] = df.groupby("store")\["amt"].rolling(4).mean().reset\_index(level=0, drop=True)

# 季節性

df\["month"] = df\["date"].dt.month

# 直近90日の顧客別集約例
win = df.groupby(["customer"]).rolling("90D", on="date")["amt"].agg(["sum","mean","count"]).reset_index()
win.columns = ["customer","date","amt_sum90","amt_mean90","txn90"]
df = df.merge(win, on=["customer","date"], how="left")

import pandas as pd
from sklearn.model_selection import KFold

def kfold\_target\_encoding(X\_col, y, n\_splits=5, seed=42):
X\_col = pd.Series(X\_col)
kf = KFold(n\_splits=n\_splits, shuffle=True, random\_state=seed)
enc = pd.Series(index=X\_col.index, dtype=float)
for tr, va in kf.split(X\_col):
m = y.iloc\[tr].groupby(X\_col.iloc\[tr]).mean()
enc.iloc\[va] = X\_col.iloc\[va].map(m).fillna(y.iloc\[tr].mean())
return enc

X\["store\_te"] = kfold\_target\_encoding(X\["store"], y)

from sklearn.impute import SimpleImputer

# 欠損フラグを別に持つ

X\["qty\_isna"] = X\["qty"].isna().astype(int)

# 数値はmedian、カテゴリは最頻

imputer\_num = SimpleImputer(strategy="median")
imputer\_cat = SimpleImputer(strategy="most\_frequent")

X["unit_price"] = X["amount"]/X["qty"]
X["stock_gap"] = X["forecast_stock"] - X["actual_stock"]
X["price_ratio"] = X["price"] / X.groupby("category")["price"].transform("median")

from sklearn.inspection import permutation_importance
clf.fit(X_tr, y_tr)
r = permutation_importance(clf, X_va, y_va, n_repeats=5, random_state=42)

from sklearn.preprocessing import FunctionTransformer

def add\_date\_parts(d):
d = d.copy()
dt = pd.to\_datetime(d\["date"])
d\["month"], d\["dow"] = dt.dt.month, dt.dt.dayofweek
return d\[\["month","dow"]]

preprocess = ColumnTransformer(\[
("num", num\_pipe, num\_cols),
("cat", cat\_pipe, cat\_cols),
("date", FunctionTransformer(add\_date\_parts), \["date"])  # date列から2列生成
])

# tests/test_features.py
import pandas as pd
from src.features import make_features

def test\_no\_future\_leak():
df = pd.DataFrame({"date": pd.date\_range("2025-01-01", periods=10, freq="D"),
"amt": range(10)})
X = make\_features(df)
\# lagは必ずNaNを含む（未来参照なしの証拠）
assert X.filter(like="lag").isna().sum().sum() > 0