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