From 9287a92461902e8b2a407fbf7fe3f11fe20f5e7b Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 12 Oct 2026 07:26:33 -0700 Subject: [PATCH] research: low_svr21 combination study (STOP); daily option-chain snapshots svr_combo (preregistered): blending low_svr21 into overnight_recovery_top10 lifts Sharpe 0.81 -> 0.87 at w=1/3 but deepens MaxDD by 6pp (G1, G3 fail); as a fourth sleeve it dilutes the overnight book (1.49 -> 1.42). fetch_options collects S&P 500 option chains after each close from Alpaca's free indicative feed (quotes, day volume, OCC open interest), solves IVs, and writes per-name iv_atm30 / skew / call-put spread / option-stock volume summaries to data/options/us/. Forward-only history for options-implied factors. Run by the quant-trading-options timer at 16:25 New York, Mon-Fri; it skips once the next session has opened. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 +- deploy/start-options-snapshot.sh | 14 + deploy/systemd/quant-trading-options.service | 10 + deploy/systemd/quant-trading-options.timer | 9 + research/lab/fetch_options.py | 404 +++++++++++++++++++ research/lab/svr_combo.py | 200 +++++++++ 6 files changed, 642 insertions(+), 1 deletion(-) create mode 100755 deploy/start-options-snapshot.sh create mode 100644 deploy/systemd/quant-trading-options.service create mode 100644 deploy/systemd/quant-trading-options.timer create mode 100644 research/lab/fetch_options.py create mode 100644 research/lab/svr_combo.py diff --git a/CLAUDE.md b/CLAUDE.md index a2693b1..d50fb0b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -109,6 +109,8 @@ PYTHONPATH=. uv run python -m research.lab.overnight_midsmall # the overni PYTHONPATH=. uv run --with numba python -m research.lab.dip_buying # buy-the-dip: SPY/QQQ RSI(2), trend-filtered stock RSI(2), drawdown tables (preregistered; STOP 2026-10-12: SPY 46 bp/trade t 3.4 but 14% of days, 3x Sharpe 0.57 = SPY; stocks breakeven 10 bp/side) PYTHONPATH=. uv run python -m research.lab.dip_leverage_overlay # +1x SPY only inside SPY RSI(2) dip windows, on the overnight_book books (preregistered; STOP 2026-10-12: Sharpe 0.98->0.99 / 1.38->1.33, placebo rank 77-88%; the books already earn the bounce at ~2x SPY beta) PYTHONPATH=. uv run --with numba python -m research.lab.factor_census [--part construction] # every lab factor (price, EDGAR, insider, Lazy Prices, EAR, SVR, FTD) measured one way on the current PIT panel: IC, quintile spread, top-20 alpha vs EW (2026-10-12: 1 A low_svr21, 2 B accruals/rev_5, 69 N; no |t|>=3) +PYTHONPATH=. uv run --with numba python -m research.lab.svr_combo # low_svr21 blended into overnight_recovery_top10 / as a fourth book sleeve, 2018-26 (preregistered; STOP 2026-10-12: blend Sharpe 0.81->0.87 but MaxDD -35%->-41%; book 1.49->1.42) +PYTHONPATH=. uv run python -m research.lab.fetch_options snapshot|status # daily S&P 500 option chains after the close (Alpaca indicative feed, forward-only from 2026-10-12; data/options/us/; timer quant-trading-options) uv run python -m research.lab.fetch_us_fundamentals --universe mid # EDGAR PIT facts for S&P 400 ever-members (name-checked CIKs) uv run python -m research.lab.fetch_us_fundamentals --universe small # the same for S&P 600 (IJR) PYTHONPATH=. uv run python -m research.lab.midcap_sleeve [--only h1..h4] [--universe small] # S&P 400 sleeves vs IJH; S&P 600 replication vs IJR (all REJECT 2026-10-08) @@ -210,7 +212,7 @@ The comparison writes its fixed budget and acceptance rule before running and re ## Server Deployment -Four `systemd --user` services, defined in `deploy/systemd/`, run from `/home/gahow/projects/quant-trading`: +Four `systemd --user` services and one timer, defined in `deploy/systemd/`, run from `/home/gahow/projects/quant-trading`: | Service | Script | Role | |---|---|---| @@ -218,11 +220,13 @@ Four `systemd --user` services, defined in `deploy/systemd/`, run from `/home/ga | `quant-trading-api` | `deploy/start-web-api.sh` | Web API on `127.0.0.1:8787` | | `quant-trading-vite` | `deploy/start-web-vite.sh` | Vite on `127.0.0.1:5174`, behind nginx on `:5443` | | `quant-trading-overnight-paper` | `deploy/start-overnight-paper.sh` | `overnight_paper.py daemon`: overnight-only momentum on the Alpaca paper account (from 2026-10-12) | +| `quant-trading-options` (timer, 16:25 New York Mon-Fri) | `deploy/start-options-snapshot.sh` | `research.lab.fetch_options snapshot`: S&P 500 option chains + per-name IV/skew/spread summary into `data/options/us/` (~11 MB/day) | ```bash cp deploy/systemd/*.service ~/.config/systemd/user/ systemctl --user daemon-reload systemctl --user enable --now quant-trading-monitor quant-trading-api quant-trading-vite +systemctl --user enable --now quant-trading-options.timer journalctl --user -u quant-trading-monitor -f ``` diff --git a/deploy/start-options-snapshot.sh b/deploy/start-options-snapshot.sh new file mode 100755 index 0000000..2596969 --- /dev/null +++ b/deploy/start-options-snapshot.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Daily S&P 500 option-chain snapshot after the US close (research/lab/fetch_options.py). +# Started by quant-trading-options.timer (16:25 America/New_York, Mon-Fri). +# Keys are read from the repo-root .env by the script itself; Alpaca is reached directly (no proxy). +set -euo pipefail + +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +PROJECT_ROOT="$( cd "$SCRIPT_DIR/.." && pwd )" +cd "$PROJECT_ROOT" + +export PATH="/home/gahow/.local/bin:$PATH" +export PYTHONPATH=. + +exec uv run python -m research.lab.fetch_options snapshot diff --git a/deploy/systemd/quant-trading-options.service b/deploy/systemd/quant-trading-options.service new file mode 100644 index 0000000..d6202c8 --- /dev/null +++ b/deploy/systemd/quant-trading-options.service @@ -0,0 +1,10 @@ +[Unit] +Description=Daily S&P 500 option-chain snapshot (Alpaca indicative feed) +After=network-online.target + +[Service] +Type=oneshot +WorkingDirectory=/home/gahow/projects/quant-trading +ExecStart=/home/gahow/projects/quant-trading/deploy/start-options-snapshot.sh +TimeoutStartSec=3600 +Environment=PYTHONUNBUFFERED=1 diff --git a/deploy/systemd/quant-trading-options.timer b/deploy/systemd/quant-trading-options.timer new file mode 100644 index 0000000..e661553 --- /dev/null +++ b/deploy/systemd/quant-trading-options.timer @@ -0,0 +1,9 @@ +[Unit] +Description=Option-chain snapshot after the US close + +[Timer] +OnCalendar=Mon..Fri 16:25 America/New_York +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/research/lab/fetch_options.py b/research/lab/fetch_options.py new file mode 100644 index 0000000..d39e0e7 --- /dev/null +++ b/research/lab/fetch_options.py @@ -0,0 +1,404 @@ +""" +Daily option-chain snapshots for the S&P 500, collected forward from +2026-10-12. Options-implied stock factors have no free history, so the +record starts with this collector: + - IV skew (Xing, Zhang & Zhao 2010) + - the call-put IV spread (Cremers & Weinbaum 2010) + - option/stock volume (Johnson & So 2012) + - implied-vol level and changes + + uv run python -m research.lab.fetch_options snapshot # after the close (systemd timer) + uv run python -m research.lab.fetch_options snapshot --limit 5 --force --out /tmp/x # smoke test + uv run python -m research.lab.fetch_options status + +Sources (keys ALPACA_API_KEY / ALPACA_API_SECRET from .env, never logged): + - stock close and volume: Alpaca SIP daily bars, unadjusted, so they + match the strikes. The free plan serves bars older than 15 minutes. + - option quotes: /v1beta1/options/snapshots/{underlying}, feed=indicative. + This is Alpaca's free derived OPRA feed. It gives quote, last trade, + today's bar, and Alpaca's own IV and greeks. The real OPRA feed returns + 403 until an OPRA agreement is signed. + - open interest: /v2/options/contracts (OCC, as of the prior day). + The filter keeps expiries 5-100 calendar days out and strikes at + 0.7-1.3 x the close. The universe is data/universe_sp500.json plus SPY, + QQQ and IWM. + +When: 16:25 America/New_York, Mon-Fri (deploy/systemd/quant-trading-options.timer). +The Alpaca calendar decides the session. The script records the latest +session that closed at least 10 minutes ago. It skips the run if the next +session has already opened (the quotes would belong to another day) and if +the session's file exists, unless --force. + +Files (data/options/us/): + chains/YYYY/chain_.csv.gz one row per contract: quote, last trade, + day volume, OI, Alpaca IV/greeks, stock + close/volume, r, implied forward F, and + two IVs from this script's own solver + summary/summary_.csv one row per underlying: iv_atm30, + skew_xzz, cp_spread, os_volume, + pc_volume, pc_oi, quote quality + manifest.jsonl one line per run +The IVs are European. iv_fwd is Black-76 on the forward implied by put-call +parity at the three strikes nearest the close, so dividends drop out. iv_spot +is Black-Scholes on the close with no dividends. cp_spread uses iv_spot, +which is what Cremers-Weinbaum measure, so a dividend payer shows a negative +spread of about its yield; adjust for that when testing. American early +exercise is ignored, so the summary uses near-the-money strikes. The raw +chains are kept so a better solver can be run later. + +NEEDS EVIDENCE: whether the indicative quotes match the OPRA NBBO closely +enough for skew and spread. Compare a sample against another source before +the first factor test. A cross-sectional test needs about 2-3 years of +snapshots. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +import time +from zoneinfo import ZoneInfo + +import numpy as np +import pandas as pd +import requests +from scipy.special import ndtr + +from research.lab.fetch_alpaca import _keys + +NY = ZoneInfo("America/New_York") +DATA = "https://data.alpaca.markets" +TRADE = "https://paper-api.alpaca.markets" +OUT = os.path.join("data", "options", "us") +UNIVERSE = os.path.join("data", "universe_sp500.json") +RF_PATH = os.path.join("data", "research", "ext", "fred_dtb3.csv") +EXTRA = ["SPY", "QQQ", "IWM"] +DTE_MIN, DTE_MAX = 5, 100 +MNY_LO, MNY_HI = 0.7, 1.3 +MIN_INTERVAL = 0.32 # <= ~190 requests a minute (limit 200) + + +class Api: + def __init__(self) -> None: + self.s = requests.Session() + self.s.headers.update(_keys()) + self.next = 0.0 + self.n = 0 + + def get(self, url: str, params: dict) -> dict: + for attempt in range(6): + wait = self.next - time.time() + if wait > 0: + time.sleep(wait) + self.next = time.time() + MIN_INTERVAL + try: + r = self.s.get(url, params=params, timeout=60) + except requests.RequestException: + time.sleep(2 ** attempt) + continue + self.n += 1 + if r.status_code == 429 or r.status_code >= 500: + time.sleep(2 ** attempt) + continue + if r.status_code != 200: + raise RuntimeError(f"HTTP {r.status_code} {r.text[:200]}") + return r.json() + raise RuntimeError("retries exhausted") + + def pages(self, url: str, params: dict, key: str): + params = dict(params) + while True: + j = self.get(url, params) + yield j.get(key) or ({} if key in ("snapshots", "bars") else []) + tok = j.get("next_page_token") + if not tok: + return + params["page_token"] = tok + + +# --------------------------------------------------------------------------- +# session and inputs +# --------------------------------------------------------------------------- + +def pick_session(api: Api, now: dt.datetime, force: bool) -> dt.date | None: + today = now.astimezone(NY).date() + cal = api.get(f"{TRADE}/v2/calendar", {"start": (today - dt.timedelta(days=10)).isoformat(), + "end": (today + dt.timedelta(days=10)).isoformat()}) + days = [] + for c in cal: + d = dt.date.fromisoformat(c["date"]) + at = lambda hm: dt.datetime.combine(d, dt.time(int(hm[:2]), int(hm[3:5])), NY) # noqa: E731 + days.append((d, at(c["open"]), at(c["close"]))) + done = [x for x in days if x[2] + dt.timedelta(minutes=10) <= now] + if not done: + print("skip: no session in the calendar window has closed") + return None + d, _, _ = done[-1] + nxt = [x for x in days if x[0] > d] + if nxt and now >= nxt[0][1] and not force: + print(f"skip: session {d} is over and {nxt[0][0]} has opened; quotes would be from another day") + return None + return d + + +def universe() -> list[str]: + with open(UNIVERSE) as fh: + ticks = json.load(fh)["tickers"] + return sorted({t.replace("-", ".") for t in ticks} | set(EXTRA)) + + +def stock_bars(api: Api, syms: list[str], session: dt.date, now: dt.datetime) -> dict[str, dict]: + end = min(now - dt.timedelta(minutes=16), dt.datetime.combine(session + dt.timedelta(days=1), dt.time(0), NY)) + out = {} + for i in range(0, len(syms), 100): + params = {"symbols": ",".join(syms[i:i + 100]), "timeframe": "1Day", "start": session.isoformat(), + "end": end.isoformat(), "feed": "sip", "adjustment": "raw", "limit": 10000} + for bars in api.pages(f"{DATA}/v2/stocks/bars", params, "bars"): + for sym, bl in bars.items(): + for b in bl: + if b["t"][:10] == session.isoformat(): + out[sym] = {"close": b["c"], "volume": b["v"]} + return out + + +def open_interest(api: Api, syms: list[str], session: dt.date) -> dict[str, dict]: + out = {} + lo, hi = session + dt.timedelta(days=DTE_MIN), session + dt.timedelta(days=DTE_MAX) + for i in range(0, len(syms), 10): + params = {"underlying_symbols": ",".join(syms[i:i + 10]), "expiration_date_gte": lo.isoformat(), + "expiration_date_lte": hi.isoformat(), "limit": 10000} + try: + for page in api.pages(f"{TRADE}/v2/options/contracts", params, "option_contracts"): + for c in page: + out[c["symbol"]] = {"oi": float(c["open_interest"]) if c.get("open_interest") else np.nan, + "oi_date": c.get("open_interest_date")} + except RuntimeError as exc: + print(f"open interest {syms[i]}..: {exc}", flush=True) + return out + + +def chain(api: Api, sym: str, close: float, session: dt.date) -> list[dict]: + params = {"feed": "indicative", "limit": 1000, + "expiration_date_gte": (session + dt.timedelta(days=DTE_MIN)).isoformat(), + "expiration_date_lte": (session + dt.timedelta(days=DTE_MAX)).isoformat(), + "strike_price_gte": round(MNY_LO * close, 2), "strike_price_lte": round(MNY_HI * close, 2)} + rows = [] + for snaps in api.pages(f"{DATA}/v1beta1/options/snapshots/{sym}", params, "snapshots"): + for occ, s in snaps.items(): + q, t, b, g = s.get("latestQuote") or {}, s.get("latestTrade") or {}, s.get("dailyBar") or {}, s.get("greeks") or {} + today = b.get("t", "")[:10] == session.isoformat() + rows.append({"underlying": sym, "symbol": occ, "type": occ[-9], + "expiry": f"20{occ[-15:-13]}-{occ[-13:-11]}-{occ[-11:-9]}", + "strike": int(occ[-8:]) / 1000, "bid": q.get("bp"), "ask": q.get("ap"), + "bid_size": q.get("bs"), "ask_size": q.get("as"), "quote_time": q.get("t"), + "last": t.get("p"), "last_time": t.get("t"), "day_volume": b.get("v", 0) if today else 0, + "alp_iv": s.get("impliedVolatility"), "delta": g.get("delta"), "gamma": g.get("gamma"), + "theta": g.get("theta"), "vega": g.get("vega")}) + return rows + + +def rate(session: dt.date) -> tuple[float, str]: + try: + rf = pd.read_csv(RF_PATH, index_col=0, parse_dates=True).iloc[:, 0] + rf = pd.to_numeric(rf, errors="coerce").dropna().loc[:pd.Timestamp(session)] + return float(rf.iloc[-1]) / 100, rf.index[-1].date().isoformat() + except (OSError, IndexError): + return 0.04, "default" + + +# --------------------------------------------------------------------------- +# implied volatility and per-underlying summary +# --------------------------------------------------------------------------- + +def black76(F, K, T, sig, call, disc): + sd = sig * np.sqrt(T) + d1 = (np.log(F / K) + 0.5 * sd ** 2) / sd + d2 = d1 - sd + return np.where(call, disc * (F * ndtr(d1) - K * ndtr(d2)), disc * (K * ndtr(-d2) - F * ndtr(-d1))) + + +def implied_vol(price, F, K, T, call, disc, iters: int = 80) -> np.ndarray: + lo, hi = np.full(len(price), 1e-4), np.full(len(price), 5.0) + intrinsic = disc * np.where(call, np.maximum(F - K, 0), np.maximum(K - F, 0)) + ok = np.isfinite(price) & (price > intrinsic + 1e-6) & (price < black76(F, K, T, hi, call, disc)) + for _ in range(iters): + mid = (lo + hi) / 2 + above = black76(F, K, T, mid, call, disc) > price + hi, lo = np.where(above, mid, hi), np.where(above, lo, mid) + return np.where(ok, (lo + hi) / 2, np.nan) + + +def add_iv(df: pd.DataFrame, r: float) -> pd.DataFrame: + df["mid"] = (df["bid"] + df["ask"]) / 2 + good = (df["bid"] > 0) & (df["ask"] > df["bid"]) + df.loc[~good, "mid"] = np.nan + df["T"] = df["dte"] / 365.0 + disc = np.exp(-r * df["T"]) + df["fwd"] = np.nan + for (_, _), g in df.groupby(["underlying", "expiry"]): + piv = g.pivot_table(index="strike", columns="type", values="mid") + if {"C", "P"} <= set(piv.columns): + piv = piv.dropna() + if len(piv): + S = g["stock_close"].iloc[0] + near = piv.iloc[np.argsort(np.abs(piv.index.to_numpy() - S))[:3]] + f = near.index.to_numpy() + np.exp(r * g["T"].iloc[0]) * (near["C"] - near["P"]).to_numpy() + df.loc[g.index, "fwd"] = float(np.median(f)) + nofwd = df["fwd"].isna() + df.loc[nofwd, "fwd"] = df.loc[nofwd, "stock_close"] * np.exp(r * df.loc[nofwd, "T"]) + call = (df["type"] == "C").to_numpy() + args = (df["strike"].to_numpy(), df["T"].to_numpy(), call, disc.to_numpy()) + df["iv_fwd"] = implied_vol(df["mid"].to_numpy(), df["fwd"].to_numpy(), *args) + spot_f = (df["stock_close"] * np.exp(r * df["T"])).to_numpy() + df["iv_spot"] = implied_vol(df["mid"].to_numpy(), spot_f, *args) + return df + + +def summarize(g: pd.DataFrame) -> dict: + S = g["stock_close"].iloc[0] + v = g[g["iv_fwd"].notna()] + out = {"underlying": g["underlying"].iloc[0], "stock_close": S, "stock_volume": g["stock_volume"].iloc[0], + "n_contracts": len(g), "n_valid": len(v), + "med_rel_spread": float(((v["ask"] - v["bid"]) / v["mid"]).median()) if len(v) else np.nan} + atm = {} + for e, x in v.groupby("expiry"): + F = x["fwd"].iloc[0] + k = x["strike"].iloc[np.argmin(np.abs(x["strike"].to_numpy() - F))] + atm[int(x["dte"].iloc[0])] = float(x.loc[x["strike"] == k, "iv_fwd"].mean()) + ds = sorted(d for d in atm if 7 <= d <= DTE_MAX) + if ds: + lo = [d for d in ds if d <= 30] + hi = [d for d in ds if d >= 30] + if lo and hi and lo[-1] != hi[0]: # total variance, linear in time + a, b = lo[-1], hi[0] + w = atm[a] ** 2 * a + (atm[b] ** 2 * b - atm[a] ** 2 * a) * (30 - a) / (b - a) + out["iv_atm30"] = float(np.sqrt(w / 30)) if w > 0 else np.nan + else: # 30 days exactly, or only one side: nearest + out["iv_atm30"] = atm[hi[0]] if hi else atm[lo[-1]] + mid_exp = v[(v["dte"] >= 10) & (v["dte"] <= 60)] + if len(mid_exp): + e = mid_exp.loc[(mid_exp["dte"] - 30).abs().idxmin(), "expiry"] + x = mid_exp[mid_exp["expiry"] == e] + m = x["strike"] / S + puts = x[(x["type"] == "P") & (m >= 0.80) & (m <= 0.95)] + calls = x[(x["type"] == "C") & (m >= 0.95) & (m <= 1.05)] + if len(puts) and len(calls): + p = puts.loc[(puts["strike"] / S - 0.95).abs().idxmin(), "iv_fwd"] + c = calls.loc[(calls["strike"] / S - 1.0).abs().idxmin(), "iv_fwd"] + out["skew_xzz"] = float(p - c) + pairs = mid_exp[(mid_exp["strike"] / S).between(0.8, 1.2) & mid_exp["iv_spot"].notna()] + pv = pairs.pivot_table(index=["expiry", "strike"], columns="type", values=["iv_spot", "oi"]) + if ("iv_spot", "C") in pv.columns and ("iv_spot", "P") in pv.columns: + pv = pv.dropna(subset=[("iv_spot", "C"), ("iv_spot", "P")]) + if len(pv): + w = pv[("oi", "C")].fillna(0) + pv[("oi", "P")].fillna(0) if ("oi", "C") in pv and ("oi", "P") in pv \ + else pd.Series(1.0, index=pv.index) + w = w if w.sum() > 0 else pd.Series(1.0, index=pv.index) + out["cp_spread"] = float(((pv[("iv_spot", "C")] - pv[("iv_spot", "P")]) * w).sum() / w.sum()) + cv, pv_ = g.loc[g["type"] == "C", "day_volume"].sum(), g.loc[g["type"] == "P", "day_volume"].sum() + co, po = g.loc[g["type"] == "C", "oi"].sum(), g.loc[g["type"] == "P", "oi"].sum() + out.update({"call_volume": cv, "put_volume": pv_, "call_oi": co, "put_oi": po, + "pc_volume": pv_ / cv if cv else np.nan, "pc_oi": po / co if co else np.nan, + "os_volume": (cv + pv_) * 100 / out["stock_volume"] if out["stock_volume"] else np.nan}) + return out + + +# --------------------------------------------------------------------------- +# commands +# --------------------------------------------------------------------------- + +def snapshot(args) -> None: + out = args.out or OUT + now = dt.datetime.now(NY) + api = Api() + session = dt.date.fromisoformat(args.session) if args.session else pick_session(api, now, args.force) + if session is None: + return + path = os.path.join(out, "chains", str(session.year), f"chain_{session}.csv.gz") + if os.path.exists(path) and not args.force: + print(f"{path} exists; nothing to do") + return + syms = universe()[: args.limit] if args.limit else universe() + t0 = time.time() + px = stock_bars(api, syms, session, now) + oi = open_interest(api, [s for s in syms if s in px], session) + rows, failed = [], [] + for k, sym in enumerate(syms): + if sym not in px: + failed.append([sym, "no stock bar"]) + continue + try: + rs = chain(api, sym, px[sym]["close"], session) + except RuntimeError as exc: + failed.append([sym, str(exc)[:120]]) + continue + for r in rs: + r.update(stock_close=px[sym]["close"], stock_volume=px[sym]["volume"], **oi.get(r["symbol"], {})) + rows += rs + if (k + 1) % 50 == 0: + print(f" {k + 1}/{len(syms)} underlyings, {len(rows)} contracts, {api.n} requests " + f"[{time.time() - t0:.0f}s]", flush=True) + if not rows: + raise SystemExit("no contracts collected") + df = pd.DataFrame(rows) + for c in ("oi", "oi_date"): + if c not in df: + df[c] = np.nan + df.insert(0, "session", session.isoformat()) + df["dte"] = (pd.to_datetime(df["expiry"]) - pd.Timestamp(session)).dt.days + r, r_date = rate(session) + df["r"] = r + df = add_iv(df, r) + os.makedirs(os.path.dirname(path), exist_ok=True) + df.drop(columns=["mid", "T"]).to_csv(path + ".tmp", index=False, compression="gzip") + os.replace(path + ".tmp", path) + summ = pd.DataFrame([summarize(g) for _, g in df.groupby("underlying")]) + summ.insert(0, "session", session.isoformat()) + spath = os.path.join(out, "summary", f"summary_{session}.csv") + os.makedirs(os.path.dirname(spath), exist_ok=True) + summ.to_csv(spath, index=False) + rec = {"session": session.isoformat(), "started": dt.datetime.fromtimestamp(t0, NY).isoformat(timespec="seconds"), + "finished": dt.datetime.now(NY).isoformat(timespec="seconds"), "underlyings": len(syms), + "with_chain": int(df["underlying"].nunique()), "contracts": len(df), + "valid_iv": int(df["iv_fwd"].notna().sum()), "requests": api.n, "r": r, "r_date": r_date, + "feed": "indicative", "failed": failed[:50], "n_failed": len(failed), "forced": bool(args.force)} + with open(os.path.join(out, "manifest.jsonl"), "a") as fh: + fh.write(json.dumps(rec) + "\n") + print(f"session {session}: {rec['with_chain']}/{len(syms)} underlyings, {len(df)} contracts " + f"({rec['valid_iv']} with IV), {api.n} requests, {time.time() - t0:.0f}s, failed {len(failed)} -> {path}") + + +def status(args) -> None: + out = args.out or OUT + man = os.path.join(out, "manifest.jsonl") + if not os.path.exists(man): + print("no snapshots yet") + return + recs = [json.loads(x) for x in open(man) if x.strip()] + print(f"{len(recs)} runs, {len({r['session'] for r in recs})} sessions, first {recs[0]['session']}, " + f"last {recs[-1]['session']}") + for r in recs[-10:]: + print(f" {r['session']} {r['with_chain']}/{r['underlyings']} underlyings {r['contracts']} contracts " + f"{r['valid_iv']} IV, failed {r['n_failed']}, {r['started']} -> {r['finished']}") + + +def main() -> None: + ap = argparse.ArgumentParser() + sub = ap.add_subparsers(dest="cmd", required=True) + s = sub.add_parser("snapshot") + s.add_argument("--limit", type=int, help="first N underlyings only (testing)") + s.add_argument("--force", action="store_true", help="record even if late or already recorded") + s.add_argument("--session", help="label the snapshot with this session date (testing)") + s.add_argument("--out", help="output directory (default data/options/us)") + s.set_defaults(fn=snapshot) + t = sub.add_parser("status") + t.add_argument("--out") + t.set_defaults(fn=status) + args = ap.parse_args() + args.fn(args) + + +if __name__ == "__main__": + main() diff --git a/research/lab/svr_combo.py b/research/lab/svr_combo.py new file mode 100644 index 0000000..9bef234 --- /dev/null +++ b/research/lab/svr_combo.py @@ -0,0 +1,200 @@ +""" +Does the low short-volume-ratio factor (low_svr21, the only class-A row of +research/lab/factor_census.py) improve the momentum strategy or the book? + + PYTHONPATH=. uv run --with numba python -m research.lab.svr_combo + +low_svr21 = -(21-day mean of FINRA REG SHO ShortVolume / TotalVolume), +shifted 2 sessions (the day-D file is published after D's close). + +Arms (research engine: PIT S&P 500, next-open fills, 10 bp a side, phase-averaged): + base overnight_mom + rec_126 ranks, top 10, every 10 days, buffer 2 + (the live overnight_recovery_top10) + blend_w (1 - w) x base score + w x low_svr21 rank, same construction; + w = 1/3 gated, w = 0.25 and 0.5 are the neighbourhood. A member + without FINRA data gets a neutral 0.5 SVR rank, so the universe + matches base + svr_sleeve low_svr21 top 20, every 21 days, buffer 2 (the census construction) +Books (overnight_book band_blend: equal shares, 25% band, month-start reset): + book3 overnight + lowrisk + trend_v7 (three_sleeve_overnight, the paper test) + book4 book3 + svr_sleeve at 1/4 each gated + reported book3 with svr_sleeve in place of lowrisk; base + svr_sleeve 50/50 + +Window: 2018-10-01 .. 2026-09-24, where SVR covers the members. Halves are +split at 2022-07-01. The SVR S&P 500 scan, a veto overlay on the momentum +sleeve (research/svr_overlay.py, reverted for a neighbourhood failure) and +the census all saw this window before this rule was written. low_svr21 was +also picked as the best of 86 census rows. This is a consistency check; +only a forward test can be independent. + +The rule is written to prereg.json before the run; the script refuses to +overwrite results.json. + +Result (2026-10-12): STOP. Blend: G2 passes; G1 and G3 fail. Sleeve: S1 +and S2 fail. + + arm / book (2018-10..2026-09) CAGR xSh halves MaxDD alpha vs EW (t) + base (top 10 momentum) 27.3% 0.81 0.52 / 1.05 -34.9% +16.4% (1.99) + blend w=0.25 28.2% 0.87 0.62 / 1.08 -41.1% +16.2% (2.24) + blend w=1/3 27.1% 0.87 0.67 / 1.05 -41.1% +15.0% (2.31) + blend w=0.5 23.5% 0.82 0.73 / 0.90 -40.9% +11.8% (2.13) + svr_sleeve (top 20) 17.5% 0.80 0.58 / 1.16 -38.7% +5.8% (2.87) + book3 (paper-test book) 25.1% 1.49 1.84 / 1.19 -15.0% + book4 (+ svr sleeve) 23.2% 1.42 1.49 / 1.34 -21.6% + svr in place of lowrisk 28.8% 1.47 1.55 / 1.39 -22.6% + base + svr sleeve 50/50 23.0% 0.88 0.56 / 1.24 -35.5% + + - The blend helps everywhere a little: Sharpe +0.06 at w=1/3, vol 33% -> + 30%, and better in 2018-24. It gives up most of 2026 (+109% -> +66%) and + deepens MaxDD by 6pp. That misses the +0.10 bar and the 3pp drawdown bar. + - The SVR sleeve correlates 0.50 monthly with the momentum sleeve. Lowrisk + and trend already diversify the book, so a fourth sleeve dilutes it + (1.49 -> 1.42) and adds 6.6pp of drawdown. + - Boundary: 8 years, one regime, and the factor was picked as the best of + 86 on this same window. +""" + +from __future__ import annotations + +import json +import os + +import numpy as np +import pandas as pd + +from research.lab.alpha import alpha_stats +from research.lab.factor_scan import RESULTS, load +from research.lab.factors import FACTORS +from research.lab.levered_book import stats, tbill +from research.lab.managed_futures_book import band_blend +from research.lab.short_volume import factors as svr_factors, load_svr +from research.lab.us_sweep import run + +OUT = os.path.join(RESULTS, "svr_combo") +WIN = ("2018-10-01", "2026-09-24") +HALVES = [("2018-10-01", "2022-06-30"), ("2022-07-01", "2026-09-24")] +STREAMS = os.path.join(RESULTS, "stream_portfolio", "streams.pkl") +OVERNIGHT = os.path.join(RESULTS, "overnight_only", "daily.pkl") +W_GATED, W_NEIGHBOURS = 1 / 3, (0.25, 0.5) + +PREREG = { + "question": "Does low_svr21 raise the Sharpe of overnight_recovery_top10 (score blend) or of the " + "overnight three-sleeve book (extra sleeve) in both halves of 2018-10..2026-09?", + "window": WIN, "halves": HALVES, + "gates": { + "blend (all of)": [ + "G1 blend w=1/3 excess Sharpe >= base + 0.10 over the window and >= base in each half", + "G2 blend w=0.25 and w=0.5 excess Sharpe >= base over the window (no lone spike)", + "G3 blend w=1/3 MaxDD >= base MaxDD - 3pp", + ], + "sleeve (all of)": [ + "S1 book4 excess Sharpe >= book3 + 0.05 in each half", + "S2 book4 MaxDD >= book3 MaxDD - 3pp", + ], + }, + "decision": { + "CONTINUE": "an arm passes -> add the daily FINRA short-volume fetch to the live pipeline and run " + "the arm forward-only (shadow book) before any live capital", + "STOP": "no arm passes -> low_svr21 stays a census row", + }, + "data_status": "window inspected before (SVR scan, veto overlay, census); factor chosen as best of 86", +} + + +def pct(df: pd.DataFrame, member: pd.DataFrame) -> pd.DataFrame: + return df.where(member).rank(axis=1, pct=True) + + +def summary(r: pd.Series, rf: pd.Series, ew: pd.Series) -> dict: + r = r.loc[WIN[0]:WIN[1]] + s = stats(r, rf) + a = alpha_stats(r, ew.reindex(r.index), rf.reindex(r.index)) + return {"cagr": float(s["cagr"]), "sharpe": float(s["sharpe"]), "vol": float(s["vol"]), "mdd": float(s["mdd"]), + "worst_yr": float(s["worst_yr"]), "alpha_ew": float(a["alpha"]), "alpha_ew_t": float(a["t"]), + "halves": [float(stats(r.loc[x:y], rf)["sharpe"]) for x, y in HALVES]} + + +def fmt(k: str, d: dict) -> str: + return (f"{k:34s} cagr {d['cagr']*100:5.1f}% xsh {d['sharpe']:5.2f} vol {d['vol']*100:4.1f}% " + f"mdd {d['mdd']*100:6.1f}% alpha/EW {d['alpha_ew']*100:+5.1f}% (t {d['alpha_ew_t']:+.2f}) " + f"halves {d['halves'][0]:5.2f} {d['halves'][1]:5.2f}") + + +def main() -> None: + os.makedirs(OUT, exist_ok=True) + res_path = os.path.join(OUT, "results.json") + if os.path.exists(res_path): + raise SystemExit(f"{res_path} exists; this study does not overwrite results") + with open(os.path.join(OUT, "prereg.json"), "w") as fh: + json.dump(PREREG, fh, indent=2) + + p, _ = load("us") + m = p.member + house = (pct(FACTORS["overnight_mom"](p), m) + pct(FACTORS["rec_126"](p), m)) / 2 + svr = -svr_factors(load_svr(p.close))["svr21"].shift(2).reindex(index=p.dates, columns=p.close.columns) + svr_r = pct(svr, m).fillna(0.5).where(m) # a member without FINRA data is neutral, not dropped + hr = pct(house, m) + sp = p.slice("2018-06-01", WIN[1]) + rf = tbill(p.dates).fillna(0.0) + ret = p.close.pct_change(fill_method=None) + ew = ret.where(m.shift(1, fill_value=False)).mean(axis=1) + live = dict(top_n=10, every=10, buffer=2.0) + subs = [WIN] + + S = {"base": run(sp, house.where(m), subs, **live)["_rets"]} + for w in (W_NEIGHBOURS[0], W_GATED, W_NEIGHBOURS[1]): + S[f"blend_w{w:.2f}"] = run(sp, ((1 - w) * hr + w * svr_r).where(m), subs, **live)["_rets"] + S["svr_sleeve"] = run(sp, svr.where(m), subs, top_n=20, every=21, buffer=2.0)["_rets"] + S = pd.DataFrame(S).loc[WIN[0]:WIN[1]] + + R = pd.read_pickle(STREAMS) + O = pd.read_pickle(OVERNIGHT) + B = pd.DataFrame({"overnight": O["sleeve_score"], "lowrisk": R["lowrisk"], "trend_v7": R["trend_v7"]}) + B = B.join(S[["svr_sleeve", "base"]], how="inner").loc[WIN[0]:WIN[1]] + books = { + "book3": band_blend(B[["overnight", "lowrisk", "trend_v7"]], rf)[0], + "book4": band_blend(B[["overnight", "lowrisk", "trend_v7", "svr_sleeve"]], rf)[0], + "book3_svr_for_lowrisk": band_blend(B[["overnight", "svr_sleeve", "trend_v7"]], rf)[0], + "base+svr_sleeve 50/50": band_blend(B[["base", "svr_sleeve"]], rf)[0], + } + res = {"arms": {}, "books": {}} + print("=== arms") + for k in S: + res["arms"][k] = summary(S[k], rf, ew) + print(fmt(k, res["arms"][k])) + print("=== books") + for k, r in books.items(): + res["books"][k] = summary(r, rf, ew) + print(fmt(k, res["books"][k])) + mon = (1 + S).resample("ME").prod() - 1 + res["monthly_corr_svr_sleeve_vs_base"] = float(mon["svr_sleeve"].corr(mon["base"])) + print(f"monthly corr svr_sleeve vs base {res['monthly_corr_svr_sleeve_vs_base']:.2f}") + yr = (1 + S).groupby(S.index.year).prod() - 1 + res["yearly"] = {k: {int(y): float(v) for y, v in yr[k].items()} for k in yr} + print("yearly %:\n", (yr * 100).round(1).to_string()) + + a, b = res["arms"], res["books"] + g = a[f"blend_w{W_GATED:.2f}"] + checks = { + "G1": g["sharpe"] >= a["base"]["sharpe"] + 0.10 and all(g["halves"][i] >= a["base"]["halves"][i] for i in range(2)), + "G2": all(a[f"blend_w{w:.2f}"]["sharpe"] >= a["base"]["sharpe"] for w in W_NEIGHBOURS), + "G3": g["mdd"] >= a["base"]["mdd"] - 0.03, + "S1": all(b["book4"]["halves"][i] >= b["book3"]["halves"][i] + 0.05 for i in range(2)), + "S2": b["book4"]["mdd"] >= b["book3"]["mdd"] - 0.03, + } + passing = [arm for arm, keys in (("blend", ("G1", "G2", "G3")), ("sleeve", ("S1", "S2"))) + if all(checks[k] for k in keys)] + res["verdict"] = {"checks": checks, "passing": passing, "decision": "CONTINUE" if passing else "STOP"} + print(f"checks {checks} -> {res['verdict']['decision']} {passing}") + with open(res_path, "w") as fh: + json.dump(res, fh, indent=2) + S.join(pd.DataFrame(books)).to_pickle(os.path.join(OUT, "returns.pkl")) + from research.lab.trials import append_trials + append_trials(os.path.join(RESULTS, "trials_us.jsonl"), + [{"name": f"svr_combo:{k}", **{x: v for x, v in d.items() if x != "halves"}} + for grp in ("arms", "books") for k, d in res[grp].items()], + {"study": "svr_combo", "role": "preregistered consistency check"}) + + +if __name__ == "__main__": + main()