#!/usr/bin/env python """Close Read study: does anything known at 15:30 ET predict the 15:30->16:00 NQ move? All signals computed as-of 15:30 ET. Label r_last30 = log(p1600/p1530). Costs: NQ round trip = spread 0.25pt + ~0.2pt/side slip (validated in 7/26 MBP-10 audit) -> 0.65pt; we report at 0.0 and at a conservative 1.0pt round trip. Splits: IS 2010-2019, OOS 2020-2026. Also per-year sign consistency. """ import os import numpy as np import pandas as pd from scipy import stats HERE = os.path.dirname(os.path.abspath(__file__)) df = pd.read_parquet(os.path.join(HERE, "sessions.parquet")) df = df.dropna(subset=["p1530", "p1600", "prev_p1600", "p1000", "p1500"]).reset_index(drop=True) df["year"] = df["date"].dt.year # label df["r_last30"] = np.log(df["p1600"] / df["p1530"]) # log return df["pts_last30"] = df["p1600"] - df["p1530"] # points # signals (as-of 15:30) df["r_day"] = np.log(df["p1530"] / df["prev_p1600"]) # prev settle -> 15:30 df["r_intraday"] = np.log(df["p1530"] / df["p0930"]) # open -> 15:30 df["r_first30"] = np.log(df["p1000"] / df["p0930"]) # 09:30 -> 10:00 df["r_lasthour_in"] = np.log(df["p1530"] / df["p1500"]) # 15:00 -> 15:30 df["r_15m_in"] = np.log(df["p1530"] / df["p1515"]) # 15:15 -> 15:30 df["vwap_dev"] = np.log(df["p1530"] / df["vwap_1530"]) # + = above vwap rng = (df["day_hi_1530"] - df["day_lo_1530"]).replace(0, np.nan) df["range_pos"] = (df["p1530"] - df["day_lo_1530"]) / rng # 0..1 pm_rng = (df["pm_hi"] - df["pm_lo"]).replace(0, np.nan) df["pm_pos"] = (df["p1530"] - df["pm_lo"]) / pm_rng df["new_extreme_1500_1530"] = np.where( df["p1530"] >= df["day_hi_1530"] - 0.25 * rng * 0.0, False, False) # placeholder, refined below OOS_START = 2020 COST_PTS = 1.0 # conservative round trip in points def evaluate(name, side): """side: +1/-1/0 per row (position for 15:30->16:00). Returns dict of stats.""" d = df.copy() d["side"] = side d = d[d["side"] != 0] if len(d) < 100: return None d["pnl_pts"] = d["side"] * d["pts_last30"] d["pnl_r"] = d["side"] * d["r_last30"] res = {} for label, sub in (("ALL", d), ("IS<2020", d[d.year < OOS_START]), ("OOS>=2020", d[d.year >= OOS_START])): n = len(sub) if n < 30: res[label] = None continue m = sub["pnl_r"].mean() t = m / (sub["pnl_r"].std(ddof=1) / np.sqrt(n)) if n > 1 else np.nan hit = (sub["pnl_r"] > 0).mean() mpts = sub["pnl_pts"].mean() net = mpts - COST_PTS res[label] = dict(n=n, mean_bps=m * 1e4, t=t, hit=hit, mean_pts=mpts, net_pts=net) # per-year sign consistency (mean pnl_r > 0) yr = d.groupby("year")["pnl_r"].mean() res["yr_pos"] = f"{(yr > 0).sum()}/{len(yr)}" res["trades_per_yr"] = len(d) / d.year.nunique() return res def report(name, side, note=""): r = evaluate(name, side) if r is None: print(f"{name:52s} (too few trades)") return a, i, o = r["ALL"], r["IS<2020"], r["OOS>=2020"] def fmt(x): if x is None: return "n<30" return f"n={x['n']:4d} {x['mean_bps']:+6.2f}bps t={x['t']:+5.2f} hit={x['hit']:.3f} net={x['net_pts']:+6.2f}pt" print(f"\n{name} {note}") print(f" ALL : {fmt(a)} yrs+ {r['yr_pos']} ({r['trades_per_yr']:.0f}/yr)") print(f" IS : {fmt(i)}") print(f" OOS : {fmt(o)}") base = df["r_last30"] print(f"sessions={len(df)} label stats: mean={base.mean()*1e4:+.2f}bps sd={base.std()*1e4:.1f}bps " f"mean|move|={np.abs(df['pts_last30']).mean():.1f}pts (all-sample), " f"last-3y mean|move|={np.abs(df[df.year>=2024]['pts_last30']).mean():.1f}pts") # H0: unconditional drift in last 30 report("H0 unconditional long last30", np.ones(len(df))) # H1: day-move continuation (leveraged ETF / MOC flow hypothesis) for q in (0.0, 0.005, 0.01): side = np.where(df["r_day"] > q, 1, np.where(df["r_day"] < -q, -1, 0)) report(f"H1 day-move continuation |r_day|>{q:.1%}", side) # H1b: fade version side = np.where(df["r_day"] > 0.01, -1, np.where(df["r_day"] < -0.01, 1, 0)) report("H1b fade big day move (>1%)", side) # H2: first-30min -> last-30min intraday momentum (Gao et al; previously refuted on NQ) side = np.sign(df["r_first30"]) report("H2 first30 -> last30 momentum", side) # H3: VWAP reversion (fade deviation) for q in (0.002, 0.004): side = np.where(df["vwap_dev"] > q, -1, np.where(df["vwap_dev"] < -q, 1, 0)) report(f"H3 fade VWAP dev >{q:.1%}", side) # H3b: VWAP continuation side = np.where(df["vwap_dev"] > 0.002, 1, np.where(df["vwap_dev"] < -0.002, -1, 0)) report("H3b ride VWAP dev >0.2%", side) # H4: afternoon range position (near pm extreme -> continuation) side = np.where(df["pm_pos"] > 0.9, 1, np.where(df["pm_pos"] < 0.1, -1, 0)) report("H4 pm-range edge continuation (pos>0.9/<0.1)", side) # H5: day range position (closing drive on trend days) for hi, lo in ((0.9, 0.1), (0.95, 0.05)): side = np.where(df["range_pos"] > hi, 1, np.where(df["range_pos"] < lo, -1, 0)) report(f"H5 day-range edge continuation ({hi}/{lo})", side) # H5b fade side = np.where(df["range_pos"] > 0.95, -1, np.where(df["range_pos"] < 0.05, 1, 0)) report("H5b fade day-range extreme (0.95/0.05)", side) # H6: last-hour momentum side = np.sign(df["r_lasthour_in"]) report("H6 15:00->15:30 momentum", side) side = np.where(np.abs(df["r_lasthour_in"]) > 0.003, np.sign(df["r_lasthour_in"]), 0) report("H6b 15:00->15:30 momentum, |move|>0.3%", side) side = np.sign(df["r_15m_in"]) report("H6c 15:15->15:30 momentum", side) # H7: trend-day composite — big day move AND closing near the matching extreme big = np.abs(df["r_day"]) > 0.01 side = np.where(big & (df["r_day"] > 0) & (df["range_pos"] > 0.8), 1, np.where(big & (df["r_day"] < 0) & (df["range_pos"] < 0.2), -1, 0)) report("H7 trend-day composite (|day|>1% & range edge)", side) # H8: interaction — day move sign vs last-hour sign agreement agree = np.sign(df["r_day"]) == np.sign(df["r_lasthour_in"]) side = np.where(agree & (np.abs(df["r_day"]) > 0.005), np.sign(df["r_day"]), 0) report("H8 day+lasthour agree, |day|>0.5%", side)