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

予測日：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()),
])


# カテゴリ: 欠損→最頻値、→One-Hot（未知カテゴリは無視、まれな値は束ねる）
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


# 入力: DataFrame（date列を含む）→ 出力: 追加する派生列だけ返す


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