from __future__ import annotations import argparse import json from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json" DEFAULT_JSON = ROOT / "GatherTradingData.json" DEFAULT_ENGINE_GATE = ROOT / "Temp" / "engine_harness_gate_result.json" DEFAULT_OUT = ROOT / "Temp" / "strategy_hardening_harness_v1.json" def _load(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: obj = json.loads(path.read_text(encoding="utf-8")) except Exception: return {} return obj if isinstance(obj, dict) else {} def _sections_map(report: dict[str, Any]) -> dict[str, dict[str, Any]]: rows = report.get("sections") if isinstance(report.get("sections"), list) else [] out: dict[str, dict[str, Any]] = {} for row in rows: if not isinstance(row, dict): continue name = str(row.get("name") or "").strip() if name: out[name] = row return out def _has_section(sec: dict[str, dict[str, Any]], name: str) -> bool: return name in sec def _score_binary(flags: list[bool]) -> float: if not flags: return 0.0 return round((sum(1 for x in flags if x) / len(flags)) * 100.0, 2) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--report", default=str(DEFAULT_REPORT)) ap.add_argument("--json", default=str(DEFAULT_JSON)) ap.add_argument("--engine-gate", default=str(DEFAULT_ENGINE_GATE)) ap.add_argument("--out", default=str(DEFAULT_OUT)) args = ap.parse_args() rp = Path(args.report) jp = Path(args.json) ep = Path(args.engine_gate) op = Path(args.out) if not rp.is_absolute(): rp = ROOT / rp if not jp.is_absolute(): jp = ROOT / jp if not ep.is_absolute(): ep = ROOT / ep if not op.is_absolute(): op = ROOT / op report = _load(rp) data_json = _load(jp) engine_gate = _load(ep) sec = _sections_map(report) data = data_json.get("data") if isinstance(data_json.get("data"), dict) else {} hctx = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {} # Domain evidence routing_serving_score = _score_binary([ _has_section(sec, "routing_serving_trace"), _has_section(sec, "routing_serving_trace_v2"), _has_section(sec, "routing_decision_explain_v1"), _has_section(sec, "export_gate_diagnosis"), ]) decision_score = _score_binary([ _has_section(sec, "QEH_AUDIT_BLOCK"), _has_section(sec, "today_decision_summary_card"), _has_section(sec, "shadow_ledger_table"), _has_section(sec, "llm_constraint_audit"), ]) fundamental_score = _score_binary([ _has_section(sec, "fundamental_quality_gate_v1"), _has_section(sec, "fundamental_multifactor_v2"), _has_section(sec, "earnings_growth_quality_v1"), _has_section(sec, "market_share_proxy_v1"), _has_section(sec, "cashflow_stability_v1"), ]) horizon_score = _score_binary([ _has_section(sec, "horizon_allocation_lock_v1"), _has_section(sec, "t1_evaluation_summary_box"), _has_section(sec, "benchmark_relative_harness_table"), _has_section(sec, "index_relative_health_table"), ]) smart_money_score = _score_binary([ _has_section(sec, "smart_money_liquidity_gate_v1"), _has_section(sec, "alpha_lead_table"), _has_section(sec, "entry_freshness_gate_table"), _has_section(sec, "anti_distribution_table"), ]) profit_preservation_score = _score_binary([ _has_section(sec, "profit_preservation_table"), _has_section(sec, "sell_value_preservation_gate_table"), _has_section(sec, "scrs_v2_sell_table"), _has_section(sec, "mandatory_reduction_plan"), ]) cash_raise_score = _score_binary([ _has_section(sec, "cash_recovery_plan_crdl"), _has_section(sec, "smart_cash_raise_table"), _has_section(sec, "portfolio_structure_risks"), ]) outcome = _load(ROOT / "Temp" / "outcome_quality_score_v1.json") exec_quality = _load(ROOT / "Temp" / "execution_quality_harness_v1.json") eval_cov = _load(ROOT / "Temp" / "evaluation_history_coverage_v1.json") data_integrity = _load(ROOT / "Temp" / "data_integrity_score_v1.json") decision_evidence = _load(ROOT / "Temp" / "decision_evidence_score_v1.json") derivation = _load(ROOT / "Temp" / "derivation_validity_score_v1.json") algo_guidance = _load(ROOT / "Temp" / "algorithm_guidance_proof_v1.json") # Performance domains are numeric, not just section existence t20_pass_rate = float((outcome.get("metrics") or {}).get("t20_pass_rate") or 0.0) outcome_score = float(outcome.get("score") or 0.0) eq_oper = (exec_quality.get("metrics") or {}).get("operational_t20") if isinstance((exec_quality.get("metrics") or {}).get("operational_t20"), dict) else {} execution_expectancy = float(eq_oper.get("expectancy_pct") or 0.0) execution_mdd = float(eq_oper.get("max_drawdown_pct") or 0.0) execution_win_rate = float(eq_oper.get("win_rate_pct") or 0.0) data_integrity_score = float(data_integrity.get("score") or 0.0) decision_evidence_score = float(decision_evidence.get("score") or 0.0) derivation_score = float(derivation.get("score") or 0.0) algo_guidance_score = float(algo_guidance.get("score") or 0.0) # Engine hardening score focuses on deterministic control + measured outcome control_score = round((routing_serving_score + decision_score + data_integrity_score + decision_evidence_score + derivation_score + algo_guidance_score) / 6.0, 2) perf_subscores = [outcome_score, t20_pass_rate] if exec_quality: perf_subscores.append(max(0.0, min(100.0, 50.0 + execution_expectancy * 10.0))) perf_subscores.append(max(0.0, min(100.0, 100.0 - execution_mdd * 3.0))) perf_subscores.append(execution_win_rate) performance_score = round(sum(perf_subscores) / max(1, len(perf_subscores)), 2) overall = round(control_score * 0.6 + performance_score * 0.4, 2) readiness_gate = "PERFORMANCE_READY" if data_integrity_score < 100.0: readiness_gate = "BLOCKED_DATA_QUALITY" elif outcome_score < 60.0 or t20_pass_rate < 60.0 or str(exec_quality.get("gate") or "") in {"FAIL", "WATCH_PENDING_SAMPLE"}: readiness_gate = "NOT_PERFORMANCE_READY" # 100% target gap breakdown gaps = { "routing_serving_gap": round(100.0 - routing_serving_score, 2), "decision_gap": round(100.0 - decision_score, 2), "fundamental_gap": round(100.0 - fundamental_score, 2), "horizon_gap": round(100.0 - horizon_score, 2), "smart_money_gap": round(100.0 - smart_money_score, 2), "profit_preservation_gap": round(100.0 - profit_preservation_score, 2), "cash_raise_gap": round(100.0 - cash_raise_score, 2), "outcome_quality_gap": round(100.0 - outcome_score, 2), "t20_pass_rate_gap": round(100.0 - t20_pass_rate, 2), "execution_expectancy_gap": round(max(0.0, 0.1 - execution_expectancy), 3), "execution_mdd_over_gap": round(max(0.0, execution_mdd - 12.0), 2), "execution_win_rate_gap": round(max(0.0, 45.0 - execution_win_rate), 2), } actions = [] if outcome_score < 60 or t20_pass_rate < 60: actions.append({ "priority": "P0", "action_id": "PERF_RECOVERY_HARNESS_V1", "why": "매수/매도 뒷북·설거지 징후가 T20 패스율에 반영됨", "required_metrics": ["t20_pass_rate", "watch_miss_rate", "late_chase_block_precision", "rebound_sell_value_damage"], "target": {"t20_pass_rate_min": 60.0, "outcome_quality_min": 60.0}, }) if str(exec_quality.get("gate") or "") in {"FAIL", "WATCH_PENDING_SAMPLE"}: actions.append({ "priority": "P0", "action_id": "EXECUTION_QUALITY_HARNESS_V1", "why": "기대값/낙폭/승률을 운영 표본으로 고정 추적해 뒷북·설거지 구조를 계량 통제", "required_metrics": ["expectancy_pct", "max_drawdown_pct", "win_rate_pct", "samples"], "target": {"expectancy_pct_min": 0.0, "max_drawdown_pct_max": 12.0, "win_rate_pct_min": 45.0}, }) if data_integrity_score < 100: actions.append({ "priority": "P0", "action_id": "DATA_INTEGRITY_100_LOCK_V1", "why": "데이터 완성도 100 미달은 실전 판단 왜곡 유발", "required_metrics": ["required_field_completeness_pct", "placeholder_safety_pct", "capture_age_hours"], "target": {"data_integrity_score": 100.0}, }) if cash_raise_score < 100: actions.append({ "priority": "P1", "action_id": "CASH_RAISE_VALUE_DAMAGE_MIN_V1", "why": "현금확보 과정에서 주식가치 훼손 최소화 필요", "required_metrics": ["immediate_sell_ratio", "rebound_wait_ratio", "value_damage_pct_avg"], "target": {"value_damage_pct_avg_max": 10.0}, }) if fundamental_score < 100: actions.append({ "priority": "P1", "action_id": "FUNDAMENTAL_FEATURE_COMPLETION_V1", "why": "펀더멘털 결손은 중장기 판단 신뢰도 저하", "required_metrics": ["roe_coverage_pct", "opm_coverage_pct", "ocf_coverage_pct", "fcf_coverage_pct"], "target": {"feature_coverage_min": 95.0}, }) result = { "formula_id": "STRATEGY_HARDENING_HARNESS_V1", "source": { "report_json": str(rp), "data_json": str(jp), "engine_gate_json": str(ep), }, "engine_gate_status": engine_gate.get("status"), "domain_scores": { "routing_serving": routing_serving_score, "decision_governance": decision_score, "fundamental": fundamental_score, "horizon_short_mid_long": horizon_score, "smart_money_liquidity": smart_money_score, "profit_preservation": profit_preservation_score, "cash_raise_execution": cash_raise_score, "outcome_quality": outcome_score, "t20_pass_rate": t20_pass_rate, "execution_expectancy_pct": execution_expectancy, "execution_max_drawdown_pct": execution_mdd, "execution_win_rate_pct": execution_win_rate, "execution_quality_gate": exec_quality.get("gate"), "data_integrity": data_integrity_score, "decision_evidence": decision_evidence_score, "derivation_validity": derivation_score, "algorithm_guidance_proof": algo_guidance_score, }, "meta_scores": { "control_score": control_score, "performance_score": performance_score, "overall_hardening_score": overall, "evaluation_history_gate": eval_cov.get("gate"), "outcome_gate": outcome.get("gate"), "readiness_gate": readiness_gate, }, "gaps_to_100": gaps, "hardening_actions": actions, "determinism_lock": { "llm_numeric_free_will_allowed": False, "harness_context_keys": len(hctx.keys()) if isinstance(hctx, dict) else 0, }, } op.parent.mkdir(parents=True, exist_ok=True) op.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())