# -*- coding: utf-8 -*- """Run downstream model comparison from the completed LLM scoring CSV. Outputs: - features.csv - splits.csv - model_comparison.csv - predictions.csv - feature_importance.csv - summary.md """ from __future__ import annotations import argparse import math import re from dataclasses import dataclass from pathlib import Path import numpy as np import pandas as pd from sklearn.compose import ColumnTransformer from sklearn.ensemble import RandomForestRegressor from sklearn.impute import SimpleImputer from sklearn.linear_model import LinearRegression, Ridge from sklearn.metrics import mean_absolute_error, mean_squared_error from sklearn.model_selection import StratifiedShuffleSplit from sklearn.pipeline import Pipeline from sklearn.preprocessing import OneHotEncoder, StandardScaler try: from xgboost import XGBRegressor except Exception: # pragma: no cover XGBRegressor = None try: from lightgbm import LGBMRegressor except Exception: # pragma: no cover LGBMRegressor = None ROOT = Path(__file__).resolve().parents[2] DEFAULT_INPUT = ROOT / "local_batch_api" / "output_full" / "training_dataset.csv" DEFAULT_OUTDIR = ROOT / "検証" / "results" / "downstream_model_comparison" RANDOM_SEED = 42 UNASSESSABLE = "判定不可" WATER_SCORE = { "水位良好": 1.0, "水位不足": 0.0, UNASSESSABLE: 0.0, } OIL_SCORE = { "油感と乳化の状態が良い": 1.0, "乳化が進んでいる": 0.85, "油感が強い": 0.45, "乳化しすぎ": 0.35, UNASSESSABLE: 0.0, } BOILING_SCORE = { "良い": 1.0, "弱い": 0.45, "強すぎる": 0.35, UNASSESSABLE: 0.0, } PHOTO_SCORE = { "良い": 1.0, "湯気が多い": 0.45, "白飛び": 0.35, "暗い": 0.35, "角度不良": 0.35, "不良": 0.0, UNASSESSABLE: 0.0, } CONFIDENCE_SCORE = { "高": 1.0, "中": 0.6, "低": 0.3, UNASSESSABLE: 0.0, } @dataclass(frozen=True) class FeatureSet: name: str numeric: tuple[str, ...] categorical: tuple[str, ...] def grade(score: np.ndarray | pd.Series) -> np.ndarray: values = np.asarray(score, dtype=float) out = np.full(values.shape, "D", dtype="= 21] = "C" out[values >= 60] = "B" out[values >= 80] = "A" return out def spearman(y_true: np.ndarray, y_pred: np.ndarray) -> float: yt = pd.Series(y_true).rank(method="average").to_numpy() yp = pd.Series(y_pred).rank(method="average").to_numpy() if np.std(yt) == 0 or np.std(yp) == 0: return float("nan") return float(np.corrcoef(yt, yp)[0, 1]) def metrics(y_true: np.ndarray, y_pred: np.ndarray) -> dict[str, float | int]: pred = np.clip(np.asarray(y_pred, dtype=float), 0, 100) true = np.asarray(y_true, dtype=float) anomaly_true = true < 60 anomaly_pred = pred < 60 out: dict[str, float | int] = { "n": int(len(true)), "rho": round(spearman(true, pred), 4), "mae": round(float(mean_absolute_error(true, pred)), 4), "rmse": round(float(math.sqrt(mean_squared_error(true, pred))), 4), "grade_match": round(float((grade(true) == grade(pred)).mean()), 4), } out["anomaly_n"] = int(anomaly_true.sum()) out["anomaly_recall"] = ( round(float((anomaly_true & anomaly_pred).sum() / anomaly_true.sum()), 4) if anomaly_true.sum() else float("nan") ) return out def video_id_from_path(path: object) -> str: name = Path(str(path)).stem match = re.search(r"\d+", name) return match.group(0) if match else name def to_density(value: object) -> float: try: return float(value) except Exception: return np.nan def map_score(series: pd.Series, mapping: dict[str, float]) -> pd.Series: return series.fillna(UNASSESSABLE).astype(str).str.strip().map(mapping).fillna(0.0) def build_features(input_path: Path) -> pd.DataFrame: raw = pd.read_csv(input_path) df = pd.DataFrame() df["row_id"] = np.arange(len(raw)) df["sample_id"] = raw["video_path"].map(video_id_from_path) df["video_path"] = raw["video_path"] df["frame_path"] = raw["frame_path"] df["store"] = raw["store"].fillna("unknown").astype(str) df["slot"] = raw["slot"].fillna("unknown").astype(str) df["brix"] = pd.to_numeric(raw["brix_raw"], errors="coerce") df["human_soup_score"] = pd.to_numeric(raw["human_soup_score"], errors="coerce") df["human_total_score"] = pd.to_numeric(raw["human_total_score"], errors="coerce") df["target_out_of_range"] = ( ~df["human_soup_score"].between(0, 100, inclusive="both") | ~df["human_total_score"].between(0, 100, inclusive="both") ).astype(int) for col in [ "visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality", ]: df[col] = raw[col].fillna(UNASSESSABLE).astype(str).str.strip() df["image_condition_score"] = pd.to_numeric(raw["image_condition_score"], errors="coerce").fillna(0) df["visual_density_num"] = df["visual_density"].map(to_density) df["visual_density_assessable"] = df["visual_density_num"].notna().astype(int) df["assessable"] = ( (df["visual_density"] != UNASSESSABLE) & (df["water_level"] != UNASSESSABLE) & (df["oil_emulsification"] != UNASSESSABLE) & (df["boiling_heat_state"] != UNASSESSABLE) ).astype(int) df["retake_recommended"] = ((df["photo_quality"] != "良い") | (df["assessable"] == 0)).astype(int) df["density_brix_abs_diff"] = (df["visual_density_num"] - df["brix"]).abs() df["density_brix_signed_diff"] = df["visual_density_num"] - df["brix"] df["density_below_brix"] = (df["visual_density_num"] < df["brix"]).fillna(False).astype(int) df["water_score"] = map_score(df["water_level"], WATER_SCORE) df["oil_score"] = map_score(df["oil_emulsification"], OIL_SCORE) df["boiling_score"] = map_score(df["boiling_heat_state"], BOILING_SCORE) df["photo_score"] = map_score(df["photo_quality"], PHOTO_SCORE) density_match = (1.0 - (df["density_brix_abs_diff"].fillna(4).clip(0, 4) / 4.0)).clip(0, 1) df["density_match_score"] = density_match df["fixed_score"] = ( density_match * 35 + df["water_score"] * 20 + df["oil_score"] * 20 + df["boiling_score"] * 15 + df["photo_score"] * 10 ).clip(0, 100) # Convenience features for simple models. df["slot_is_opening"] = df["slot"].str.contains("開店", na=False).astype(int) df["slot_is_evening"] = df["slot"].str.contains("17|18|19|夜|夕", regex=True, na=False).astype(int) df["low_quality_flag"] = (df["image_condition_score"] <= 3).astype(int) df["unassessable_axis_count"] = ( (df[["visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality"]] == UNASSESSABLE) .sum(axis=1) .astype(int) ) df = df.dropna(subset=["human_soup_score", "human_total_score"]).reset_index(drop=True) return df def make_splits(df: pd.DataFrame, target: str, seed: int) -> pd.DataFrame: labels = grade(df[target]) first = StratifiedShuffleSplit(n_splits=1, test_size=0.2, random_state=seed) train_val_idx, test_idx = next(first.split(df, labels)) train_val = df.iloc[train_val_idx].copy() train_val_labels = grade(train_val[target]) second = StratifiedShuffleSplit(n_splits=1, test_size=0.125, random_state=seed) train_rel, val_rel = next(second.split(train_val, train_val_labels)) split = pd.Series(index=df.index, dtype=object) split.iloc[test_idx] = "test" split.iloc[train_val_idx[train_rel]] = "train" split.iloc[train_val_idx[val_rel]] = "val" return pd.DataFrame({"row_id": df["row_id"], "sample_id": df["sample_id"], "split": split.values}) def preprocessor(numeric: list[str], categorical: list[str]) -> ColumnTransformer: try: encoder = OneHotEncoder(handle_unknown="ignore", sparse_output=False) except TypeError: # older sklearn encoder = OneHotEncoder(handle_unknown="ignore", sparse=False) return ColumnTransformer( transformers=[ ("num", Pipeline([("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler())]), numeric), ("cat", Pipeline([("imputer", SimpleImputer(strategy="most_frequent")), ("onehot", encoder)]), categorical), ], remainder="drop", verbose_feature_names_out=False, ) def model_specs(seed: int): specs = [ ("固定配点", None), ("線形回帰", LinearRegression()), ("Ridge回帰", Ridge(alpha=10.0, random_state=seed)), ("Random Forest", RandomForestRegressor(n_estimators=500, min_samples_leaf=6, random_state=seed, n_jobs=-1)), ] if XGBRegressor is not None: specs.append( ( "XGBoost", XGBRegressor( n_estimators=400, max_depth=2, learning_rate=0.03, subsample=0.85, colsample_bytree=0.85, reg_lambda=5.0, objective="reg:squarederror", random_state=seed, n_jobs=-1, ), ) ) if LGBMRegressor is not None: specs.append( ( "LightGBM", LGBMRegressor( n_estimators=400, max_depth=2, learning_rate=0.03, subsample=0.85, colsample_bytree=0.85, reg_lambda=5.0, random_state=seed, verbose=-1, ), ) ) return specs def feature_sets() -> list[FeatureSet]: visual_numeric = ( "visual_density_num", "image_condition_score", "water_score", "oil_score", "boiling_score", "photo_score", "visual_density_assessable", "assessable", "retake_recommended", "low_quality_flag", "unassessable_axis_count", ) visual_cat = ( "visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality", ) return [ FeatureSet("visual_only", visual_numeric, visual_cat), FeatureSet( "visual_plus_brix", visual_numeric + ( "brix", "density_brix_abs_diff", "density_brix_signed_diff", "density_below_brix", "density_match_score", ), visual_cat, ), FeatureSet( "visual_plus_brix_slot", visual_numeric + ( "brix", "density_brix_abs_diff", "density_brix_signed_diff", "density_below_brix", "density_match_score", "slot_is_opening", "slot_is_evening", ), visual_cat + ("slot",), ), FeatureSet("brix_only", ("brix",), ()), ] def transformed_feature_names(pipe: Pipeline) -> list[str]: prep = pipe.named_steps["prep"] try: return list(prep.get_feature_names_out()) except Exception: return [] def collect_importance(model, names: list[str]) -> pd.DataFrame: estimator = model.named_steps["model"] values = None if hasattr(estimator, "feature_importances_"): values = estimator.feature_importances_ elif hasattr(estimator, "coef_"): values = np.abs(np.ravel(estimator.coef_)) if values is None or not names: return pd.DataFrame() out = pd.DataFrame({"feature": names, "importance": values}) return out.sort_values("importance", ascending=False) def run_comparison(features: pd.DataFrame, splits: pd.DataFrame, outdir: Path, seed: int) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: data = features.merge(splits[["row_id", "split"]], on="row_id", how="left") comparison_rows = [] prediction_rows = [] importance_rows = [] targets = ["human_soup_score", "human_total_score"] for target in targets: train = data[data["split"] == "train"].copy() val = data[data["split"] == "val"].copy() test = data[data["split"] == "test"].copy() for fs in feature_sets(): numeric = list(fs.numeric) categorical = list(fs.categorical) for model_name, estimator in model_specs(seed): if model_name == "固定配点": for split_name, part in [("val", val), ("test", test)]: pred = part["fixed_score"].to_numpy() row = { "target": target, "feature_set": fs.name, "model": model_name, "split": split_name, **metrics(part[target].to_numpy(), pred), } comparison_rows.append(row) for row_id, sid, truth, value in zip(part["row_id"], part["sample_id"], part[target], pred): prediction_rows.append( { "row_id": row_id, "target": target, "feature_set": fs.name, "model": model_name, "split": split_name, "sample_id": sid, "human_score": truth, "prediction": round(float(value), 4), "error": round(float(value - truth), 4), } ) continue pipe = Pipeline([("prep", preprocessor(numeric, categorical)), ("model", estimator)]) pipe.fit(train[numeric + categorical], train[target]) names = transformed_feature_names(pipe) imp = collect_importance(pipe, names).head(30) if not imp.empty: imp.insert(0, "target", target) imp.insert(1, "feature_set", fs.name) imp.insert(2, "model", model_name) importance_rows.extend(imp.to_dict("records")) for split_name, part in [("val", val), ("test", test)]: pred = np.clip(pipe.predict(part[numeric + categorical]), 0, 100) comparison_rows.append( { "target": target, "feature_set": fs.name, "model": model_name, "split": split_name, **metrics(part[target].to_numpy(), pred), } ) for row_id, sid, truth, value in zip(part["row_id"], part["sample_id"], part[target], pred): prediction_rows.append( { "row_id": row_id, "target": target, "feature_set": fs.name, "model": model_name, "split": split_name, "sample_id": sid, "human_score": truth, "prediction": round(float(value), 4), "error": round(float(value - truth), 4), } ) comparison = pd.DataFrame(comparison_rows) predictions = pd.DataFrame(prediction_rows) importance = pd.DataFrame(importance_rows) return comparison, predictions, importance def pivot_comparison(comparison: pd.DataFrame) -> pd.DataFrame: index = ["target", "feature_set", "model"] values = ["n", "rho", "mae", "rmse", "grade_match", "anomaly_recall", "anomaly_n"] wide = comparison.pivot_table(index=index, columns="split", values=values, aggfunc="first") wide.columns = [f"{split}_{metric}" for metric, split in wide.columns] wide = wide.reset_index() order = ["target", "feature_set", "model"] for split in ["val", "test"]: for metric in values: col = f"{split}_{metric}" if col in wide.columns: order.append(col) return wide[order].sort_values(["target", "feature_set", "test_mae", "model"]) def write_summary(outdir: Path, features: pd.DataFrame, wide: pd.DataFrame, excluded: pd.DataFrame) -> None: best = ( wide[wide["feature_set"] != "brix_only"] .sort_values(["target", "test_mae"]) .groupby("target") .head(5) ) def markdown_table(df: pd.DataFrame) -> str: if df.empty: return "_no rows_" cols = list(df.columns) rows = ["| " + " | ".join(cols) + " |", "| " + " | ".join(["---"] * len(cols)) + " |"] for _, row in df.iterrows(): vals = [] for col in cols: value = row[col] if isinstance(value, float): vals.append("" if pd.isna(value) else f"{value:.4g}") else: vals.append("" if pd.isna(value) else str(value)) rows.append("| " + " | ".join(vals) + " |") return "\n".join(rows) lines = [ "# 後段モデル比較結果", "", f"- 入力: `local_batch_api/output_full/training_dataset.csv`", f"- 件数: {len(features)}", f"- 100点スケール外で除外: {len(excluded)}", f"- assessable: {int(features['assessable'].sum())} / {len(features)}", f"- 判定不可含む行: {int((features['unassessable_axis_count'] > 0).sum())} / {len(features)}", f"- split: train/val/test = 70/10/20, stratified by `human_total_score`, seed={RANDOM_SEED}", "", "## 出力ファイル", "", "- `features.csv`: モデル投入用特徴量", "- `splits.csv`: 固定分割", "- `model_comparison.csv`: 6モデル比較結果", "- `predictions.csv`: 各動画ごとの予測と誤差", "- `feature_importance.csv`: 重要特徴量", "", "## 上位結果(test MAE順)", "", markdown_table(best), "", "## 注意", "", "- 人間コメントは特徴量に入れていません。", "- 固定配点は学習なしの基準値です。", "- `visual_density` は既存出力を使用し、評価方針自体は変更していません。", ] (outdir / "summary.md").write_text("\n".join(lines), encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input", type=Path, default=DEFAULT_INPUT) parser.add_argument("--outdir", type=Path, default=DEFAULT_OUTDIR) parser.add_argument("--seed", type=int, default=RANDOM_SEED) args = parser.parse_args() args.outdir.mkdir(parents=True, exist_ok=True) features = build_features(args.input) excluded = features[features["target_out_of_range"] == 1].copy() features = features[features["target_out_of_range"] == 0].reset_index(drop=True) splits = make_splits(features, "human_total_score", args.seed) comparison_long, predictions, importance = run_comparison(features, splits, args.outdir, args.seed) comparison = pivot_comparison(comparison_long) features.to_csv(args.outdir / "features.csv", index=False, encoding="utf-8-sig") excluded.to_csv(args.outdir / "excluded_target_outliers.csv", index=False, encoding="utf-8-sig") splits.to_csv(args.outdir / "splits.csv", index=False, encoding="utf-8-sig") comparison_long.to_csv(args.outdir / "model_comparison_long.csv", index=False, encoding="utf-8-sig") comparison.to_csv(args.outdir / "model_comparison.csv", index=False, encoding="utf-8-sig") predictions.to_csv(args.outdir / "predictions.csv", index=False, encoding="utf-8-sig") importance.to_csv(args.outdir / "feature_importance.csv", index=False, encoding="utf-8-sig") write_summary(args.outdir, features, comparison, excluded) print(f"saved: {args.outdir}") print(comparison.sort_values(["target", "test_mae"]).groupby("target").head(8).to_string(index=False)) if __name__ == "__main__": main()