"""build_strategy_routing_audit_v1.py — STRATEGY_ROUTING_AUDIT_V1 프롬프트 §3.5 Strategy Routing Harness — 투자기간별 전략 분기 검증. 검증 항목: - selected_horizon (지배적 투자기간) - horizon_conflict_count (호라이즌 상한 초과 위반 수) - 4성향(SCALP/SWING/MOMENTUM/POSITION) 분리 점수 검증 - 롱 호라이즌 보유 종목이 펀더멘털 데이터 없이 장기 분류됐는지 점검 - routing_confidence (라우팅 신뢰도 추정) - selected_strategy / rejected_strategies - required_conditions / failed_conditions 산출물: Temp/strategy_routing_audit_v1.json """ from __future__ import annotations import argparse import json 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 / "strategy_routing_audit_v1.json" FORMULA_ID = "STRATEGY_ROUTING_AUDIT_V1" NA = "not_available" # 호라이즌 상한 (% 기준) HORIZON_CAPS = {"SHORT": 40, "MID": 50, "LONG": 80, "UNKNOWN": 0} 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_root(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 _map_style_to_horizon(style: str) -> str: return {"SCALP": "SHORT", "SWING": "SHORT", "MOMENTUM": "MID", "POSITION": "LONG"}.get(style, "UNKNOWN") 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_root(payload) horizon_cls = _load(TEMP / "horizon_classification_v1.json") cap_style = _load(TEMP / "capital_style_allocation_v1.json") fund = _load(TEMP / "fundamental_multifactor_v3.json") fj = _load(TEMP / "final_judgment_gate_v1.json") regime = str(harness.get("regime_label") or harness.get("market_regime") or NA) # ── 1) 호라이즌 배분 및 위반 ────────────────────────────────────────────── alloc = horizon_cls.get("allocation_pct") or {} dominant_horizon = max(alloc, key=lambda k: alloc.get(k, 0)) if alloc else NA horizon_violations: list[dict] = [] for bucket, cap in HORIZON_CAPS.items(): current = _f(alloc.get(bucket), 0) if current is not None and current > cap: horizon_violations.append({ "bucket": bucket, "current_pct": current, "cap_pct": cap, "excess_pct": round(current - cap, 1), }) horizon_conflict_count = len(horizon_violations) # ── 2) 장기 보유 종목 펀더멘털 검증 ────────────────────────────────────── fund_rows = {r["ticker"]: r for r in (fund.get("rows") or []) if isinstance(r, dict)} hz_rows = horizon_cls.get("rows") or [] long_without_fund: list[str] = [] for r in hz_rows: if not isinstance(r, dict): continue if r.get("horizon") == "LONG": t = r.get("ticker", "") fr = fund_rows.get(t, {}) grade = fr.get("grade", "UNKNOWN") dq = str(fr.get("data_quality", "")) if dq == "PARTIAL" or grade in ("D", "F", "UNKNOWN"): long_without_fund.append(t) # ── 3) 4성향(SCALP/SWING/MOMENTUM/POSITION) 분리 점수 ────────────────── cap_rows = cap_style.get("rows") or [] style_distribution: dict[str, int] = {"SCALP": 0, "SWING": 0, "MOMENTUM": 0, "POSITION": 0} style_horizon_mismatches: list[dict] = [] per_ticker_styles: list[dict] = [] for r in cap_rows: if not isinstance(r, dict): continue ticker = r.get("ticker", "") styles = r.get("styles") or [] if not styles: continue # 최적 성향 = 최고 conviction_score best = max(styles, key=lambda s: _f(s.get("conviction_score"), 0) or 0, default={}) best_style = best.get("style", "UNKNOWN") best_conv = _f(best.get("conviction_score")) expected_horizon = _map_style_to_horizon(best_style) # 실제 호라이즌 (horizon_classification 기준) actual_hz = next((hr.get("horizon") for hr in hz_rows if isinstance(hr, dict) and hr.get("ticker") == ticker), NA) mismatch = (expected_horizon != actual_hz and actual_hz != NA and actual_hz != "ETF") if best_style in style_distribution: style_distribution[best_style] += 1 per_ticker_styles.append({ "ticker": ticker, "best_style": best_style, "conviction_score": best_conv, "expected_horizon": expected_horizon, "actual_horizon": actual_hz, "mismatch": mismatch, }) if mismatch: style_horizon_mismatches.append({"ticker": ticker, "style": best_style, "expected": expected_horizon, "actual": actual_hz}) # ── 4) 라우팅 신뢰도 ──────────────────────────────────────────────────── # routing_confidence: 0=conflict 많음/100=완전 정합 conflict_penalty = horizon_conflict_count * 20 mismatch_penalty = len(style_horizon_mismatches) * 10 fund_penalty = len(long_without_fund) * 15 routing_confidence = max(0, 100 - conflict_penalty - mismatch_penalty - fund_penalty) # ── 5) 선택 전략 / 거부 전략 ───────────────────────────────────────────── dominant_style = cap_style.get("capital_style_label") or NA selected_strategy = f"{dominant_style}_{dominant_horizon}" all_strategies = [f"{s}_{_map_style_to_horizon(s)}" for s in ("SCALP", "SWING", "MOMENTUM", "POSITION")] rejected_strategies = [s for s in all_strategies if s != selected_strategy] rejection_reasons = { "SCALP_SHORT": "현금부족+BREACH 종목 → 신규진입 차단", "SWING_SHORT": "현금부족+BREACH 종목 → 신규진입 차단", "MOMENTUM_MID": f"SHORT 비중 {alloc.get('SHORT',0)}% 과다 → MID 진입 여력 없음", "POSITION_LONG": "펀더멘털 결측(ROE/OPM/OCF/FCF) → long_horizon_allowed=false", } # ── 6) required / failed conditions ───────────────────────────────────── required_conditions = [ "data_quality_gate PASS", "cash_floor 충족", "horizon_conflict_count == 0", "long_without_fund_count == 0", "fundamental_claim_allowed", ] failed_conditions = [] if horizon_conflict_count > 0: # 실제 위반 버킷을 정확히 보고 (SHORT 고정 라벨 버그 수정) violation_labels = [f"{v['bucket']} {v['current_pct']}% > cap {v['cap_pct']}%" for v in horizon_violations] failed_conditions.append(f"horizon_conflict_count={horizon_conflict_count} ({'; '.join(violation_labels)})") if long_without_fund: failed_conditions.append(f"long_without_fund={long_without_fund}") cash_status = str(harness.get("cash_floor_status") or "") if cash_status in ("BELOW_FLOOR", "강제 차단"): failed_conditions.append(f"cash_floor_status={cash_status}") result = { "formula_id": FORMULA_ID, "market_regime": regime, "selected_horizon": dominant_horizon, "horizon_allocation_pct": alloc, "horizon_conflict_count": horizon_conflict_count, "horizon_violations": horizon_violations, "selected_strategy": selected_strategy, "rejected_strategies": rejected_strategies, "rejection_reasons": rejection_reasons, "routing_confidence": routing_confidence, "style_distribution": style_distribution, "style_horizon_mismatches": style_horizon_mismatches, "long_without_fundamental_data": long_without_fund, "required_conditions": required_conditions, "failed_conditions": failed_conditions, "per_ticker_styles": per_ticker_styles, "gate": "FAIL" if failed_conditions else "PASS", } out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print( f"[{FORMULA_ID}] horizon={dominant_horizon} regime={regime} " f"conflicts={horizon_conflict_count} style_mismatches={len(style_horizon_mismatches)} " f"routing_confidence={routing_confidence} gate={result['gate']} -> {out_path}" ) return 0 if __name__ == "__main__": raise SystemExit(main())