# -*- coding: utf-8 -*- """店舗別の後段モデル検証（追加課金ゼロ・LLM再実行なし）。 狙い: - 提出元店舗ごとに後段モデルを作成し、採点データの正確度を確認する。 - 採点にブレ（同一観測状態でスコアが割れる）があれば、その店舗の モデル精度は下がるはず — という仮説を、モデル精度＋ラベルノイズ床で検証する。 手順: 1. SQLite(metadata)→base_export.csv(提出者) を添付ファイル名で突合し ID→店舗 を作成。 2. 既存の build_features() で特徴量生成、target_out_of_range除外、assessable=1ゲート(打ち手1)。 3. 店舗ごとに 5-fold OOF（線形回帰 / visual_plus_brix_slot）で MAE/グレード一致/±15点 を測定。 4. 店舗ごとにラベルノイズ床（同一観測状態内のスコア分散＝採点ブレ）を §9.7 と同手法で測定。 5. プール（全店舗まとめ）をベースラインとして比較。 """ from __future__ import annotations import importlib.util, sys, json, sqlite3 from pathlib import Path import numpy as np, pandas as pd from sklearn.model_selection import KFold from sklearn.pipeline import Pipeline from sklearn.base import clone ROOT = Path(__file__).resolve().parents[2] LEGACY = ROOT / "検証" / "scripts" / "run_downstream_model_comparison.py" INPUT = ROOT / "local_batch_api" / "output_full" / "training_dataset.csv" SQLITE = ROOT / "local_batch_api" / "output_full" / "ramen_soup_evaluations.sqlite3" BASE_EXPORT = ROOT / "data" / "raw_csv" / "base_export.csv" OUTDIR = ROOT / "精度改善_20260619" / "10_store_validation" SEED, N = 42, 5 MIN_N_MODEL = 40 # これ未満の店舗はモデル学習が不安定なため参考値扱い def load_legacy(): spec = importlib.util.spec_from_file_location("legacy_cmp", LEGACY) m = importlib.util.module_from_spec(spec); sys.modules["legacy_cmp"] = m spec.loader.exec_module(m); return m def build_store_map() -> pd.DataFrame: """ID(=sample_id) -> 店舗 の対応表を作る。""" con = sqlite3.connect(SQLITE) ev = pd.read_sql("select metadata_json from evaluations", con) con.close() meta = ev["metadata_json"].map(lambda s: json.loads(s) if s else {}) ev["sample_id"] = meta.map(lambda d: str(d.get("ID")).strip() if d.get("ID") else None) ev["att"] = meta.map(lambda d: str(d.get("添付ファイル")).strip() if d.get("添付ファイル") else None) be = pd.read_csv(BASE_EXPORT, encoding="utf-8") att_col, sub_col = be.columns[4], be.columns[1] be = be[[att_col, sub_col]].copy() be.columns = ["att", "store_raw"] be["att"] = be["att"].astype(str).str.strip() # 添付ファイルは重複し得るので店舗で最頻値をとる be = be.dropna(subset=["att"]).drop_duplicates(subset=["att"], keep="first") m = ev.merge(be, on="att", how="left") m["store"] = m["store_raw"].fillna("不明").astype(str) return m[["sample_id", "store"]].dropna(subset=["sample_id"]) def oof(data, fs, target, mod, kf): X = data[list(fs.numeric) + list(fs.categorical)] y = data[target].to_numpy(); pred = np.zeros(len(y)) for tr, te in kf.split(X): p = Pipeline([("prep", mod.preprocessor(list(fs.numeric), list(fs.categorical))), ("model", clone(mod.model_specs(SEED)[1][1]))]) # 線形回帰 p.fit(X.iloc[tr], y[tr]); pred[te] = np.clip(p.predict(X.iloc[te]), 0, 100) return y, pred def model_stat(y, pred, grade): return {"MAE": round(float(np.abs(y - pred).mean()), 2), "グレード一致%": round(float((grade(y) == grade(pred)).mean()) * 100, 1), "±15点%": round(float((np.abs(y - pred) <= 15).mean()) * 100, 1)} def noise_floor(df, target, grade): """同一観測状態（LLM評価＋濃度＋時間が同一）でのスコアばらつき＝採点ブレ。 §9.7 と同手法。群平均予測でも残る誤差＝モデルで原理的に当てられないノイズ床。""" keys = ["visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality", "brix", "slot"] d = df.dropna(subset=[target]).copy() d["brix"] = d["brix"].fillna(-1) # 欠損もひとつの状態として扱う(groupby脱落防止) g = d.groupby(keys, dropna=False)[target] grp_mean = g.transform("mean") resid = d[target].to_numpy() - grp_mean.to_numpy() var_total = float(np.var(d[target].to_numpy())) var_resid = float(np.var(resid)) r2_upper = (1 - var_resid / var_total) if var_total > 0 else float("nan") grp_size = g.transform("size") # 群内でグレードが割れる件数（群サイズ2以上のみ対象） multi = d[grp_size >= 2].copy() if len(multi): gm = multi.groupby(keys)["__grade"] if "__grade" in multi else None d["__grade"] = grade(d[target].to_numpy()) split_cnt = 0 eligible = 0 for _, sub in d.groupby(keys, dropna=False): if len(sub) >= 2: eligible += len(sub) if sub["__grade"].nunique() > 1: split_cnt += len(sub) return { "ノイズ床MAE": round(float(np.abs(resid).mean()), 2), "ノイズ床std": round(float(np.std(resid)), 1), "説明可能上限R2%": round(r2_upper * 100, 1), "ユニーク観測状態数": int(d.groupby(keys, dropna=False).ngroups), "グレード割れ件数": int(split_cnt), "群内2件以上の件数": int(eligible), "グレード割れ率%": round(100 * split_cnt / eligible, 1) if eligible else float("nan"), } def evaluate_group(name, df, mod, kf, fsets): out = {"store": name, "n": len(df)} for target, label in [("human_total_score", "total"), ("human_soup_score", "soup")]: nf = noise_floor(df, target, mod.grade) if len(df) >= MIN_N_MODEL: y, pred = oof(df, fsets["visual_plus_brix_slot"], target, mod, kf) ms = model_stat(y, pred, mod.grade) else: ms = {"MAE": None, "グレード一致%": None, "±15点%": None} for k, v in ms.items(): out[f"{label}_{k}"] = v out[f"{label}_ノイズ床MAE"] = nf["ノイズ床MAE"] out[f"{label}_説明可能上限R2%"] = nf["説明可能上限R2%"] out[f"{label}_観測状態数"] = nf["ユニーク観測状態数"] out[f"{label}_グレード割れ率%"] = nf["グレード割れ率%"] return out def subsample_match(df_big, n, target, mod, fsets, seeds=(1, 7, 13, 21, 42)): """大店舗を小店舗とnを揃えてサブサンプルし、5-fold OOFを複数seed平均。 n差の交絡を排除して『採点の当てやすさ』だけを比較する。""" maes, grades = [], [] for s in seeds: sub = df_big.sample(n=n, random_state=s).reset_index(drop=True) kf = KFold(N, shuffle=True, random_state=s) y, pred = oof(sub, fsets["visual_plus_brix_slot"], target, mod, kf) maes.append(float(np.abs(y - pred).mean())) grades.append(float((mod.grade(y) == mod.grade(pred)).mean()) * 100) return round(float(np.mean(maes)), 2), round(float(np.mean(grades)), 1) def main(): OUTDIR.mkdir(parents=True, exist_ok=True) mod = load_legacy() f = mod.build_features(INPUT) f = f[f["target_out_of_range"] == 0].reset_index(drop=True) smap = build_store_map() f = f.drop(columns=["store"], errors="ignore") # legacyのstore(全件同一)を破棄 f["sample_id"] = f["sample_id"].astype(str).str.strip() f = f.merge(smap.drop_duplicates(subset=["sample_id"]), on="sample_id", how="left") f["store"] = f["store"].fillna("不明") # 打ち手1ゲート(assessable=1) ＝最終精度と同条件 g = f[f["assessable"] == 1].reset_index(drop=True) fsets = {x.name: x for x in mod.feature_sets()} kf = KFold(N, shuffle=True, random_state=SEED) # 店舗分布（ゲート前後） dist = pd.DataFrame({ "全体(target整合)": f["store"].value_counts(), "assessable=1": g["store"].value_counts(), }).fillna(0).astype(int) dist.index.name = "store" dist = dist.sort_values("assessable=1", ascending=False) dist.to_csv(OUTDIR / "store_distribution.csv", encoding="utf-8-sig") print("=== 店舗分布 ===") print(dist.to_string()) print() # 店舗別評価（assessable=1基準、十分なnの店舗のみモデル学習） rows = [evaluate_group("【プール】全店舗", g, mod, kf, fsets)] target_stores = dist[dist["assessable=1"] >= 10].index.tolist() for st in target_stores: sub = g[g["store"] == st].reset_index(drop=True) rows.append(evaluate_group(st, sub, mod, kf, fsets)) res = pd.DataFrame(rows) res.to_csv(OUTDIR / "store_validation.csv", index=False, encoding="utf-8-sig") # 表示 cols_total = ["store", "n", "total_MAE", "total_グレード一致%", "total_±15点%", "total_ノイズ床MAE", "total_説明可能上限R2%", "total_観測状態数", "total_グレード割れ率%"] cols_soup = ["store", "n", "soup_MAE", "soup_グレード一致%", "soup_±15点%", "soup_ノイズ床MAE", "soup_説明可能上限R2%", "soup_観測状態数", "soup_グレード割れ率%"] print("=== [total] 店舗別 ===") print(res[cols_total].to_string(index=False)) print() print("=== [soup] 店舗別 ===") print(res[cols_soup].to_string(index=False)) # n差の交絡排除: 神保町をn=56にサブサンプルして小田原と同条件で比較 print("\n=== n揃え対照(神保町をn=56にサブサンプル, 5seed平均) ===") jin = g[g["store"] == "豚山　神保町店"].reset_index(drop=True) oda = g[g["store"] == "豚山　小田原店"].reset_index(drop=True) n_match = len(oda) sub_rows = [] for target, label in [("human_total_score", "total"), ("human_soup_score", "soup")]: jm, jg = subsample_match(jin, n_match, target, mod, fsets) oy, op = oof(oda, fsets["visual_plus_brix_slot"], target, mod, KFold(N, shuffle=True, random_state=SEED)) om, og = round(float(np.abs(oy - op).mean()), 2), round(float((mod.grade(oy) == mod.grade(op)).mean()) * 100, 1) sub_rows.append({"target": label, "n": n_match, "神保町(n揃え)_MAE": jm, "神保町(n揃え)_グレード一致%": jg, "小田原_MAE": om, "小田原_グレード一致%": og}) sub_df = pd.DataFrame(sub_rows) sub_df.to_csv(OUTDIR / "subsample_matched.csv", index=False, encoding="utf-8-sig") print(sub_df.to_string(index=False)) print(f"\nsaved: {OUTDIR}") if __name__ == "__main__": main()