"""build_continuous_evaluation_dashboard_v1.py — CONTINUOUS_EVALUATION_DASHBOARD_V1 P2-020: 주간 성과 대시보드. - LIVE T+20 표본에서 기대수익/승률/MDD/수익반납 지표 산출 - REPLAY 표본은 informational 섹션에만 집계 (성과 계산 혼입 금지) - T+20 미확정 → None으로 표기, INSUFFICIENT_DATA 게이트 """ from __future__ import annotations import argparse import json from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any from v7_hardening_common import ROOT, TEMP, load_json, save_json DEFAULT_HIST = ROOT / "Temp" / "proposal_evaluation_history.json" DEFAULT_OUT = TEMP / "continuous_evaluation_dashboard_v1.json" MIN_T20_FOR_METRICS = 30 # 성과 지표 신뢰성 최소 표본 수 _REPLAY_ORIGINS = {"REPLAY_FROM_KRX_EOD", "HISTORICAL_REPLAY", "BACKTEST"} _REPLAY_VALIDATION = {"REPLAY", "HISTORICAL_REPLAY"} def _is_replay(r: dict) -> bool: return ( str(r.get("data_origin") or "").upper() in _REPLAY_ORIGINS or str(r.get("validation_status") or "").upper() in _REPLAY_VALIDATION or str(r.get("record_type") or "").upper().startswith("HISTORICAL_REPLAY") ) def _is_evaluated_t20(r: dict) -> bool: return ( r.get("t20_evaluation_status") == "EVALUATED_T20" and r.get("t20_return_pct") is not None ) def _iso_week(date_str: str | None) -> str | None: if not date_str: return None try: dt = datetime.strptime(str(date_str)[:10], "%Y-%m-%d") return dt.strftime("%G-W%V") except ValueError: return None def _compute_metrics(t20_returns: list[float]) -> dict[str, Any]: if not t20_returns: return { "expectancy_pct": None, "win_rate_pct": None, "max_drawdown_pct": None, "trade_count": 0, } wins = [r for r in t20_returns if r > 0] return { "expectancy_pct": round(sum(t20_returns) / len(t20_returns), 4), "win_rate_pct": round(len(wins) / len(t20_returns) * 100, 2), "max_drawdown_pct": round(min(t20_returns), 4), "trade_count": len(t20_returns), } def _build_weekly_scorecard(live_eval: list[dict]) -> list[dict]: weeks: dict[str, list[float]] = defaultdict(list) for r in live_eval: week = _iso_week(r.get("proposal_date") or r.get("created_at")) if week: weeks[week].append(float(r["t20_return_pct"])) scorecard = [] for week in sorted(weeks.keys()): returns = weeks[week] m = _compute_metrics(returns) m["week"] = week scorecard.append(m) return scorecard def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--hist", default=str(DEFAULT_HIST)) ap.add_argument("--out", default=str(DEFAULT_OUT)) args = ap.parse_args() hist_raw = load_json(Path(args.hist)) records: list[dict] = ( hist_raw.get("records", []) if isinstance(hist_raw, dict) else (hist_raw if isinstance(hist_raw, list) else []) ) # ── 분류 ───────────────────────────────────────────────────────────────── live_all = [r for r in records if not _is_replay(r)] replay_all = [r for r in records if _is_replay(r)] live_eval = [r for r in live_all if _is_evaluated_t20(r)] live_t20_count = len(live_eval) insufficient = live_t20_count < MIN_T20_FOR_METRICS # ── 성과 지표 (LIVE T+20 전체) ──────────────────────────────────────── returns = [float(r["t20_return_pct"]) for r in live_eval] if live_eval else [] overall = _compute_metrics(returns) # ── 주간 스코어카드 ────────────────────────────────────────────────── weekly_scorecard = _build_weekly_scorecard(live_eval) # ── gate 판정 ───────────────────────────────────────────────────────── exp = overall.get("expectancy_pct") wr = overall.get("win_rate_pct") if insufficient: gate = "INSUFFICIENT_DATA" elif exp is not None and wr is not None and (exp < 0 or wr < 40): gate = "WARNING" else: gate = "PASS" result = { "formula_id": "CONTINUOUS_EVALUATION_DASHBOARD_V1", "generated_at": datetime.now(timezone.utc).isoformat(), "gate": gate, # ── 전체 지표 ──────────────────────────────────────────────────── "weekly_scorecard_generated": len(weekly_scorecard) > 0, "expectancy_pct": overall.get("expectancy_pct"), "win_rate_pct": overall.get("win_rate_pct"), "max_drawdown_pct": overall.get("max_drawdown_pct"), "profit_giveback_pct": None, # T+20 이후 추적 미구현 "total_live_evaluated_t20": live_t20_count, "total_live_pending": len(live_all) - live_t20_count, # ── 주간 스코어카드 ───────────────────────────────────────────── "weekly_scorecard": weekly_scorecard, "weekly_scorecard_count": len(weekly_scorecard), # ── informational (REPLAY 분리) ────────────────────────────────── "replay_informational": { "replay_record_count": len(replay_all), "note": "REPLAY 표본은 성과 지표 계산에 포함되지 않음", }, # ── 데이터 신뢰성 ───────────────────────────────────────────── "data_confidence": { "sufficient_for_metrics": not insufficient, "min_required": MIN_T20_FOR_METRICS, "current_live_t20": live_t20_count, "gap": max(0, MIN_T20_FOR_METRICS - live_t20_count), "estimated_ready": "~2026-07-15" if insufficient else "NOW", }, "prohibitions": [ "REPLAY 표본을 성과 지표 계산에 포함 금지", "T+20 미확정 거래를 EVALUATED_T20으로 분류 금지", "외부 가격 데이터 직접 조회 금지 (history 기록 기준만 사용)", ], } save_json(args.out, result) # ── Excel 파일 기입 (GatherTradingData.xlsx > evaluation_dashboard) ── try: import openpyxl xlsx_path = ROOT / "GatherTradingData.xlsx" if xlsx_path.exists(): wb = openpyxl.load_workbook(xlsx_path) sheet_name = "evaluation_dashboard" # 기존 시트 클리어 및 재생성 if sheet_name in wb.sheetnames: ws = wb[sheet_name] ws.delete_rows(1, ws.max_row + 10) else: ws = wb.create_sheet(sheet_name) # 1. SUMMARY 섹션 기입 ws.append(["Metric", "Value"]) summary_metrics = [ ("Generated At", result["generated_at"]), ("Gate Status", result["gate"]), ("Expectancy Pct", result["expectancy_pct"]), ("Win Rate Pct", result["win_rate_pct"]), ("Max Drawdown Pct", result["max_drawdown_pct"]), ("Total Live Evaluated T+20", result["total_live_evaluated_t20"]), ("Total Live Pending", result["total_live_pending"]), ("Replay Record Count", result["replay_informational"]["replay_record_count"]), ("Sufficient For Metrics", result["data_confidence"]["sufficient_for_metrics"]), ("Min Required", result["data_confidence"]["min_required"]), ("Current Live T+20", result["data_confidence"]["current_live_t20"]), ("Gap", result["data_confidence"]["gap"]), ("Estimated Ready", result["data_confidence"]["estimated_ready"]) ] for m, v in summary_metrics: ws.append([m, v]) # 2. WEEKLY SCORECARD 섹션 기입 (오른쪽 열인 D열부터 시작) ws.cell(row=1, column=4, value="Week") ws.cell(row=1, column=5, value="Expectancy_Pct") ws.cell(row=1, column=6, value="Win_Rate_Pct") ws.cell(row=1, column=7, value="Max_Drawdown_Pct") ws.cell(row=1, column=8, value="Trade_Count") for idx, w in enumerate(weekly_scorecard): r_num = idx + 2 ws.cell(row=r_num, column=4, value=w.get("week")) ws.cell(row=r_num, column=5, value=w.get("expectancy_pct")) ws.cell(row=r_num, column=6, value=w.get("win_rate_pct")) ws.cell(row=r_num, column=7, value=w.get("max_drawdown_pct")) ws.cell(row=r_num, column=8, value=w.get("trade_count")) wb.save(xlsx_path) print(f"[EVALUATION_DASHBOARD] Saved evaluation_dashboard sheet to {xlsx_path.name}") except Exception as e: print(f"[EVALUATION_DASHBOARD][ERROR] Failed to save Excel sheet: {e}") suffix = f"(need {max(0, MIN_T20_FOR_METRICS - live_t20_count)} more LIVE T+20)" if insufficient else "" print( f"[CONTINUOUS_EVALUATION_DASHBOARD_V1] gate={gate} " f"live_t20={live_t20_count} " f"expectancy={overall.get('expectancy_pct')} " f"win_rate={overall.get('win_rate_pct')}% " f"MDD={overall.get('max_drawdown_pct')} " f"weeks={len(weekly_scorecard)} {suffix}" ) return 0 if __name__ == "__main__": raise SystemExit(main())