#!/usr/bin/env python """Deep-dive on late-day momentum: 15:00->15:30 move continues into 16:00. Validation gauntlet: 1. Threshold sweep (monotonicity, not cherry-picking 0.3%) 2. Vol-normalized signal variant (z-score vs session 1m vol) 3. Delayed entry 15:31/15:33/15:35 (stale-trigger check) 4. Per-year table, long/short split, FOMC-proxy split 5. Bootstrap p-value (10k), equity curve, max drawdown 6. Minute-by-minute mean cumulative pnl path after 15:30 """ import os import numpy as np import pandas as pd HERE = os.path.dirname(os.path.abspath(__file__)) sess = pd.read_parquet(os.path.join(HERE, "sessions.parquet")) paths = pd.read_parquet(os.path.join(HERE, "pm_paths.parquet")) paths.index = pd.to_datetime(paths.index) paths.columns = paths.columns.astype(int) sess = sess.dropna(subset=["p1500", "p1530", "p1600"]).copy() sess["year"] = sess["date"].dt.year sess = sess.set_index("date") common = sess.index.intersection(paths.index) sess = sess.loc[common] P = paths.loc[common] r_sig = np.log(sess["p1530"] / sess["p1500"]) rv30 = sess["rv1m"] * np.sqrt(30) # session 1m vol scaled to 30m z = r_sig / rv30 label = np.log(sess["p1600"] / sess["p1530"]) pts = sess["p1600"] - sess["p1530"] COST = 1.0 def stats_for(mask, side, entry_pts=None, entry_r=None, name=""): side = side[mask] if entry_r is None: pl_r, pl_p = side * label[mask], side * pts[mask] else: pl_r, pl_p = side * entry_r[mask], side * entry_pts[mask] n = len(pl_r) if n < 20: return f"{name:44s} n={n} (too few)" m, t = pl_r.mean(), pl_r.mean() / (pl_r.std(ddof=1) / np.sqrt(n)) return (f"{name:44s} n={n:4d} {m*1e4:+6.2f}bps t={t:+5.2f} hit={(pl_r>0).mean():.3f} " f"net={pl_p.mean()-COST:+6.2f}pt") print("=== 1. threshold sweep (fixed %) ===") for q in (0.0015, 0.002, 0.0025, 0.003, 0.0035, 0.004, 0.005): mask = np.abs(r_sig) > q print(stats_for(mask, np.sign(r_sig), name=f"|r_1500_1530| > {q:.2%}")) print("\n=== 2. vol-normalized sweep ===") for qz in (1.0, 1.5, 2.0, 2.5, 3.0): mask = np.abs(z) > qz print(stats_for(mask, np.sign(r_sig), name=f"|z| > {qz}")) print("\n=== 3. delayed entry (|r|>0.3%), exit 16:00 ===") for entry_tod, lbl in ((930, "15:30"), (931, "15:31"), (933, "15:33"), (935, "15:35")): pe = P[entry_tod - 1] if entry_tod - 1 in P.columns else P[930] er = np.log(sess["p1600"] / pe) ep = sess["p1600"] - pe mask = (np.abs(r_sig) > 0.003) & pe.notna() print(stats_for(mask, np.sign(r_sig), entry_pts=ep, entry_r=er, name=f"enter {lbl}")) print("\n=== 4a. per-year (|r|>0.3%) ===") mask = np.abs(r_sig) > 0.003 d = pd.DataFrame({"year": sess["year"], "pl_r": np.sign(r_sig) * label, "pl_p": np.sign(r_sig) * pts, "side": np.sign(r_sig)})[mask] tbl = d.groupby("year").agg(n=("pl_r", "size"), bps=("pl_r", lambda x: x.mean()*1e4), net_pts=("pl_p", lambda x: x.mean()-COST), hit=("pl_r", lambda x: (x>0).mean())) print(tbl.round(2).to_string()) print("\n=== 4b. long vs short (|r|>0.3%) ===") print(stats_for(mask & (r_sig > 0), np.sign(r_sig), name="longs (up momentum)")) print(stats_for(mask & (r_sig < 0), np.sign(r_sig), name="shorts (down momentum)")) print("\n=== 4c. 14:00-event-day proxy split (|r|>0.3%) ===") # proxy: abs 13:59->14:05 move in top 5% of all days => likely FOMC/2pm event ev = np.abs(np.log(P[844] / P[858 - 19])) # 14:05 close vs 13:59 close -> cols 844 (14:04 open bar close ~14:05) and 839 ev = np.abs(np.log(P[844] / P[839])) thr = ev.quantile(0.95) is_ev = ev > thr print(stats_for(mask & is_ev, np.sign(r_sig), name=f"event days (n_ev={int((mask & is_ev).sum())})")) print(stats_for(mask & ~is_ev, np.sign(r_sig), name="non-event days")) print("\n=== 5. bootstrap + equity (|r|>0.3%) ===") rng = np.random.default_rng(7) obs = d["pl_r"].mean() boot = np.array([rng.choice(d["pl_r"].values - d["pl_r"].mean() + 0, size=len(d), replace=True).mean() for _ in range(10000)]) # p-value under null mean 0: center at 0 null = d["pl_r"].values - d["pl_r"].mean() boot = np.array([rng.choice(null, size=len(null), replace=True).mean() for _ in range(10000)]) p = (boot >= obs).mean() eq = d["pl_p"].cumsum() dd = (eq - eq.cummax()).min() ann_sharpe = d["pl_r"].mean() / d["pl_r"].std(ddof=1) * np.sqrt(len(d) / (len(sess) / sess["year"].nunique()) * 252 / 252 * (len(d)/d['year'].nunique())) # simpler: per-trade sharpe * sqrt(trades/yr) sh = d["pl_r"].mean() / d["pl_r"].std(ddof=1) * np.sqrt(len(d) / d["year"].nunique()) print(f"bootstrap p(one-sided)={p:.4f} gross_total={eq.iloc[-1]:+.0f}pts maxDD={dd:.0f}pts " f"trade-Sharpe(ann)={sh:.2f} worst day={d['pl_p'].min():+.0f}pts best={d['pl_p'].max():+.0f}pts") print("\n=== 6. minute path after 15:30 (mean cum pnl, pts, |r|>0.3% days) ===") sel = P[mask] side_sel = np.sign(r_sig)[mask] base_p = sess["p1530"][mask] outrows = [] for tod in range(930, 960, 3): col = tod - 1 if col not in P.columns: continue cum = (side_sel * (sel[col] - base_p)).mean() outrows.append((f"{tod//60:02d}:{tod%60:02d}", round(cum, 2))) cum1600 = (side_sel * (sess["p1600"][mask] - base_p)).mean() outrows.append(("16:00", round(cum1600, 2))) print(pd.DataFrame(outrows, columns=["time", "mean_cum_pts"]).to_string(index=False)) print("\n=== 7. combo: momentum + day-move agreement (|r|>0.3%) ===") r_day = np.log(sess["p1530"] / sess["prev_p1600"]) agree = np.sign(r_day) == np.sign(r_sig) print(stats_for(mask & agree, np.sign(r_sig), name="momentum agrees with day move")) print(stats_for(mask & ~agree, np.sign(r_sig), name="momentum against day move")) print("\n=== 8. also at range edge? (pm_pos overlap) ===") pm_rng = (sess["pm_hi"] - sess["pm_lo"]).replace(0, np.nan) pm_pos = (sess["p1530"] - sess["pm_lo"]) / pm_rng edge = (pm_pos > 0.9) | (pm_pos < 0.1) print(f"overlap: of {int(mask.sum())} momentum days, {int((mask & edge).sum())} are also pm-range-edge days") print(stats_for(mask & edge, np.sign(r_sig), name="momentum AND pm edge")) print(stats_for(mask & ~edge, np.sign(r_sig), name="momentum, NOT pm edge")) print(stats_for(edge & ~mask, np.where(pm_pos > 0.5, 1, -1), name="pm edge only (no momentum)"))