"""Slipstream supervisor — daily health & drift monitor (deterministic, flag-only). Ported from DeepQuant's P4 supervisor, adapted to Slipstream's monthly cadence and caps. Three checks, each OK / WATCH / ALERT — it NEVER trades: 1. tracking — trailing ~4-week fund-vs-SPY gap. No weekly backtest distribution exists yet for this strategy (only summary stats), so v1 uses static bands: WATCH < -3pp, ALERT < -6pp. (Rule v2's 2014–2026 backtest: excess return +6.0pp/yr, alpha +3.9pp/yr at β 1.17 — a -6pp MONTH is a genuine outlier against a ~+0.5pp/month edge.) Upgrade path: replace bands with the percentile envelope once a weekly walk-forward curve is produced. 2. guardrails— paper-endpoint invariant (ALERT if live trading is enabled), no margin (cash >= -0.2% of equity — the executor now keeps ~1% cash, so any negative balance is a fault), gross exposure <= 100.5%, and position drift: the 15% name cap applies AT rebalance, so intra-month drift is expected — WATCH above cap+3pp (18%), ALERT above cap+7pp (22%). 3. freshness — signals bundle (data/meta.json as_of) <= 10 days old (weekly refresh + slack; the regime test pulls SPY live so it is safe even when the bundle lags); last rebalance order <= 35 days (monthly cadence + slack). GET /api/supervisor returns the cached report (recomputed if older than TTL or force=1). """ from __future__ import annotations import datetime as dt import json import threading import time from pathlib import Path import alpaca_client as alp import config from book import MAX_POS_W _TTL_S = 6 * 3600 _lock = threading.Lock() _cache: dict = {"ts": 0.0, "report": None} _DATA = Path(__file__).resolve().parent.parent / "data" def _check_tracking() -> dict: name = "Tracking vs S&P 500" try: h = alp.nav_history(period="1M", timeframe="1D") pts = [(dt.datetime.fromtimestamp(pt["t"], dt.timezone.utc).date(), pt["equity"]) for pt in (h.get("points") or []) if pt.get("equity")] pts = [p for p in pts if (pts[-1][0] - p[0]).days <= 28] if len(pts) < 5: return {"name": name, "status": "WATCH", "detail": "not enough NAV history yet (<5 days in window)"} fund = pts[-1][1] / pts[0][1] - 1.0 spy_rows = alp.spy_series(start=str(pts[0][0])) if len(spy_rows) < 2: return {"name": name, "status": "WATCH", "detail": "could not fetch SPY series"} spy = spy_rows[-1]["c"] / spy_rows[0]["c"] - 1.0 gap = fund - spy st = "ALERT" if gap < -0.06 else ("WATCH" if gap < -0.03 else "OK") return {"name": name, "status": st, "detail": (f"trailing-4wk gap {gap*100:+.1f}pp (fund {fund*100:+.1f}% vs SPY {spy*100:+.1f}%) — " f"static bands (WATCH <-3pp, ALERT <-6pp); rule-v2 backtest excess ~+6.0pp/yr (2014–2026)")} except Exception as e: # noqa: BLE001 return {"name": name, "status": "WATCH", "detail": f"could not compute ({type(e).__name__}: {str(e)[:60]})"} def _check_guardrails() -> dict: name = "Guardrails (paper / margin / exposure / caps)" try: problems, worst = [], "OK" def esc(level): nonlocal worst order = {"OK": 0, "WATCH": 1, "ALERT": 2} if order[level] > order[worst]: worst = level if getattr(config, "ALPACA_ALLOW_LIVE", False): problems.append("ALPACA_ALLOW_LIVE is enabled — fund must stay paper-only"); esc("ALERT") acct = alp.account() equity = float(acct.get("equity") or 0.0) cash = float(acct.get("cash") or 0.0) if equity > 0 and cash < -0.002 * equity: problems.append(f"negative cash {cash:,.0f} beyond -0.2% of equity (margin in use)"); esc("ALERT") pos = alp.positions() gross = sum(abs(float(p.get("market_value") or 0.0)) for p in pos) if equity > 0 and gross > 1.005 * equity: problems.append(f"gross exposure {gross/equity*100:.1f}% > 100.5%"); esc("ALERT") if equity > 0 and pos: w = {p["symbol"]: abs(float(p.get("market_value") or 0.0)) / equity for p in pos} top_sym = max(w, key=w.get); top_w = w[top_sym] if top_w > MAX_POS_W + 0.07: problems.append(f"{top_sym} at {top_w*100:.1f}% — far above the {MAX_POS_W*100:.0f}% rebalance cap"); esc("ALERT") elif top_w > MAX_POS_W + 0.03: problems.append(f"{top_sym} drifted to {top_w*100:.1f}% (cap {MAX_POS_W*100:.0f}% applies at rebalance)"); esc("WATCH") detail = "; ".join(problems) if problems else ( f"paper-only ✓ · cash {cash:,.0f} · exposure {gross/equity*100:.1f}% · " f"{len(pos)} positions, largest within drift tolerance") return {"name": name, "status": worst, "detail": detail} except Exception as e: # noqa: BLE001 return {"name": name, "status": "WATCH", "detail": f"could not compute ({type(e).__name__}: {str(e)[:60]})"} def _check_freshness() -> dict: name = "Data & rebalance freshness" try: today = dt.date.today() problems, worst = [], "OK" def esc(level): nonlocal worst order = {"OK": 0, "WATCH": 1, "ALERT": 2} if order[level] > order[worst]: worst = level meta = json.loads((_DATA / "meta.json").read_text()) as_of = dt.date.fromisoformat(str(meta.get("as_of"))) age = (today - as_of).days if age > 14: problems.append(f"signals bundle {age}d old (as_of {as_of}) — weekly refresh has missed 2+ cycles"); esc("ALERT") elif age > 10: problems.append(f"signals bundle {age}d old (as_of {as_of}) — weekly refresh may have failed"); esc("WATCH") last_reb = alp.last_order_date() if last_reb is None: problems.append("no orders found — fund has never rebalanced"); esc("WATCH") else: if isinstance(last_reb, str): last_reb = dt.date.fromisoformat(last_reb[:10]) elif isinstance(last_reb, dt.datetime): last_reb = last_reb.date() rb_age = (today - last_reb).days if rb_age > 45: problems.append(f"last rebalance {rb_age}d ago — monthly cron has missed a cycle"); esc("ALERT") elif rb_age > 35: problems.append(f"last rebalance {rb_age}d ago (monthly cadence + slack is 35d)"); esc("WATCH") detail = "; ".join(problems) if problems else ( f"signals as_of {as_of} ({age}d) ✓ · last rebalance " f"{last_reb if last_reb else 'n/a'} ✓") return {"name": name, "status": worst, "detail": detail} except Exception as e: # noqa: BLE001 return {"name": name, "status": "WATCH", "detail": f"could not compute ({type(e).__name__}: {str(e)[:60]})"} def run_checks() -> dict: checks = [_check_tracking(), _check_guardrails(), _check_freshness()] order = {"OK": 0, "WATCH": 1, "ALERT": 2} overall = max((c["status"] for c in checks), key=lambda s: order.get(s, 1)) return {"overall": overall, "checks": checks, "computed_at": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), "note": "flag-only monitor — it never trades"} def report(force: bool = False) -> dict: with _lock: if force or _cache["report"] is None or (time.time() - _cache["ts"]) > _TTL_S: _cache["report"] = run_checks() _cache["ts"] = time.time() return _cache["report"]