QuantEngineByItz/tools/build_strategy_harness_v2.py

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_JSON = ROOT / "GatherTradingData.json"
DEFAULT_OUT = ROOT / "Temp" / "strategy_harness_v2.json"
KST = timezone(timedelta(hours=9))


def load_json(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("payload must be object")
    return payload


def rows(v: Any) -> list[dict[str, Any]]:
    return [r for r in (v or []) if isinstance(r, dict)] if isinstance(v, list) else []


def num(v: Any, default: float | None = None) -> float | None:
    try:
        if v is None or v == "":
            return default
        return float(v)
    except Exception:
        return default


def build(payload: dict[str, Any]) -> dict[str, Any]:
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    h = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {}
    core = rows(data.get("core_satellite"))
    cash = rows(h.get("cash_raise_plan_json"))
    smart = rows(h.get("smart_cash_raise_json"))

    buy_rows = []
    for r in core:
        state = str(r.get("Execution_Recommendation_State") or "")
        late = num(r.get("Late_Chase_Risk_Score"), 0) or 0
        t1 = num(r.get("T1_Forced_Sell_Risk_Score"), 0) or 0
        timing = num(r.get("Timing_Score_Entry"), 0) or 0
        blocked = state.startswith("BUY_BLOCKED") or late >= 70 or t1 >= 60 or timing < 50
        buy_rows.append(
            {
                "ticker": r.get("Ticker"),
                "buy_permission_state": "BLOCK" if blocked else "ALLOW_PILOT",
                "alpha_lead_score_proxy": timing,
                "late_chase_risk_score": late,
                "t1_forced_sell_risk_score": t1,
                "block_reason_code": state if blocked else "PASS",
            }
        )

    sell_rows = []
    cash_by_ticker = {str(r.get("ticker") or ""): r for r in cash}
    for ticker, r in cash_by_ticker.items():
        immediate = int(num(r.get("immediate_qty"), 0) or 0)
        wait = int(num(r.get("rebound_wait_qty"), 0) or 0)
        emergency = bool(r.get("emergency_full_sell") is True)
        style = str(r.get("execution_style") or "")
        valid = emergency or style != "OVERSOLD_REBOUND_SELL" or (immediate >= 0 and wait >= 0)
        sell_rows.append(
            {
                "ticker": ticker,
                "execution_style": style or "UNKNOWN",
                "immediate_qty": immediate,
                "rebound_wait_qty": wait,
                "rebound_trigger_price": r.get("rebound_trigger"),
                "expected_slippage_band": "LOW" if immediate <= 100 else "MID",
                "state": "PASS" if valid else "BLOCK",
            }
        )

    cash_rows = []
    smart_by_ticker = {str(r.get("ticker") or ""): r for r in smart}
    for ticker, r in cash_by_ticker.items():
        s = smart_by_ticker.get(ticker, {})
        cash_rows.append(
            {
                "ticker": ticker,
                "target_cash_krw": r.get("target_cash_krw"),
                "immediate_qty": r.get("immediate_qty"),
                "protected_qty": r.get("protected_qty"),
                "rebound_wait_qty": r.get("rebound_wait_qty"),
                "route": s.get("smart_cash_raise_route"),
                "reason_code": s.get("reason_code") or r.get("reason_code"),
            }
        )

    return {
        "as_of": datetime.now(KST).isoformat(timespec="seconds"),
        "schema_version": "2026-05-22-strategy-harness-v2",
        "buy_anti_late_chase_harness_v2": buy_rows,
        "sell_value_preserve_harness_v2": sell_rows,
        "cash_raise_optimizer_harness_v2": cash_rows,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Build strategy harness v2 artifact from GatherTradingData.")
    parser.add_argument("--json", default=str(DEFAULT_JSON))
    parser.add_argument("--output", default=str(DEFAULT_OUT))
    args = parser.parse_args()

    json_path = Path(args.json)
    out_path = Path(args.output)
    if not json_path.is_absolute():
        json_path = ROOT / json_path
    if not out_path.is_absolute():
        out_path = ROOT / out_path

    payload = load_json(json_path)
    out = build(payload)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"STRATEGY_HARNESS_V2_BUILT: {out_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())