"""build_performance_monitoring_dashboard_v1.py — PERFORMANCE_MONITORING_DASHBOARD_V1 모든 핵심 지표(스코어·라우팅·매도·예측·완료기준·달성률)를 단일 JSON으로 집계하는 성과 모니터링 대시보드. 산출물: Temp/performance_monitoring_dashboard_v1.json """ from __future__ import annotations import argparse import json from datetime import date from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] TEMP = ROOT / "Temp" DEFAULT_JSON = ROOT / "GatherTradingData.json" DEFAULT_OUT = TEMP / "performance_monitoring_dashboard_v1.json" FORMULA_ID = "PERFORMANCE_MONITORING_DASHBOARD_V1" NA = "not_available" def _load(path: Path) -> Any: if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def _f(v: Any, default: float | None = None) -> float | None: try: return float(v) except Exception: return default def _extract_harness(payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): return {} h = payload.get("hApex") dc = (payload.get("data") or {}).get("_harness_context") if isinstance(h, dict) and isinstance(dc, dict): m = dict(dc); m.update(h); return m return h if isinstance(h, dict) else dc if isinstance(dc, dict) else payload def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--json", default=str(DEFAULT_JSON)) ap.add_argument("--out", default=str(DEFAULT_OUT)) args = ap.parse_args() json_path = Path(args.json); json_path = json_path if json_path.is_absolute() else ROOT / json_path out_path = Path(args.out); out_path = out_path if out_path.is_absolute() else ROOT / args.out payload = _load(json_path) harness = _extract_harness(payload) # 하네스 출력 로드 audit = _load(TEMP / "engine_audit_v1.json") truth = _load(TEMP / "operational_truth_score_v1.json") gap = _load(TEMP / "completion_gap_v1.json") scores = _load(TEMP / "scores_harness_v1.json") routing = _load(TEMP / "strategy_routing_audit_v1.json") sell = _load(TEMP / "sell_engine_audit_v1.json") pred = _load(TEMP / "prediction_accuracy_harness_v2.json") perf = _load(TEMP / "realized_performance_v1.json") horizon_plan = _load(TEMP / "horizon_rebalance_plan_v1.json") ycc = _load(TEMP / "yaml_code_coverage_v1.json") exp = (audit.get("imputed_data_exposure") or {}) fv = (audit.get("final_verdict") or {}) # 핵심 지표 집계 dashboard = { "formula_id": FORMULA_ID, "as_of_date": date.today().isoformat(), # 1. 엔진 상태 요약 "engine_status": { "audit_status": fv.get("status"), "investment_decision_allowed": fv.get("investment_decision_allowed"), "global_execution_gate": (audit.get("decision") or {}).get("global_execution_gate"), "decision_source": "deterministic_rule_engine", }, # 2. spec/30 달성 현황 "spec30": { "pass_rate_pct": gap.get("pass_rate_pct"), "passed": gap.get("passed_count"), "total": gap.get("total_criteria"), "immediate_actions": gap.get("immediate_actions", []), "immediate_count": len(gap.get("immediate_actions") or []), }, # 3. 스코어 현황 (SCORES_HARNESS_V1) "scores": { "fundamental": (scores.get("scores") or {}).get("fundamental_score"), "smart_money": (scores.get("scores") or {}).get("smart_money_score"), "liquidity": (scores.get("scores") or {}).get("liquidity_score"), "momentum": (scores.get("scores") or {}).get("momentum_score"), "valuation": (scores.get("scores") or {}).get("valuation_score"), "risk": (scores.get("scores") or {}).get("risk_score"), "final": (scores.get("final_score") or {}).get("value"), "dominant_horizon": scores.get("dominant_horizon"), "action_implied": "sell/partial_sell" if (_f((scores.get("final_score") or {}).get("value"), 100) or 100) < 45 else "hold", }, # 4. 신뢰도 캡 정직성 "confidence": { "raw_cap": exp.get("raw_confidence_cap_basis"), "honest_cap": exp.get("effective_confidence_honest"), "inflation_gap": exp.get("confidence_cap_inflation_gap"), "imputed_field_ratio": exp.get("imputed_field_ratio"), "gate": exp.get("gate_status"), "long_horizon_allowed": exp.get("long_horizon_allowed"), "fundamental_claim_allowed": exp.get("fundamental_claim_allowed"), }, # 5. 라우팅 현황 (STRATEGY_ROUTING_AUDIT_V1) "routing": { "selected_horizon": routing.get("selected_horizon"), "current_short_pct": horizon_plan.get("current_short_pct"), "short_cap_pct": horizon_plan.get("short_cap_pct"), "excess_pct": horizon_plan.get("excess_pct"), "routing_confidence": routing.get("routing_confidence"), "horizon_conflict_count": routing.get("horizon_conflict_count"), "gate": routing.get("gate"), }, # 6. 매도 현황 (SELL_ENGINE_AUDIT_V1) "sell_engine": { "gate": sell.get("gate"), "sell_type_counts": sell.get("sell_type_counts"), "breach_tickers": sell.get("breach_tickers"), "cash_shortfall_min_krw": (sell.get("scr_plan") or {}).get("cash_shortfall_min_krw"), "execution_allowed": (sell.get("scr_plan") or {}).get("execution_allowed"), }, # 7. 예측 정확도 (PREDICTION_ACCURACY_HARNESS_V2) "prediction": { "t1_op_rate": pred.get("t1_op_rate"), "t5_op_rate": pred.get("t5_op_rate"), "t20_op_rate": pred.get("t20_op_rate"), "t20_replay_rate": pred.get("t20_replay_rate"), "calibration_state": pred.get("calibration_state"), "replay_calibration_state": pred.get("replay_calibration_state"), }, # 8. 커버리지 (YAML_TO_CODE_COVERAGE_V1) "coverage": { "yaml_to_code_ratio": _f(ycc.get("coverage_ratio")), "golden_test_ratio": _f(ycc.get("golden_coverage_ratio")), "golden_test_count": ycc.get("golden_test_count"), "yaml_formula_count": ycc.get("yaml_formula_count"), "behavioral_coverage_pct": (audit.get("audit") or {}).get("behavioral_coverage_pct"), }, # 9. 운영 진실성 점수 (OPERATIONAL_TRUTH_SCORE_V1) "truth_score": { "score_0_100": truth.get("score_0_100"), "gate": truth.get("gate"), "data_truth_score": truth.get("data_truth_score"), "decision_truth_score": truth.get("decision_truth_score"), "performance_readiness_score": truth.get("performance_readiness_score"), "report_consistency_score": truth.get("report_consistency_score"), }, # 10. 은퇴 목표 달성 현황 "retirement_goal": { "total_asset_krw": harness.get("total_asset_krw"), "goal_achievement_pct": harness.get("goal_achievement_pct"), "goal_remaining_krw": harness.get("goal_remaining_krw"), "goal_status": harness.get("goal_status"), }, # 11. 즉시 실행 가능한 개선 요약 "immediate_improvement_summary": { "item_count": len(gap.get("immediate_actions") or []), "actions": (gap.get("priority_roadmap") or {}).get("P1_immediately", []), "estimated_effect": ( "GAS JSON 내보내기 후: schema_presence SLA 해소, " "fundamentals 로드 시 honest_cap 48.4→65+, " "BREACH 4종목 정리 시 routing_gate 개선 가능" ), }, } out_path.write_text(json.dumps(dashboard, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") spec30 = dashboard["spec30"] scores_d = dashboard["scores"] print( f"[{FORMULA_ID}] " f"spec30={spec30['passed']}/{spec30['total']}({spec30['pass_rate_pct']}%) " f"final_score={scores_d['final']} " f"breach={len(dashboard['sell_engine'].get('breach_tickers') or [])} " f"honest_cap={dashboard['confidence']['honest_cap']} -> {out_path}" ) return 0 if __name__ == "__main__": raise SystemExit(main())