QuantEngineByItz/tools/build_relative_underperformance_alert_v1.py

from __future__ import annotations

import argparse
import json
import math
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_JSON = ROOT / "GatherTradingData.json"
DEFAULT_OUT = ROOT / "Temp" / "relative_underperformance_alert_v1.json"


def _load(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return payload if isinstance(payload, dict) else {}


def _to_float(v: Any) -> float | None:
    try:
        n = float(v)
    except Exception:
        return None
    return n if math.isfinite(n) else None


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

    jp = Path(args.json)
    op = Path(args.out)
    if not jp.is_absolute():
        jp = ROOT / jp
    if not op.is_absolute():
        op = ROOT / op

    payload = _load(jp)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    df_rows = data.get("data_feed") if isinstance(data.get("data_feed"), list) else []
    df_map = {}
    for row in df_rows:
      if isinstance(row, dict):
        tk = str(row.get("Ticker") or row.get("ticker") or "").strip()
        if tk:
            df_map[tk] = row

    kospi_ret20d = None
    macro_rows = data.get("macro") if isinstance(data.get("macro"), list) else []
    for row in macro_rows:
        if not isinstance(row, dict):
            continue
        if str(row.get("Name", "")).strip().upper() == "KOSPI":
            kospi_ret20d = _to_float(row.get("Ret20D"))
            break

    snap_rows = data.get("account_snapshot") if isinstance(data.get("account_snapshot"), list) else []
    holdings = []
    for row in snap_rows:
        if not isinstance(row, dict):
            continue
        if row.get("parse_status") != "CAPTURE_READ_OK":
            continue
        if str(row.get("user_confirmed") or "").upper() != "Y":
            continue
        if _to_float(row.get("holding_quantity")) in (None, 0):
            continue
        holdings.append(row)

    relative_rows = []
    absolute_rows = []
    ladder_rows = []

    for h in holdings:
        ticker = str(h.get("ticker") or "").strip()
        df = df_map.get(ticker, {})
        close = _to_float(h.get("close")) or _to_float(df.get("Close")) or 0.0
        ret20d = _to_float(df.get("Ret20D"))
        atr20 = _to_float(df.get("ATR20"))
        profit_pct = _to_float(h.get("return_pct"))
        hold_days = int(_to_float(h.get("holding_days")) or 0)

        abs_status = "INSUFFICIENT_DATA"
        recommended_stop = None
        if atr20 and close > 0:
            avg_cost = _to_float(h.get("avgCost")) or _to_float(h.get("average_cost")) or close
            atr_mul = 2.0 if (atr20 / max(close, 1e-9) * 100 >= 8.0) else 1.5
            recommended_stop = max(avg_cost * 0.92, avg_cost - atr20 * atr_mul)
            recommended_stop = round(recommended_stop)
            stop_price = _to_float(h.get("stopPrice")) or _to_float(h.get("stop_price"))
            if stop_price is not None:
                if stop_price < recommended_stop * 0.60:
                    abs_status = "STOP_CRITICAL"
                elif stop_price < recommended_stop * 0.85:
                    abs_status = "STOP_WIDE"
                else:
                    abs_status = "PASS"
            else:
                abs_status = "PASS"

        beta = 1.0
        if kospi_ret20d is not None and abs(kospi_ret20d) >= 0.5 and ret20d is not None:
            beta = min(3.0, max(0.3, ret20d / kospi_ret20d))
        excess_ret = (ret20d or 0.0) - beta * (kospi_ret20d or 0.0)
        sigma_proxy = ((atr20 or 0.0) / close * 100.0) * math.sqrt(20) if close > 0 and atr20 else None
        threshold = (-2.0 * sigma_proxy) if sigma_proxy is not None else None
        rel_trigger = False
        signal_type = "INSUFFICIENT_DATA"
        if sigma_proxy is not None and ret20d is not None and close > 0:
            rel_trigger = excess_ret < threshold
            abs_floor = profit_pct is not None and profit_pct < -20.0
            time_stop = hold_days >= 60 and excess_ret < 0
            signal_type = "ABS_FLOOR" if abs_floor else ("REL_EXCESS" if rel_trigger else ("TIME_STOP" if time_stop else "PASS"))

        relative_rows.append({
            "ticker": ticker,
            "name": h.get("name") or "",
            "signal": bool(signal_type != "PASS" and signal_type != "INSUFFICIENT_DATA"),
            "signal_type": signal_type,
            "beta_proxy": round(beta, 2) if beta is not None else None,
            "excess_ret20d": round(excess_ret, 2),
            "threshold": round(threshold, 2) if threshold is not None else None,
            "profit_pct": profit_pct,
            "formula_id": "RELATIVE_UNDERPERF_ALERT_V1",
        })

        absolute_rows.append({
            "ticker": ticker,
            "name": h.get("name") or "",
            "stop_price": _to_float(h.get("stopPrice")) or _to_float(h.get("stop_price")) or recommended_stop,
            "recommended_stop": recommended_stop,
            "adequacy_status": abs_status,
            "formula_id": "ABSOLUTE_RISK_STOP_V1",
        })

        action = "HOLD"
        ratio_pct = 0
        if signal_type == "ABS_FLOOR":
            action = "EXIT_100"
            ratio_pct = 100
        elif signal_type == "REL_EXCESS":
            action = "TRIM_50"
            ratio_pct = 50
        elif signal_type == "TIME_STOP":
            action = "TIME_EXIT_100" if hold_days >= 60 else "TIME_TRIM_50"
            ratio_pct = 100 if hold_days >= 60 else 50
        ladder_rows.append({
            "ticker": ticker,
            "name": h.get("name") or "",
            "action": action,
            "ratio_pct": ratio_pct,
            "limit_price": recommended_stop,
            "reason": signal_type,
            "formula_id": "STOP_ACTION_LADDER_V1",
        })

    out = {
        "formula_id": "RELATIVE_UNDERPERF_ALERT_V1",
        "gate": "PASS" if not any(r["signal"] for r in relative_rows) else "WARN",
        "kospi_ret20d": kospi_ret20d,
        "absolute_risk_stop_rows": absolute_rows,
        "relative_underperf_alert_rows": relative_rows,
        "stop_action_ladder_rows": ladder_rows,
        "triggered_count": sum(1 for r in relative_rows if r["signal"]),
        "holding_count": len(holdings),
    }

    op.parent.mkdir(parents=True, exist_ok=True)
    op.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(out, ensure_ascii=False, indent=2))
    return 0


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