Spaces:
Running
Running
| """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"] | |