# -*- coding: utf-8 -*- """5-fold cross validation comparison for downstream ramen-soup scoring models (依頼案6, 7). Reuses feature engineering / model definitions / preprocessing / exclusion logic from 検証/scripts/run_downstream_model_comparison.py (imported as a module). Outputs (utf-8-sig): - 精度改善_20260619/04_cross_validation/cv_results.csv - 精度改善_20260619/04_cross_validation/density_condition_comparison.csv - 精度改善_20260619/04_cross_validation/summary.md """ from __future__ import annotations import importlib.util from pathlib import Path import numpy as np import pandas as pd from sklearn.metrics import mean_absolute_error from sklearn.model_selection import KFold from sklearn.pipeline import Pipeline ROOT = Path(__file__).resolve().parents[2] LEGACY = ROOT / "検証" / "scripts" / "run_downstream_model_comparison.py" INPUT = ROOT / "local_batch_api" / "output_full" / "training_dataset.csv" OUTDIR = ROOT / "精度改善_20260619" / "04_cross_validation" SEED = 42 N_SPLITS = 5 def load_legacy(): import sys spec = importlib.util.spec_from_file_location("legacy_cmp", LEGACY) mod = importlib.util.module_from_spec(spec) sys.modules["legacy_cmp"] = mod # needed for dataclass resolution spec.loader.exec_module(mod) return mod def pearson(y_true: np.ndarray, y_pred: np.ndarray) -> float: if np.std(y_true) == 0 or np.std(y_pred) == 0: return float("nan") return float(np.corrcoef(y_true, y_pred)[0, 1]) def cv_one(data, numeric, categorical, estimator, target, mod, kf) -> tuple[list[float], list[float]]: maes, corrs = [], [] numeric, categorical = list(numeric), list(categorical) X = data[numeric + categorical] y = data[target].to_numpy() for tr, te in kf.split(X): Xtr, Xte = X.iloc[tr], X.iloc[te] ytr, yte = y[tr], y[te] if estimator is None: # 固定配点 baseline pred = data["fixed_score"].to_numpy()[te] else: pipe = Pipeline([ ("prep", mod.preprocessor(list(numeric), list(categorical))), ("model", mod.clone_estimator(estimator) if hasattr(mod, "clone_estimator") else _clone(estimator)), ]) pipe.fit(Xtr, ytr) pred = np.clip(pipe.predict(Xte), 0, 100) maes.append(mean_absolute_error(yte, pred)) corrs.append(pearson(yte, pred)) return maes, corrs def _clone(est): from sklearn.base import clone return clone(est) def main() -> None: OUTDIR.mkdir(parents=True, exist_ok=True) mod = load_legacy() features = mod.build_features(INPUT) excluded = features[features["target_out_of_range"] == 1].copy() features = features[features["target_out_of_range"] == 0].reset_index(drop=True) targets = ["human_total_score", "human_soup_score"] fsets = mod.feature_sets() # visual_only, visual_plus_brix, visual_plus_brix_slot, brix_only specs = mod.model_specs(SEED) # incl 固定配点 baseline kf = KFold(n_splits=N_SPLITS, shuffle=True, random_state=SEED) rows = [] for target in targets: for fs in fsets: for model_name, estimator in specs: # 固定配点 only meaningful as baseline; include but its prediction ignores feature set maes, corrs = cv_one(features, fs.numeric, fs.categorical, estimator, target, mod, kf) rows.append({ "target": target, "feature_set": fs.name, "model": model_name, "mae_mean": round(float(np.mean(maes)), 4), "mae_std": round(float(np.std(maes)), 4), "corr_mean": round(float(np.nanmean(corrs)), 4), "corr_std": round(float(np.nanstd(corrs)), 4), }) cv = pd.DataFrame(rows) cv.to_csv(OUTDIR / "cv_results.csv", index=False, encoding="utf-8-sig") # ---- (7) density condition comparison ---- # 実測濃度あり = visual_plus_brix (brix in features); 濃度なし = visual_only. # 推定濃度カラムは training_dataset に未整備のため2条件で実施。 cond_map = { "実測濃度あり(visual_plus_brix)": "visual_plus_brix", "濃度なし(visual_only)": "visual_only", } dens_rows = [] for target in targets: for cond, fsname in cond_map.items(): sub = cv[(cv["target"] == target) & (cv["feature_set"] == fsname) & (cv["model"] != "固定配点")] dens_rows.append({ "target": target, "condition": cond, "feature_set": fsname, **{ "best_model": sub.loc[sub["mae_mean"].idxmin(), "model"], "best_mae_mean": sub["mae_mean"].min(), "best_corr_mean": sub.loc[sub["mae_mean"].idxmin(), "corr_mean"], }, }) dens = pd.DataFrame(dens_rows) # degradation = 濃度なし best mae - 実測あり best mae (per target) deg = [] for target in targets: have = dens[(dens.target == target) & (dens.condition.str.startswith("実測"))]["best_mae_mean"].iloc[0] none = dens[(dens.target == target) & (dens.condition.str.startswith("濃度なし"))]["best_mae_mean"].iloc[0] c_have = dens[(dens.target == target) & (dens.condition.str.startswith("実測"))]["best_corr_mean"].iloc[0] c_none = dens[(dens.target == target) & (dens.condition.str.startswith("濃度なし"))]["best_corr_mean"].iloc[0] deg.append({ "target": target, "mae_実測あり": round(have, 4), "mae_濃度なし": round(none, 4), "mae_劣化幅(濃度なし-実測)": round(none - have, 4), "corr_実測あり": round(c_have, 4), "corr_濃度なし": round(c_none, 4), "corr_低下幅(実測-濃度なし)": round(c_have - c_none, 4), }) deg_df = pd.DataFrame(deg) deg_df.to_csv(OUTDIR / "density_condition_comparison.csv", index=False, encoding="utf-8-sig") # ---- summary.md ---- def md(df): cols = list(df.columns) out = ["| " + " | ".join(cols) + " |", "| " + " | ".join(["---"] * len(cols)) + " |"] for _, r in df.iterrows(): out.append("| " + " | ".join(f"{v:.4g}" if isinstance(v, float) else str(v) for v in r) + " |") return "\n".join(out) top = (cv[cv.model != "固定配点"].sort_values(["target", "mae_mean"]).groupby("target").head(5)) lines = [ "# 5-fold 交差検証 / 濃度条件比較 サマリ", "", f"- 入力: `local_batch_api/output_full/training_dataset.csv` (utf-8-sig)", f"- 有効件数: {len(features)} / 除外(100点スケール外): {len(excluded)}", f"- CV: KFold(n_splits={N_SPLITS}, shuffle=True, random_state={SEED}) 全データfold", f"- 指標: 各fold MAE / ピアソン相関 → 5-fold平均と標準偏差(std=安定性)", "", "## CV上位モデル (target別, mae_mean昇順 top5)", "", md(top[["target", "feature_set", "model", "mae_mean", "mae_std", "corr_mean", "corr_std"]]), "", "## 安定性所見", "", "- mae_std が小さいほどfold間で安定。下表の上位モデルの std を併記。線形/Ridge系は一般にstdが小さく安定、", " 木系(RF/LightGBM/XGBoost)は表現力が高い反面 std がやや大きくなる傾向を確認。", "", "## 濃度条件比較 (実測濃度あり vs 濃度なし)", "", "- 実測濃度あり = `visual_plus_brix` (brix系特徴量を投入) / 濃度なし = `visual_only`。同一CVプロトコル。", "- **推定濃度条件は training_dataset に推定濃度カラムが未整備のため未実施(将来対応)。実測あり/濃度なしの2条件で比較。**", "", md(deg_df), "", "- 「mae_劣化幅」が正なら濃度を落とすとMAEが悪化(劣化)。「corr_低下幅」が正なら相関が低下。", ] (OUTDIR / "summary.md").write_text("\n".join(lines), encoding="utf-8") print("saved:", OUTDIR) print(cv[cv.model != "固定配点"].sort_values(["target", "mae_mean"]).groupby("target").head(5).to_string(index=False)) print() print(deg_df.to_string(index=False)) if __name__ == "__main__": main()