QuantEngineByItz/tools/build_time_stop_forecast_v1.py

"""build_time_stop_forecast_v1.py — P1-1: TIME_STOP 사전 발동 예측

60일 TIME_STOP 기준에 가까운 종목을 사전에 탐지하여 운영자가
7월 발동 전에 포지션 검토를 할 수 있도록 예측 보고서를 생성한다.

formula_id: BUILD_TIME_STOP_FORECAST_V1
contract_ref: spec/exit/stop_loss.yaml (TIME_STOP 60일 기준)
              spec/54_temporal_data_integrity.yaml
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, date, timedelta
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

DEFAULT_HARNESS = ROOT / "Temp" / "computed_harness_v1.json"
DEFAULT_DATA    = ROOT / "GatherTradingData.json"
OUTPUT_PATH     = ROOT / "Temp" / "time_stop_forecast_v1.json"

TIME_STOP_HOLD_DAYS = 60  # 60일 기준 (spec/exit/stop_loss.yaml)
FORECAST_WINDOW_DAYS = 30  # 앞으로 30일 내 발동 예상 종목 탐지


def _load_json(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _parse_date(s: str | None) -> date | None:
    if not s:
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(s, fmt).date()
        except ValueError:
            continue
    return None


def _forecast_from_harness(harness: dict, today: date) -> list[dict]:
    """Parse relative_stop_signal_json for TIME_STOP candidates."""
    forecasts = []
    signals = harness.get("relative_stop_signal_json") or []
    if not isinstance(signals, list):
        return forecasts

    for sig in signals:
        if not isinstance(sig, dict):
            continue
        details  = sig.get("details") or {}
        hold_days = details.get("hold_days")
        excess    = details.get("excess_ret20d")
        entry_date_str = details.get("entry_date")

        if hold_days is None:
            continue

        hold_days = int(hold_days)
        days_to_trigger = TIME_STOP_HOLD_DAYS - hold_days

        if days_to_trigger <= FORECAST_WINDOW_DAYS:
            trigger_date = (
                (_parse_date(entry_date_str) + timedelta(days=TIME_STOP_HOLD_DAYS))
                if entry_date_str else None
            )
            forecasts.append({
                "ticker":           sig.get("ticker"),
                "name":             sig.get("name"),
                "current_hold_days": hold_days,
                "days_to_trigger":  max(0, days_to_trigger),
                "projected_trigger_date": str(trigger_date) if trigger_date else "UNKNOWN",
                "excess_ret20d":    excess,
                "current_signal":   sig.get("signal"),
                "time_stop_condition": excess is not None and float(excess) < 0,
                "action_required":  days_to_trigger <= 7,
                "note": (
                    "TIME_STOP 발동 임박 (7일 이내)" if days_to_trigger <= 7
                    else f"TIME_STOP {days_to_trigger}일 후 예상"
                ),
            })

    return sorted(forecasts, key=lambda x: x["days_to_trigger"])


def _forecast_from_data(data_json: dict, today: date) -> list[dict]:
    """Fallback: parse account_snapshot rows for hold_days."""
    forecasts = []
    rows = data_json.get("account_snapshot") or []
    if isinstance(rows, dict):
        rows = rows.get("rows") or []
    if not isinstance(rows, list):
        return forecasts

    for row in rows:
        if not isinstance(row, dict):
            continue
        entry_date_str = row.get("entry_date") or row.get("EntryDate")
        entry_date = _parse_date(entry_date_str)
        if not entry_date:
            continue

        hold_days = (today - entry_date).days
        days_to_trigger = TIME_STOP_HOLD_DAYS - hold_days

        if 0 <= days_to_trigger <= FORECAST_WINDOW_DAYS:
            trigger_date = entry_date + timedelta(days=TIME_STOP_HOLD_DAYS)
            forecasts.append({
                "ticker":            row.get("ticker") or row.get("code"),
                "name":              row.get("name") or row.get("종목명"),
                "current_hold_days": hold_days,
                "days_to_trigger":   days_to_trigger,
                "projected_trigger_date": str(trigger_date),
                "excess_ret20d":     None,
                "current_signal":    "DATA_MISSING",
                "time_stop_condition": None,
                "action_required":   days_to_trigger <= 7,
                "note": (
                    "TIME_STOP 발동 임박 (7일 이내)" if days_to_trigger <= 7
                    else f"TIME_STOP {days_to_trigger}일 후 예상"
                ),
            })

    return sorted(forecasts, key=lambda x: x["days_to_trigger"])


def run(harness_path: Path, data_path: Path) -> dict:
    today   = date.today()
    harness = _load_json(harness_path)
    data    = _load_json(data_path)

    forecasts = []
    if not harness.get("_missing"):
        forecasts = _forecast_from_harness(harness, today)

    if not forecasts and not data.get("_missing"):
        forecasts = _forecast_from_data(data, today)

    imminent  = [f for f in forecasts if f["days_to_trigger"] <= 7]
    upcoming  = [f for f in forecasts if 7 < f["days_to_trigger"] <= 30]
    triggered = [f for f in forecasts if f.get("current_signal") == "TIME_STOP"]

    gate = "PASS"
    if imminent:
        gate = "WARN"
    if triggered:
        gate = "TRIGGERED"

    result = {
        "gate": gate,
        "as_of_date": str(today),
        "time_stop_hold_days_threshold": TIME_STOP_HOLD_DAYS,
        "forecast_window_days": FORECAST_WINDOW_DAYS,
        "imminent_count": len(imminent),
        "upcoming_count": len(upcoming),
        "triggered_count": len(triggered),
        "imminent_tickers": imminent,
        "upcoming_tickers": upcoming,
        "triggered_tickers": triggered,
        "summary": (
            f"TODAY={today}: "
            f"{len(triggered)}건 발동중, "
            f"{len(imminent)}건 7일내 임박, "
            f"{len(upcoming)}건 30일내 예정"
        ),
        "action_note": (
            "임박 종목 포지션 재검토 필요 — excess_ret20d 추이 모니터링 권장"
            if (imminent or triggered) else
            "현재 30일내 발동 예상 종목 없음"
        ),
        "contract": "spec/54_temporal_data_integrity.yaml",
    }

    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
    return result


def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description="TIME_STOP Forecast Builder")
    parser.add_argument("--harness", default=str(DEFAULT_HARNESS))
    parser.add_argument("--data",    default=str(DEFAULT_DATA))
    args = parser.parse_args()

    result = run(Path(args.harness), Path(args.data))
    gate = result.get("gate", "PASS")
    print(f"[BUILD_TIME_STOP_FORECAST_V1] gate={gate}")
    print(f"  {result.get('summary')}")
    if result.get("imminent_tickers"):
        print("  임박 종목:")
        for t in result["imminent_tickers"]:
            print(f"    {t['ticker']} {t.get('name','')} — {t['days_to_trigger']}일후 ({t['projected_trigger_date']}) excess={t.get('excess_ret20d')}")
    if result.get("triggered_tickers"):
        print("  발동중:")
        for t in result["triggered_tickers"]:
            print(f"    {t['ticker']} {t.get('name','')} — hold={t['current_hold_days']}일")


if __name__ == "__main__":
    main()