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