QuantEngineByItz/tools/build_macro_event_synchronizer_v2.py

"""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())