""" train 十字路分類モデルの学習・評価スクリプト 複数モデルを 5-fold CV で比較し，結果をドキュメントに出力する 最良モデルを全データで再学習して params/ に保存する 使い方: $ cd src && python -m pc.data.train """ import sys from datetime import datetime from pathlib import Path import joblib import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.linear_model import LogisticRegression from sklearn.metrics import ( classification_report, f1_score, ) from sklearn.model_selection import StratifiedKFold from sklearn.neural_network import MLPClassifier from sklearn.preprocessing import StandardScaler from sklearn.svm import SVC from common.json_utils import PARAMS_DIR from pc.data.dataset import load_dataset # ── 定数 ────────────────────────────────────────────── N_FOLDS: int = 5 RANDOM_STATE: int = 42 # モデル保存先 MODEL_PATH: Path = PARAMS_DIR / "intersection_model.pkl" SCALER_PATH: Path = PARAMS_DIR / "intersection_scaler.pkl" # ドキュメント出力先 _PROJECT_ROOT: Path = ( Path(__file__).resolve().parent.parent.parent.parent ) REPORT_PATH: Path = ( _PROJECT_ROOT / "docs" / "03_TECH" / "TECH_06_十字路分類モデル評価.txt" ) # ── モデル定義 ──────────────────────────────────────── def _build_models() -> list[tuple[str, object]]: """比較対象のモデル一覧を返す Returns: (名前, モデルインスタンス) のリスト """ return [ ( "LogisticRegression", LogisticRegression( max_iter=1000, random_state=RANDOM_STATE, ), ), ( "SVM_RBF", SVC( kernel="rbf", random_state=RANDOM_STATE, ), ), ( "SVM_Linear", SVC( kernel="linear", random_state=RANDOM_STATE, ), ), ( "RandomForest", RandomForestClassifier( n_estimators=100, random_state=RANDOM_STATE, ), ), ( "MLP_1layer", MLPClassifier( hidden_layer_sizes=(64,), max_iter=500, random_state=RANDOM_STATE, ), ), ( "MLP_2layer", MLPClassifier( hidden_layer_sizes=(128, 64), max_iter=500, random_state=RANDOM_STATE, ), ), ] # ── 評価 ────────────────────────────────────────────── def _evaluate_models( x: np.ndarray, y: np.ndarray, ) -> list[dict]: """全モデルを StratifiedKFold CV で評価する Args: x: 特徴量行列 y: ラベル配列 Returns: 各モデルの評価結果辞書のリスト """ skf = StratifiedKFold( n_splits=N_FOLDS, shuffle=True, random_state=RANDOM_STATE, ) results: list[dict] = [] for name, model in _build_models(): f1_scores: list[float] = [] for train_idx, test_idx in skf.split(x, y): x_train, x_test = x[train_idx], x[test_idx] y_train, y_test = y[train_idx], y[test_idx] # スケーリング scaler = StandardScaler() x_train_s = scaler.fit_transform(x_train) x_test_s = scaler.transform(x_test) model_clone = _clone_model(name) model_clone.fit(x_train_s, y_train) y_pred = model_clone.predict(x_test_s) f1 = f1_score(y_test, y_pred, average="macro") f1_scores.append(f1) results.append({ "name": name, "f1_mean": float(np.mean(f1_scores)), "f1_std": float(np.std(f1_scores)), "f1_scores": f1_scores, }) print( f" {name:25s} " f"F1={np.mean(f1_scores):.4f} " f"(±{np.std(f1_scores):.4f})" ) return results def _clone_model(name: str) -> object: """モデル名から新しいインスタンスを生成する Args: name: モデル名 Returns: モデルインスタンス """ for n, m in _build_models(): if n == name: return m raise ValueError(f"不明なモデル: {name}") # ── 最良モデルの再学習・保存 ────────────────────────── def _train_best_and_save( best_name: str, x: np.ndarray, y: np.ndarray, ) -> str: """最良モデルを全データで再学習して保存する Args: best_name: 最良モデルの名前 x: 全特徴量行列 y: 全ラベル配列 Returns: 全データでの classification_report 文字列 """ scaler = StandardScaler() x_scaled = scaler.fit_transform(x) model = _clone_model(best_name) model.fit(x_scaled, y) y_pred = model.predict(x_scaled) report = classification_report( y, y_pred, target_names=["normal", "intersection"], ) # 保存 PARAMS_DIR.mkdir(parents=True, exist_ok=True) joblib.dump(model, MODEL_PATH) joblib.dump(scaler, SCALER_PATH) return report # ── ドキュメント出力 ────────────────────────────────── def _write_report( results: list[dict], best_name: str, n_samples: int, n_intersection: int, n_normal: int, full_report: str, ) -> None: """評価結果をドキュメントとして出力する Args: results: 各モデルの評価結果 best_name: 最良モデルの名前 n_samples: 全サンプル数 n_intersection: intersection のサンプル数 n_normal: normal のサンプル数 full_report: 全データでの classification_report """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") # 結果テーブル sorted_results = sorted( results, key=lambda r: r["f1_mean"], reverse=True, ) table_lines: list[str] = [] table_lines.append( f" {'モデル':25s} {'F1(平均)':>10s}" f" {'F1(標準偏差)':>14s}" ) table_lines.append(f" {'─' * 55}") for r in sorted_results: marker = " ← 採用" if r["name"] == best_name else "" table_lines.append( f" {r['name']:25s}" f" {r['f1_mean']:10.4f}" f" {r['f1_std']:14.4f}{marker}" ) table_str = "\n".join(table_lines) # fold 詳細 fold_lines: list[str] = [] for r in sorted_results: scores_str = ", ".join( f"{s:.4f}" for s in r["f1_scores"] ) fold_lines.append( f" ・{r['name']}: [{scores_str}]" ) fold_str = "\n".join(fold_lines) # classification_report のインデント report_indented = "\n".join( f" {line}" for line in full_report.splitlines() ) content = f"""\ ======================================================================== 十字路分類モデル評価 (Intersection Classifier Evaluation) ======================================================================== 1. 概要 (Overview) ------------------------------------------------------------------------ 1-0. 目的 十字路（intersection）と通常区間（normal）を分類する 二値画像分類モデルの比較評価結果を記録する． 1-1. 評価日時 ・実施日時: {timestamp} 1-2. データセット ・入力: 40×30 二値画像（1200 特徴量，0.0/1.0） ・全サンプル数: {n_samples} - intersection: {n_intersection} - normal: {n_normal} ・クラス比率: intersection:normal\ = {n_intersection}:{n_normal} 2. 評価方法 (Evaluation Method) ------------------------------------------------------------------------ 2-1. 交差検証 ・手法: Stratified {N_FOLDS}-Fold Cross-Validation ・指標: マクロ平均 F1 スコア ・前処理: StandardScaler（fold ごとに fit） ・乱数シード: {RANDOM_STATE} 3. 評価結果 (Results) ------------------------------------------------------------------------ 3-1. モデル比較（F1 スコア降順） {table_str} 3-2. 各 Fold の F1 スコア {fold_str} 3-3. 採用モデル ・モデル: {best_name} ・保存先: params/intersection_model.pkl ・スケーラ: params/intersection_scaler.pkl 3-4. 全データでの分類レポート（再学習後） {report_indented} """ REPORT_PATH.parent.mkdir(parents=True, exist_ok=True) with open( REPORT_PATH, "w", encoding="utf-8", ) as f: f.write(content) print(f"\nドキュメント出力: {REPORT_PATH}") # ── メイン ──────────────────────────────────────────── def main() -> None: """学習・評価のメインフロー""" print("=== 十字路分類モデルの学習・評価 ===\n") # データ読み込み print("データ読み込み中...") try: x, y = load_dataset() except FileNotFoundError as e: print(f"エラー: {e}") sys.exit(1) n_samples = len(y) n_intersection = int(np.sum(y == 1)) n_normal = int(np.sum(y == 0)) print( f" サンプル数: {n_samples}" f" (intersection={n_intersection}," f" normal={n_normal})\n" ) # CV 評価 print(f"{N_FOLDS}-Fold CV 評価中...") results = _evaluate_models(x, y) # 最良モデル選定 best = max(results, key=lambda r: r["f1_mean"]) best_name = best["name"] print( f"\n最良モデル: {best_name}" f" (F1={best['f1_mean']:.4f})\n" ) # 全データで再学習・保存 print("全データで再学習・保存中...") full_report = _train_best_and_save(best_name, x, y) print(f" モデル保存: {MODEL_PATH}") print(f" スケーラ保存: {SCALER_PATH}") # ドキュメント出力 _write_report( results, best_name, n_samples, n_intersection, n_normal, full_report, ) print("\n完了") if __name__ == "__main__": main()