""" Does adding index exposure during short-term SPY dips improve the US books? research/lab/dip_buying.py found that the SPY RSI(2) dip is real (46 bp a trade, t 3.4, 2008-26) but is in the market on only 14% of days. Its reported-only row "SPY 1x, 3x inside dips" raised SPY's Sharpe from 0.57 to 0.69. This study puts that overlay on top of the books instead of SPY. PYTHONPATH=. uv run python -m research.lab.dip_leverage_overlay Books (daily returns, costs inside; research/lab/overnight_book.py books.pkl): three_sleeve_v7 momentum + lowrisk + trend_v7 (the live us_three_sleeve) gated three_sleeve_overnight overnight-only momentum + lowrisk + trend_v7 (paper test) gated two_sleeve_overnight_trend reported Signal: dip_buying.index_rule on SPY. Enter when the close is above its 200-day mean and RSI(2) < 10. Exit at the first close above the 5-day mean. Overlay (gated: k = 1, next_open fills): book + k x SPY from the open after the signal close to the open after the exit close, as an MES / margin position. It is financed at T-bill + 1% for every day touched, including both partial days, and pays 1 bp a side on k. reported k = 0.5 and 2; close fills; financing at T-bill + 3%; QQQ signal and instrument; "book scale", which levers the whole book (1 + k) inside the windows, with 10 bp a side and either close timing (optimistic) or only full days inside next_open windows (conservative) baselines - the same average SPY exposure held all the time (no timing) - P1: the same windows shifted circularly by a random offset - P2 (gated): the same windows at random non-overlapping starts whose signal day also has SPY above its 200-day mean. P2 separates the dip timing from the trend filter. 500 draws each, seed 20261012. Span 2006-10-02 .. 2026-09-24, halves split at 2017 (overnight_book.HALVES). All of it was inspected before, and the SPY version of this overlay was seen on 2008-26 before this rule was written. This is a consistency check on the books, not out-of-sample evidence. The rule is written to prereg.json before the run; the script refuses to overwrite results.json. Result (2026-10-12): STOP. No gated book passes. three_sleeve_v7 fails D1 and D3; three_sleeve_overnight fails D1, D2 and D3. The overlay alone (k = 1): 160 windows, 14% of days, +3.2%/yr, vol 6.4%, self-financed Sharpe 0.50, positive in 15 of 21 years (2011 and 2018 -6.7%). book Sharpe + SPY k1 (halves) MaxDD P2 rank constant SPY three_sleeve_v7 0.98 0.99 (0.75 / 1.28) -37.2 -> -37.2% 88% 0.94 three_sleeve_overnight 1.38 1.33 (1.14 / 1.55) -16.5 -> -20.4% 77% 1.31 two_sleeve_overnight_trend 1.39 1.38 (1.13 / 1.64) -20.7 -> -20.7% 84% 1.36 - k = 1 adds about 3.3pp of CAGR to each book, but vol rises by the same proportion, so the Sharpe ratio does not move. k = 2 lowers it (0.94 / 1.20 / 1.29). k = 0.5 is flat. Close fills give +0.04 / -0.01 / +0.02. Financing at +3% costs 0.02. A QQQ signal and instrument is worse. - Dip timing beats random trend-up timing. The P2 placebo gives -0.05 to -0.10 Sharpe; the real gain is -0.05 to +0.01. That ranks 77-88%, under the 95% bar. It also beats holding the same average SPY exposure all the time (+0.02 to +0.05 Sharpe, smaller MaxDD) because the 200-day filter keeps it out of bear markets. - Why it adds nothing: the books already earn the bounce. Inside the windows the books earn 50-64%/yr, against 13-22% outside. Their correlation with the SPY leg there is 0.57-0.75, and they move about 2x SPY, through TrendRider's 3x ETFs and high-beta momentum names. A Sharpe-0.5 leg correlated with a Sharpe-1.4 book lowers the book's Sharpe. - Book scale (reported) helps only the weakest book, and only with close timing: v7 0.98 -> 1.04. With full days only it is 0.95. Overnight: 1.36 / 1.29. - Leverage for 50%/yr: three_sleeve_overnight 2.35x, MaxDD -36.7%, worst yr -18.6% with the overlay 2.02x, -38.4%, -9.3% two_sleeve_overnight_trend 1.73x, -34.8%, -24.3% with the overlay 1.54x, -31.6%, -16.8% three_sleeve_v7 about 2.9x, -79% (grid check after the run), against 3.6x, -86% without it The overlay swaps one kind of leverage for another, at about the same risk. """ from __future__ import annotations import json import os import pickle import numpy as np import pandas as pd from research.lab.dip_buying import index_rule from research.lab.factor_scan import RESULTS from research.lab.levered_book import stats, tbill from research.lab.overnight_book import HALVES, SPAN, leverage_for, levered OUT = os.path.join(RESULTS, "dip_leverage_overlay") BOOKS_PKL = os.path.join(RESULTS, "overnight_book", "books.pkl") GATED = ["three_sleeve_v7", "three_sleeve_overnight"] REPORTED = ["two_sleeve_overnight_trend"] K = 1.0 SPREAD, SPREAD_STRESS = 0.01, 0.03 IDX_COST, BOOK_COST = 0.0001, 0.001 N_PLACEBO, SEED = 500, 20261012 PREREG = { "question": "Does +k x SPY held only inside SPY RSI(2) dip windows raise a book's Sharpe, beyond what " "the same exposure at random trend-up times gives?", "span": SPAN, "halves": HALVES, "gated_books": GATED, "reported_books": REPORTED, "overlay": "k = 1, next_open fills, financing T-bill + 1% on every day touched, 1 bp a side", "gates_per_book": [ "D1 excess Sharpe (book + overlay) >= book + 0.05 over the span, and higher than the book in each half", "D2 MaxDD (book + overlay) >= book MaxDD - 3pp", "D3 the span Sharpe gain > 95th percentile of P2 (trend-matched random windows, 500 draws)", ], "decision": { "CONTINUE": "a gated book passes D1-D3 -> paper-trade the overlay alongside that book from a " "15:40 signal (Alpaca 1-min SPY bars), then decide on live use", "STOP": "no gated book passes -> do not add a dip overlay; leverage stays static or vol-managed", }, "reported": "k 0.5/2, close fills, financing stress, QQQ, book scale, constant-exposure baseline, P1, " "leverage for 50%/yr with and without the overlay, yearly overlay P&L, book return inside " "vs outside the windows", "data_status": "2006-2026 previously inspected; the SPY overlay result was seen before this rule", } def windows(c: pd.Series, o: pd.Series, fill: str) -> list[tuple[int, int]]: """(a, b) positions: next_open holds from the open of a to the open of b; close holds from the close of a to the close of b.""" _, _, trades = index_rule(c, o, fill=fill) pos = {d: i for i, d in enumerate(c.index)} return [(pos[t[0]], pos[t[1]]) for t in trades] def overlay_parts(cv: np.ndarray, ov: np.ndarray, wins, fill: str): """In-market index return, financed-day flag and trade count per day.""" n = len(cv) rcc = np.r_[0.0, cv[1:] / cv[:-1] - 1] m, inside, tr = np.zeros(n), np.zeros(n), np.zeros(n) for a, b in wins: if fill == "next_open": m[a] += cv[a] / ov[a] - 1 m[a + 1:b] += rcc[a + 1:b] m[b] += ov[b] / cv[b - 1] - 1 inside[a:b + 1] = 1 else: m[a + 1:b + 1] += rcc[a + 1:b + 1] inside[a + 1:b + 1] = 1 tr[a] += 1 tr[b] += 1 return m, inside, tr def overlay(parts, rfd: np.ndarray, k: float, spread: float = SPREAD, cost: float = IDX_COST) -> np.ndarray: m, inside, tr = parts return k * (m - inside * (rfd + spread / 252)) - k * cost * tr def xsh(r: np.ndarray, rfd: np.ndarray) -> float: return float((r - rfd).mean() / r.std() * np.sqrt(252)) def summary(r: pd.Series, rf: pd.Series) -> dict: s = stats(r, rf) d = {k: float(s[k]) for k in ("cagr", "sharpe", "vol", "mdd", "worst_yr")} d["halves"] = [float(stats(r.loc[a:b], rf)["sharpe"]) for a, b in HALVES] return d def placebo_trend(rng, wins, elig, lo, hi): """Same window lengths at random non-overlapping starts with a trend-up signal day.""" cand = np.where(elig[lo:hi - 1])[0] + lo taken = np.zeros(hi + 2, bool) out = [] for D in rng.permutation([b - a for a, b in wins]): for _ in range(200): a = int(rng.choice(cand)) + 1 b = a + D if b >= hi or taken[max(a - 1, 0):b + 2].any(): continue taken[a:b + 1] = True out.append((a, b)) break return out def placebo_shift(rng, wins, lo, hi): s = int(rng.integers(63, hi - lo - 63)) out = [] for a, b in wins: a2 = lo + (a - lo + s) % (hi - lo) b2 = a2 + (b - a) if b2 < hi: out.append((a2, b2)) return out def fmt(name: str, d: dict) -> str: return (f"{name:46s} cagr {d['cagr']*100:5.1f}% xsh {d['sharpe']:5.2f} vol {d['vol']*100:4.1f}% " f"mdd {d['mdd']*100:6.1f}% worst yr {d['worst_yr']*100:6.1f}% halves " f"{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) B = pd.read_pickle(BOOKS_PKL).loc[SPAN[0]:SPAN[1]] idx = B.index with open(os.path.join("data", "research", "long_history.pkl"), "rb") as fh: L = pickle.load(fh) rf = tbill(idx).fillna(0.0) rfd = (rf / 252).to_numpy() inst = {} for tk in ("SPY", "QQQ"): c = L["close"][tk].dropna().loc[:SPAN[1]] o = L["open"][tk].reindex(c.index) miss = idx.difference(c.index) assert len(miss) == 0, f"{tk} lacks {len(miss)} book dates" sma = c.rolling(200).mean() # windows on the full history, then positions in the book's calendar on_idx = lambda w: [(idx.get_loc(c.index[a]), idx.get_loc(c.index[b])) for a, b in w if c.index[a] >= idx[0] and c.index[b] in idx] cv, ov = c.reindex(idx).to_numpy(), o.reindex(idx).to_numpy() inst[tk] = {"cv": cv, "ov": ov, "elig": (c > sma).reindex(idx).fillna(False).to_numpy(), "w": {f: on_idx(windows(c, o, f)) for f in ("next_open", "close")}} spy = inst["SPY"] parts = {f: overlay_parts(spy["cv"], spy["ov"], spy["w"][f], f) for f in ("next_open", "close")} ov1 = overlay(parts["next_open"], rfd, K) share_in = float(parts["next_open"][1].mean()) res = {"n_windows": len(spy["w"]["next_open"]), "share_days": share_in, "overlay": {}, "books": {}} ovs = pd.Series(ov1, idx) res["overlay"] = {"ann_mean": float(ovs.mean() * 252), "vol": float(ovs.std() * np.sqrt(252)), "sharpe_self_financed": float(ovs.mean() / ovs.std() * np.sqrt(252)), "yearly": {int(y): float(v) for y, v in ovs.groupby(idx.year).sum().items()}} yrs = pd.Series(res["overlay"]["yearly"]) print(f"windows {res['n_windows']}, share of days {share_in:.1%}; overlay k=1 mean {res['overlay']['ann_mean']*100:.2f}%/yr " f"vol {res['overlay']['vol']*100:.1f}% self-financed Sharpe {res['overlay']['sharpe_self_financed']:.2f}; " f"positive years {(yrs > 0).sum()}/{len(yrs)}") print("overlay P&L by year %:", (yrs * 100).round(1).to_dict()) rng = np.random.default_rng(SEED) lo, hi = 0, len(idx) P2 = [overlay(overlay_parts(spy["cv"], spy["ov"], placebo_trend(rng, spy["w"]["next_open"], spy["elig"], lo, hi), "next_open"), rfd, K) for _ in range(N_PLACEBO)] P1 = [overlay(overlay_parts(spy["cv"], spy["ov"], placebo_shift(rng, spy["w"]["next_open"], lo, hi), "next_open"), rfd, K) for _ in range(N_PLACEBO)] m_in = parts["next_open"][1] > 0 r_spy_cc = np.r_[0.0, spy["cv"][1:] / spy["cv"][:-1] - 1] const = share_in * K * (r_spy_cc - (rfd + SPREAD / 252)) gates = {} for name in GATED + REPORTED: r = B[name].to_numpy() base = summary(B[name], rf) row = {"book": base, "variants": {}} print(f"\n=== {name}") print(fmt("book", base)) variants = { "spy k1 next_open": r + ov1, "spy k0.5 next_open": r + overlay(parts["next_open"], rfd, 0.5), "spy k2 next_open": r + overlay(parts["next_open"], rfd, 2.0), "spy k1 close": r + overlay(parts["close"], rfd, K), "spy k1 next_open fin+3%": r + overlay(parts["next_open"], rfd, K, SPREAD_STRESS), "constant SPY, same avg exposure": r + const, } q = inst["QQQ"] variants["qqq k1 next_open"] = r + overlay(overlay_parts(q["cv"], q["ov"], q["w"]["next_open"], "next_open"), rfd, K) # book scale: lever the book itself inside the windows _, in_close, tr_close = parts["close"] variants["book scale k1 close timing"] = (r * (1 + K * in_close) - K * in_close * (rfd + SPREAD / 252) - K * BOOK_COST * tr_close) full_days = np.zeros(len(r)) for a, b in spy["w"]["next_open"]: full_days[a + 1:b] = 1 variants["book scale k1 full days only"] = (r * (1 + K * full_days) - K * parts["next_open"][1] * (rfd + SPREAD / 252) - K * BOOK_COST * parts["next_open"][2]) for vname, x in variants.items(): d = summary(pd.Series(x, idx), rf) row["variants"][vname] = d print(fmt(" " + vname, d)) s0 = xsh(r, rfd) g2 = np.array([xsh(r + p, rfd) - s0 for p in P2]) g1 = np.array([xsh(r + p, rfd) - s0 for p in P1]) gain = row["variants"]["spy k1 next_open"]["sharpe"] - base["sharpe"] row["placebo"] = {"gain": gain, "P2_mean": float(g2.mean()), "P2_p95": float(np.quantile(g2, 0.95)), "P2_pct_rank": float((g2 < gain).mean()), "P1_mean": float(g1.mean()), "P1_p95": float(np.quantile(g1, 0.95)), "P1_pct_rank": float((g1 < gain).mean())} print(f" Sharpe gain {gain:+.3f} | P2 trend-matched mean {g2.mean():+.3f} p95 {np.quantile(g2, .95):+.3f} " f"rank {row['placebo']['P2_pct_rank']:.1%} | P1 shift mean {g1.mean():+.3f} p95 {np.quantile(g1, .95):+.3f} " f"rank {row['placebo']['P1_pct_rank']:.1%}") ins, outs = r[m_in], r[~m_in] row["book_inside_vs_outside"] = {"inside_ann": float(ins.mean() * 252), "outside_ann": float(outs.mean() * 252), "inside_vol": float(ins.std() * np.sqrt(252)), "outside_vol": float(outs.std() * np.sqrt(252)), "corr_with_overlay_inside": float(np.corrcoef(ins, ov1[m_in])[0, 1])} bi = row["book_inside_vs_outside"] print(f" book inside windows {bi['inside_ann']*100:5.1f}%/yr (vol {bi['inside_vol']*100:4.1f}%) vs outside " f"{bi['outside_ann']*100:5.1f}%/yr (vol {bi['outside_vol']*100:4.1f}%); corr with SPY leg inside " f"{bi['corr_with_overlay_inside']:.2f}") lev = {} for lname, x in (("book", B[name]), ("book + spy k1", pd.Series(r + ov1, idx))): Lx = leverage_for(x, rf) d = {"L": Lx} if np.isfinite(Lx): s = stats(levered(x, rf, Lx), rf) d.update({"mdd": float(s["mdd"]), "worst_yr": float(s["worst_yr"]), "vol": float(s["vol"]), "halves_cagr": [float(stats(levered(x.loc[a:b], rf, Lx), rf)["cagr"]) for a, b in HALVES]}) print(f" 50%/yr: {lname:14s} L {Lx:4.2f} mdd {d['mdd']*100:6.1f}% worst yr {d['worst_yr']*100:6.1f}% " f"halves cagr {d['halves_cagr'][0]*100:5.1f}% {d['halves_cagr'][1]*100:5.1f}%") lev[lname] = d row["leverage_for_50"] = lev res["books"][name] = row if name in GATED: v = row["variants"]["spy k1 next_open"] gates[name] = { "D1": v["sharpe"] >= base["sharpe"] + 0.05 and all(v["halves"][i] > base["halves"][i] for i in range(2)), "D2": v["mdd"] >= base["mdd"] - 0.03, "D3": gain > row["placebo"]["P2_p95"], } passing = [k for k, g in gates.items() if all(g.values())] res["verdict"] = {"gates": gates, "passing": passing, "decision": "CONTINUE" if passing else "STOP"} print(f"\ngates {gates}\n-> {res['verdict']['decision']} {passing}") with open(res_path, "w") as fh: json.dump(res, fh, indent=2) pd.DataFrame({"overlay_spy_k1": ov1, "inside": parts["next_open"][1]}, index=idx).to_pickle( os.path.join(OUT, "overlay.pkl")) from research.lab.trials import append_trials append_trials(os.path.join(RESULTS, "trials_us.jsonl"), [{"name": f"dip_leverage_overlay:{b}:{v}", **{x: y for x, y in d.items() if x != "halves"}} for b, row in res["books"].items() for v, d in row["variants"].items()], {"study": "dip_leverage_overlay", "role": "preregistered consistency check"}) if __name__ == "__main__": main()