"""eval_harness 러너 — 정책 비교 + 학습곡선 (H5, PoC 본체). E2E: python -m eval_harness.runner --config configs/exp_default.yaml --tenant imarketkorea 판정: 학습형(qtable_ucb)이 random/static 대비 평균보상·성공률 우상향이면 "학습 루프 유효". 구매자는 카드별 효과가 다른 시뮬(HeuristicBuyer) — 학습 정책만 좋은 카드를 알아내 성과가 오른다. """ import argparse import json import os from typing import Dict, List import yaml from eval_harness.buyer import HeuristicBuyer, Scenario, best_actions, make_card_effectiveness from eval_harness.metrics import aggregate, good_card_hit_rate, hit_curve, learning_curve from eval_harness.registry import build_policy from eval_harness.simulator import run_episode from tenancy.config_loader import TenantConfigLoader _HERE = os.path.dirname(os.path.abspath(__file__)) def _resolve_config(path: str) -> str: if os.path.isabs(path) and os.path.exists(path): return path for cand in (path, os.path.join(_HERE, path), os.path.join(_HERE, "configs", os.path.basename(path))): if os.path.exists(cand): return cand raise FileNotFoundError(f"config not found: {path}") def run(config_path: str, tenant_id: str) -> dict: with open(_resolve_config(config_path), "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) tcfg = TenantConfigLoader().load(tenant_id) A = tcfg.action_mapping.action_space_size if A == 0: raise ValueError(f"tenant {tenant_id} 에 action 매핑이 없습니다 (action_space_size=0)") scn_raw = cfg.get("scenario", {}) scenario = Scenario( anchor_price=scn_raw.get("anchor_price", 8000), target_price=scn_raw.get("target_price", 10000), revenue_amount=scn_raw.get("revenue_amount", 20_000_000), distribution_code=scn_raw.get("distribution_code", "A"), partner_count=scn_raw.get("partner_count", 1), acceptance_ratio=scn_raw.get("acceptance_ratio", 0.05), ) episodes = int(cfg.get("episodes", 400)) seed = int(cfg.get("seed", 42)) max_turns = int(cfg.get("max_turns", 5)) buckets = int(cfg.get("curve_buckets", 10)) n_good = int(cfg.get("n_good_cards", 3)) policy_names = cfg.get("policies", ["random", "qtable_ucb"]) # 카드 효과(숨김) + 좋은 카드 — 구매자는 테넌트/seed 로 고정. 정책은 모른다. eff = make_card_effectiveness(A, seed=seed, n_good=n_good) good = best_actions(eff, k=n_good) report = {"tenant": tenant_id, "episodes": episodes, "seed": seed, "action_space_size": A, "good_cards": good, "policies": {}} for name in policy_names: policy = build_policy(name, tcfg.state, A, tcfg.policy, seed=seed) buyer = HeuristicBuyer(eff, seed=seed, max_turns=max_turns) results = [] for i in range(episodes): buyer.reseed(seed * 100_000 + i) # 페어드: 정책 간 동일 구매자 randomness learn = name in ("qtable_ucb", "qtable", "ucb") results.append(run_episode(policy, buyer, scenario, tcfg.state, tcfg.reward, A, max_turns=max_turns, learn=learn)) agg = aggregate(results) firsts = [r.first_action for r in results] # 초반 구간 미세 곡선(첫 50 에피소드, 5개씩) — 빠른 수렴 시 cold→warm 상승을 드러냄. early = hit_curve(firsts[:50], good, buckets=10) report["policies"][name] = { **agg.as_row(), "learning_curve": learning_curve(results, buckets=buckets), "good_card_hit_rate": round(good_card_hit_rate(firsts, good), 3), "hit_curve": hit_curve(firsts, good, buckets=buckets), "early_hit_curve": early, } report["verdict"] = _verdict(report) return report def _verdict(report: dict) -> dict: pols = report["policies"] learner = pols.get("qtable_ucb") baseline = pols.get("random") or pols.get("static") if not learner or not baseline: return {"pass": None, "note": "학습/비교군 부재"} # PoC 판정(계획서 G): 학습형이 random/정적 대비 성과 우상향. # - 평균보상 우위 + 95%CI 비중첩(통계적 분리) # - '좋은 카드' 적중 우위(학습으로 카드 우열을 알아냄) # - cold→warm: 초반 미세곡선이 baseline 수준에서 상승 beats_reward = learner["mean_reward"] - learner["reward_ci95"] > baseline["mean_reward"] + baseline["reward_ci95"] beats_hit = learner["good_card_hit_rate"] >= baseline["good_card_hit_rate"] + 0.2 ec = learner.get("early_hit_curve") or [0, 0] cold_warm = len(ec) >= 2 and ec[-1] > ec[0] passed = beats_reward and beats_hit return { "pass": bool(passed), "ci_separated": bool(beats_reward), "learner_mean_reward": learner["mean_reward"], "baseline_mean_reward": baseline["mean_reward"], "reward_uplift": round(learner["mean_reward"] - baseline["mean_reward"], 4), "learner_good_hit": learner["good_card_hit_rate"], "baseline_good_hit": baseline["good_card_hit_rate"], "early_cold_to_warm": [ec[0], ec[-1]] if ec else [], "cold_warm_rising": bool(cold_warm), "note": ("학습형이 baseline 대비 평균보상 우위(95%CI 분리) + '좋은 카드' 적중 우위 → 학습 루프 유효" if passed else "개선 미확인"), } def _print(report: dict): print("=" * 70) print(f" 알고리즘 비교 — tenant={report['tenant']} episodes={report['episodes']} seed={report['seed']}") print(f" (숨은) 좋은 카드 action: {report['good_cards']} / 총 {report['action_space_size']}개") print("=" * 70) hdr = f"{'policy':<12} {'success':>8} {'settled/tgt':>12} {'turns':>7} {'mean_rwd':>10} {'±95%CI':>9} {'good_hit':>9}" print(hdr); print("-" * len(hdr)) for name, p in report["policies"].items(): print(f"{name:<12} {p['success_rate']:>8.3f} {p['mean_settled_ratio']:>12.3f} " f"{p['mean_turns']:>7.2f} {p['mean_reward']:>10.4f} {p['reward_ci95']:>9.4f} {p['good_card_hit_rate']:>9.3f}") print("-" * len(hdr)) print("\n학습 개선곡선 ('좋은 카드' 선택률 구간별, 세일즈용):") for name, p in report["policies"].items(): print(f" {name:<12} 전구간 {p.get('hit_curve')}") print(f"\n qtable_ucb 초반 cold→warm (첫 50ep, 5개씩): {report['policies'].get('qtable_ucb', {}).get('early_hit_curve')}") v = report["verdict"] flag = "✅ PASS" if v.get("pass") else ("— " if v.get("pass") is None else "❌ FAIL") print(f"\n판정 {flag}: {v.get('note')}") if v.get("pass") is not None: print(f" 평균보상 학습형 {v['learner_mean_reward']} vs baseline {v['baseline_mean_reward']} " f"(uplift {v['reward_uplift']}, 95%CI 분리={v['ci_separated']})") print(f" 좋은카드 적중 학습형 {v['learner_good_hit']} vs baseline {v['baseline_good_hit']} · " f"cold→warm {v['early_cold_to_warm']}") def main(): ap = argparse.ArgumentParser(description="협상 정책 비교 하네스 (H5)") ap.add_argument("--config", default="configs/exp_default.yaml") ap.add_argument("--tenant", default="imarketkorea") ap.add_argument("--save", action="store_true", help="reports/ 에 JSON 저장") args = ap.parse_args() report = run(args.config, args.tenant) _print(report) if args.save: out_dir = os.path.join(_HERE, "reports") os.makedirs(out_dir, exist_ok=True) out = os.path.join(out_dir, f"report_{args.tenant}.json") with open(out, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) print(f"\n[저장] {out}") if __name__ == "__main__": main()