""" Do option-volume signals predict S&P 500 stock returns? This uses OPRA daily option bars from 2024-02 (research/lab/fetch_options.py backfill, volume_panel.csv.gz). uv run python -m research.lab.fetch_options backfill # once: bars + volume_panel.csv.gz PYTHONPATH=. uv run --with numba python -m research.lab.option_volume Signals, known at the close of day t (OPRA bars are final after 16:15 New York): option volume contracts of every listed option expiring within 0-100 days (all strikes), calls C and puts P stock volume unadjusted SIP volume, so a split does not break the ratio low_os_5 (gated) -(sum_5d (C + P) x 100 / sum_5d stock volume) Johnson & So (2012): high O/S marks negative private information that short-sale costs push into options, and predicts low returns. low_pc_5 (gated) -(sum_5d P / sum_5d (C + P)) Pan & Poteshman (2006) use signed open-buy put/call volume; signed data is not available, so this is the unsigned ratio and a weaker test of the same idea. reported low_os_21 (the monthly version), low_os_abn (-(os_5 / its 126-day mean)), low_os_band_5 (O/S on the 5-100 day, 0.7-1.3 moneyness band only), call_shift (-(pc_5 - pc_63)) Test (PIT S&P 500 research panel, members only, >= 50 names a day): - Daily rank IC and top-minus-bottom quintile returns, from open(t+1) to close(t+h), for h = 5 (gated) and h = 21. Newey-West t with h lags, annualised x 252/h. - Long-only top 20 rebalanced every h days, buffer 2, next-open fills, 10 bp a side, phase-averaged: alpha vs the EW member basket. - Correlation with house_overnight_recovery (overnight_mom + rec_126) and the spanning t on it, as in research/lab/factor_census.py. Window: 2024-03-01 .. the panel's last date (2026-09-24), about 2.5 years. Halves are split at 2025-06-01. Gate per gated signal (fixed before the backfill finished; the factor census asks |t| >= 3 of a new factor): V1 h=5 quintile spread with the predicted sign and NW t >= 3.0 V2 the spread has the same sign in both halves V3 top-20 (every 5 days) alpha vs EW > 0 Decision: CONTINUE a gated signal passes all three -> preregister a combination with overnight_recovery_top10 (score blend and veto) and add the daily signal to the live data path STOP none passes -> keep collecting (the daily snapshot stores the bars) and retest when the history is about 5 years long This is the first look at these data. Nothing in the window was used before. The rule is written to prereg.json before the run; the script refuses to overwrite results.json. """ 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_census import cross_section, nw_mean_t, pct_rank, span_t 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.us_sweep import run OUT = os.path.join(RESULTS, "option_volume") PANEL = os.path.join("data", "options", "us", "volume_panel.csv.gz") START = "2024-03-01" SPLIT = "2025-06-01" GATED = ["low_os_5", "low_pc_5"] PREREG = { "question": "Do option-volume signals (Johnson-So O/S, put/call volume) predict S&P 500 returns, 2024-03..2026-09?", "signals": {"gated": GATED, "reported": ["low_os_21", "low_os_abn", "low_os_band_5", "call_shift"]}, "gates_per_signal": ["V1 h=5 quintile spread predicted sign, NW t >= 3.0", "V2 same sign in both halves (split 2025-06-01)", "V3 top-20 every-5-days alpha vs EW > 0"], "decision": {"CONTINUE": "preregister a combination with overnight_recovery_top10 and add the live signal", "STOP": "keep collecting; retest at ~5 years of history"}, "data_status": "first look; option bars backfilled 2026-10-12, not inspected before this rule", } def wide(P: pd.DataFrame, col: str, dates, cols) -> pd.DataFrame: w = P.pivot_table(index="date", columns="underlying", values=col, aggfunc="sum") w.index = pd.to_datetime(w.index) w.columns = [c.replace(".", "-") for c in w.columns] return w.reindex(index=dates, columns=cols) def signals(p, P: pd.DataFrame) -> dict[str, pd.DataFrame]: dates, cols = p.dates, p.close.columns have = wide(P, "volume", dates, cols) has_options = have.notna() # stock-volume row from the options store exists C = wide(P, "v_C", dates, cols).where(has_options).fillna(0).where(has_options) Pv = wide(P, "v_P", dates, cols).where(has_options).fillna(0).where(has_options) Cb = wide(P, "v_band_C", dates, cols).where(has_options).fillna(0).where(has_options) Pb = wide(P, "v_band_P", dates, cols).where(has_options).fillna(0).where(has_options) sv = have.where(have > 0) s = lambda x, n: x.rolling(n, min_periods=n).sum() # noqa: E731 os5 = s(C + Pv, 5) * 100 / s(sv, 5) os21 = s(C + Pv, 21) * 100 / s(sv, 21) tot5 = s(C + Pv, 5) pc5 = s(Pv, 5) / tot5.where(tot5 > 0) tot63 = s(C + Pv, 63) pc63 = s(Pv, 63) / tot63.where(tot63 > 0) return { "low_os_5": -os5, "low_pc_5": -pc5, "low_os_21": -os21, "low_os_abn": -(os5 / os5.rolling(126, min_periods=100).mean()), "low_os_band_5": -(s(Cb + Pb, 5) * 100 / s(sv, 5)), "call_shift": -(pc5 - pc63), } 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") if not os.path.exists(PANEL): raise SystemExit(f"{PANEL} missing; run fetch_options backfill first") with open(os.path.join(OUT, "prereg.json"), "w") as fh: json.dump(PREREG, fh, indent=2) p, _ = load("us") P = pd.read_csv(PANEL) S = signals(p, P) end = str(p.dates[-1].date()) m = p.member house = (pct_rank(FACTORS["overnight_mom"](p), m) + pct_rank(FACTORS["rec_126"](p), m)) / 2 o = p.open.where(p.open > 0) later = p.close.ffill() fwd = {h: later.shift(-h) / o.shift(-1) - 1 for h in (5, 21)} ret = p.close.pct_change(fill_method=None) ew = ret.where(m.shift(1, fill_value=False)).mean(axis=1) rf = tbill(p.dates).fillna(0.0) sp = p.slice("2024-01-02", end) res = {"window": [START, end], "signals": {}} for name, sc in {**S, "house_overnight_recovery": house}.items(): row = {} for h in (5, 21): ic, ls, tx, turn, n = cross_section(sc, fwd[h], m) ic, ls, tx, n = (x.loc[START:end] for x in (ic, ls, tx, n)) r = {"names": float(n.mean())} r["ic"], r["ic_t"], _ = nw_mean_t(ic, h) mu, r["ls_t"], _ = nw_mean_t(ls, h) r["ls"] = mu * 252 / h mu, r["top_x_t"], _ = nw_mean_t(tx, h) r["top_x"] = mu * 252 / h halves = [] for a, b in ((START, SPLIT), (SPLIT, end)): v = ls.loc[a:b].dropna() halves.append(float(v.mean() * 252 / h) if len(v) > 60 else float("nan")) r["ls_halves"] = halves hr = cross_section(house, fwd[h], m)[1].loc[START:end] r["corr_house"] = float(ls.corr(hr)) r["span_t_house"] = span_t(ls, hr) st = run(sp, sc.where(m), [(START, end)], top_n=20, buffer=2.0, every=h) rr = st["_rets"].loc[START:end] s_, al = stats(rr, rf), alpha_stats(rr, ew.reindex(rr.index), rf.reindex(rr.index)) r.update({"top20_cagr": float(s_["cagr"]), "top20_sharpe": float(s_["sharpe"]), "top20_mdd": float(s_["mdd"]), "alpha": float(al["alpha"]), "alpha_t": float(al["t"])}) row[f"h{h}"] = r print(f"{name:26s} h{h:2d} names {r['names']:5.0f} ic {r['ic']:+.3f} (t {r['ic_t']:+.2f}) " f"ls {r['ls']*100:+6.1f}%/yr t {r['ls_t']:+5.2f} halves {halves[0]*100:+6.1f} / {halves[1]*100:+6.1f} " f"| top20 sh {r['top20_sharpe']:.2f} alpha {r['alpha']*100:+5.1f}% (t {r['alpha_t']:+.2f}) " f"| corr house {r['corr_house']:+.2f} span t {r['span_t_house']:+.2f}", flush=True) res["signals"][name] = row checks = {} for g in GATED: r = res["signals"][g]["h5"] checks[g] = {"V1": r["ls_t"] >= 3.0, "V2": all(np.isfinite(x) and x > 0 for x in r["ls_halves"]), "V3": r["alpha"] > 0} passing = [g for g, c in checks.items() if all(c.values())] 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) from research.lab.trials import append_trials append_trials(os.path.join(RESULTS, "trials_us.jsonl"), [{"name": f"option_volume:{k}:{h}", **{x: v for x, v in d.items() if x != "ls_halves"}} for k, row in res["signals"].items() for h, d in row.items()], {"study": "option_volume", "role": "preregistered first look"}) if __name__ == "__main__": main()