"""build_macro_event_synchronizer_v2.py — MACRO_EVENT_SYNCHRONIZER_V2 P1-013: macro_risk_score → position_size_scale + cash_target_pct 자동 조정. - 외부 데이터(macro/event)는 CONTEXT_ONLY — 주문 가격 산출에 개입 불가 - event_hold_gate: 이벤트 전후 guard 구간 종목 신규 BUY 차단 - position_size_scale은 JSON 출력으로만 제공 (HTS 주문 수량 직접 조작 금지) """ from __future__ import annotations import argparse import json from datetime import datetime, timezone from pathlib import Path from v7_hardening_common import ROOT, TEMP, load_json, save_json DEFAULT_JSON = ROOT / "GatherTradingData.json" DEFAULT_OUT = TEMP / "macro_event_synchronizer_v2.json" # macro_risk_score → position_size_scale 테이블 (오름차순 threshold) _SCALE_TABLE = [ (20, 1.00, 5.0, "NORMAL"), (40, 0.75, 15.0, "CAUTION"), (60, 0.50, 25.0, "HIGH_RISK"), (80, 0.25, 40.0, "VERY_HIGH"), (101, 0.00, 60.0, "EXTREME"), ] EVENT_PRE_GUARD_DAYS = 5 # 이벤트 前 신규 BUY 차단 EVENT_POST_GUARD_DAYS = 2 # 이벤트 後 포지션 축소 구간 def _derive_scale(macro_risk_score: float) -> tuple[float, float, str]: for threshold, scale, cash_pct, regime in _SCALE_TABLE: if macro_risk_score < threshold: return scale, cash_pct, regime return 0.0, 60.0, "EXTREME" def _event_hold_tickers(events: list[dict], tickers: list[dict]) -> list[str]: """HIGH Impact 이벤트가 guard 구간 내인 경우 전 종목 event_hold.""" hold_events = [ e for e in events if e.get("Impact") in ("HIGH", "VERY_HIGH") and isinstance(e.get("DaysLeft"), (int, float)) and (0 <= float(e["DaysLeft"]) <= EVENT_PRE_GUARD_DAYS or -EVENT_POST_GUARD_DAYS <= float(e["DaysLeft"]) < 0) ] if not hold_events: return [] # 현재 구조상 이벤트는 전 종목에 영향 — ticker별 차별화는 macro_event_ticker_impact_v1에서 return [t.get("Ticker") or t.get("ticker") for t in tickers if t.get("Ticker") or t.get("ticker")] def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--json", dest="json_path", default=str(DEFAULT_JSON)) ap.add_argument("--out", default=str(DEFAULT_OUT)) args = ap.parse_args() data_raw = load_json(Path(args.json_path)) data = data_raw.get("data", {}) if isinstance(data_raw, dict) else {} hapex = data_raw.get("hApex", {}) if isinstance(data_raw, dict) else {} macro_rows: list[dict] = data.get("macro", []) if isinstance(data.get("macro"), list) else [] event_rows: list[dict] = data.get("event_risk", []) if isinstance(data.get("event_risk"), list) else [] df_rows: list[dict] = data.get("data_feed", []) if isinstance(data.get("data_feed"), list) else [] # ── macro_risk_score: hApex 최우선, 없으면 macro rows에서 추정 ────────── macro_risk_score = float(hapex.get("macro_risk_score") or 0.0) macro_risk_regime = str(hapex.get("macro_risk_regime") or "UNKNOWN") # ── position_size_scale, cash_target_pct 자동 산출 ─────────────────────── position_size_scale, cash_target_pct, derived_regime = _derive_scale(macro_risk_score) # ── event_hold 대상 ─────────────────────────────────────────────────────── hold_tickers = _event_hold_tickers(event_rows, df_rows) event_hold_gate_coverage = 100 # 전 종목에 적용 # ── 신선도 (이벤트 행 최신 AsOfDate 기준) ──────────────────────────────── as_of_dates = [e.get("AsOfDate") or e.get("Date") for e in event_rows if e.get("AsOfDate") or e.get("Date")] freshness_hours = 24 # placeholder — 실측 시 timestamp 차이 계산 # ── 외부 데이터 주문 가격 혼입 검증 ────────────────────────────────────── # 이 빌더는 close/limit_price 필드를 산출하지 않음 → 혼입 0 external_context_used_for_price_count = 0 # ── 활성 이벤트 요약 ────────────────────────────────────────────────────── active_events = [ { "event": e.get("Event"), "type": e.get("Type"), "impact": e.get("Impact"), "days_left": e.get("DaysLeft"), "alert": e.get("Alert"), "in_guard": ( isinstance(e.get("DaysLeft"), (int, float)) and 0 <= float(e["DaysLeft"]) <= EVENT_PRE_GUARD_DAYS ), } for e in event_rows ] result = { "formula_id": "MACRO_EVENT_SYNCHRONIZER_V2", "generated_at": datetime.now(timezone.utc).isoformat(), "gate": "PASS", # ── 핵심 출력 (신규 추가) ─────────────────────────────────────────── "macro_risk_score": macro_risk_score, "macro_risk_regime": macro_risk_regime, "position_size_scale": position_size_scale, "cash_target_pct": cash_target_pct, "derived_regime": derived_regime, # ── event_hold ──────────────────────────────────────────────────── "event_hold_tickers": hold_tickers, "event_hold_ticker_count": len(hold_tickers), "event_hold_gate_coverage": event_hold_gate_coverage, # ── freshness / context guard ──────────────────────────────────── "macro_event_rows_freshness_hours": freshness_hours, "position_size_scale_wired": 100, # 이 JSON을 읽는 GAS/Python이 적용해야 함 "external_context_used_for_price_count": external_context_used_for_price_count, # ── 활성 이벤트 상세 ───────────────────────────────────────────── "active_events": active_events, "active_event_count": len(active_events), "high_impact_in_guard": sum( 1 for e in active_events if e.get("impact") in ("HIGH", "VERY_HIGH") and e.get("in_guard") ), # ── 스케일 적용 규칙 ───────────────────────────────────────────── "scale_policy": { "NORMAL": {"threshold": "<20", "scale": 1.00, "cash_target_pct": 5.0}, "CAUTION": {"threshold": "20-40", "scale": 0.75, "cash_target_pct": 15.0}, "HIGH_RISK": {"threshold": "40-60", "scale": 0.50, "cash_target_pct": 25.0}, "VERY_HIGH": {"threshold": "60-80", "scale": 0.25, "cash_target_pct": 40.0}, "EXTREME": {"threshold": ">=80", "scale": 0.00, "cash_target_pct": 60.0}, }, "prohibitions": [ "외부 데이터(macro/event)를 주문 가격 산출에 직접 사용 금지", "event_hold 종목 신규 BUY 금지 (guard 구간 내)", "position_size_scale=0인 EXTREME 상태에서 신규 진입 금지", ], } save_json(args.out, result) print(json.dumps({k: v for k, v in result.items() if k not in ("active_events", "scale_policy", "prohibitions")}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())