演習：モデルの評価と改善を繰り返そう

import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_validate, StratifiedKFold, train_test_split
from sklearn.metrics import confusion_matrix, classification_report, f1_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder

# データ生成（Step 3 と同様）
np.random.seed(42)
n = 5000
df = pd.DataFrame({
    'purchase_count': np.random.poisson(8, n),
    'avg_amount': np.random.lognormal(8, 0.8, n).astype(int),
    'days_inactive': np.random.exponential(25, n).astype(int),
    'page_views': np.random.poisson(30, n),
    'support_tickets': np.random.poisson(1.5, n),
    'membership_months': np.random.exponential(18, n).astype(int),
    'device_type': np.random.choice(['PC', 'スマホ', 'タブレット'], n, p=[0.3, 0.55, 0.15]),
    'membership_rank': np.random.choice(['ブロンズ', 'シルバー', 'ゴールド'], n, p=[0.5, 0.3, 0.2]),
})
churn_prob = 1 / (1 + np.exp(-(df['days_inactive'] - 35) / 12 + df['purchase_count'] * 0.08 - df['support_tickets'] * 0.15))
df['is_churned'] = (np.random.random(n) < churn_prob).astype(int)

# 基本特徴量のみでベースライン
basic_features = ['purchase_count', 'avg_amount', 'days_inactive', 'page_views', 'support_tickets']
X = df[basic_features]
y = df['is_churned']

X_temp, X_test, y_temp, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
X_train, X_val, y_train, y_val = train_test_split(X_temp, y_temp, test_size=0.25, random_state=42, stratify=y_temp)

# ベースライン: ロジスティック回帰
baseline = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler()),
    ('model', LogisticRegression(max_iter=1000, random_state=42)),
])

# 交差検証
cv = StratifiedKFold(5, shuffle=True, random_state=42)
cv_results = cross_validate(
    baseline, X_train, y_train, cv=cv,
    scoring=['f1', 'precision', 'recall'],
    return_train_score=True,
)

print("=== ベースライン評価 ===")
for metric in ['f1', 'precision', 'recall']:
    train_m = cv_results[f'train_{metric}'].mean()
    val_m = cv_results[f'test_{metric}'].mean()
    print(f"{metric}: 学習={train_m:.3f}, 検証={val_m:.3f}, 差={train_m-val_m:.3f}")

# テストデータでの評価
baseline.fit(X_train, y_train)
y_pred = baseline.predict(X_test)
print(f"\nテストF1: {f1_score(y_test, y_pred):.3f}")
print(f"\n{classification_report(y_test, y_pred, target_names=['継続', '離反'])}")

# 改善記録
improvement_log = [{'ステップ': 'ベースライン', 'F1': f1_score(y_test, y_pred)}]

import lightgbm as lgb_module

# 特徴量追加
df['purchase_per_view'] = df['purchase_count'] / df['page_views'].replace(0, 1)
df['support_per_purchase'] = df['support_tickets'] / df['purchase_count'].replace(0, 1)
df['log_amount'] = np.log1p(df['avg_amount'])
df['is_long_inactive'] = (df['days_inactive'] > 30).astype(int)
df['purchase_amount_interaction'] = df['purchase_count'] * df['avg_amount']

enhanced_features = basic_features + [
    'purchase_per_view', 'support_per_purchase', 'log_amount',
    'is_long_inactive', 'purchase_amount_interaction', 'membership_months',
]

X_enhanced = df[enhanced_features]
X_temp2, X_test2, y_temp2, y_test2 = train_test_split(X_enhanced, y, test_size=0.2, random_state=42, stratify=y)
X_train2, X_val2, y_train2, y_val2 = train_test_split(X_temp2, y_temp2, test_size=0.25, random_state=42, stratify=y_temp2)

# LightGBM で構築
enhanced_pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('model', lgb_module.LGBMClassifier(n_estimators=200, max_depth=6, learning_rate=0.1, random_state=42, verbose=-1)),
])

cv_results2 = cross_validate(
    enhanced_pipeline, X_train2, y_train2, cv=cv,
    scoring=['f1', 'precision', 'recall'],
    return_train_score=True,
)

print("=== 改善モデル評価 ===")
for metric in ['f1', 'precision', 'recall']:
    train_m = cv_results2[f'train_{metric}'].mean()
    val_m = cv_results2[f'test_{metric}'].mean()
    print(f"{metric}: 学習={train_m:.3f}, 検証={val_m:.3f}")

enhanced_pipeline.fit(X_train2, y_train2)
y_pred2 = enhanced_pipeline.predict(X_test2)
f1_enhanced = f1_score(y_test2, y_pred2)
print(f"\nテストF1: {f1_enhanced:.3f}")

improvement_log.append({'ステップ': '特徴量改善+LightGBM', 'F1': f1_enhanced})
print(f"改善幅: +{f1_enhanced - improvement_log[0]['F1']:.3f}")

import optuna
from sklearn.metrics import precision_score, recall_score

optuna.logging.set_verbosity(optuna.logging.WARNING)

def objective(trial):
    params = {
        'n_estimators': trial.suggest_int('n_estimators', 100, 500),
        'max_depth': trial.suggest_int('max_depth', 3, 12),
        'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.2, log=True),
        'num_leaves': trial.suggest_int('num_leaves', 10, 80),
        'subsample': trial.suggest_float('subsample', 0.6, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
        'min_child_samples': trial.suggest_int('min_child_samples', 5, 30),
    }

    model = Pipeline([
        ('imputer', SimpleImputer(strategy='median')),
        ('model', lgb_module.LGBMClassifier(**params, random_state=42, verbose=-1)),
    ])

    scores = cross_val_score(model, X_train2, y_train2, cv=cv, scoring='f1')
    return scores.mean()

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=30)

print(f"最適パラメータ: {study.best_params}")
print(f"最適CV F1: {study.best_value:.3f}")

# 最適パラメータでモデル構築
best_model = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('model', lgb_module.LGBMClassifier(**study.best_params, random_state=42, verbose=-1)),
])
best_model.fit(X_train2, y_train2)

# 閾値最適化
y_prob = best_model.predict_proba(X_test2)[:, 1]

best_f1 = 0
best_threshold = 0.5
for t in np.arange(0.1, 0.8, 0.02):
    y_pred_t = (y_prob >= t).astype(int)
    f1_t = f1_score(y_test2, y_pred_t)
    if f1_t > best_f1:
        best_f1 = f1_t
        best_threshold = t

print(f"\n最適閾値: {best_threshold:.2f}")
print(f"最終F1スコア: {best_f1:.3f}")

y_final = (y_prob >= best_threshold).astype(int)
print(f"\n{classification_report(y_test2, y_final, target_names=['継続', '離反'])}")

improvement_log.append({'ステップ': 'Optuna最適化+閾値調整', 'F1': best_f1})

# 改善履歴の表示
print("\n=== 改善履歴 ===")
log_df = pd.DataFrame(improvement_log)
for _, row in log_df.iterrows():
    print(f"{row['ステップ']:25s}: F1 = {row['F1']:.3f}")
print(f"\n総改善幅: +{improvement_log[-1]['F1'] - improvement_log[0]['F1']:.3f}")