"""_rebal01.py - REBAL01: 60/40 rebalancing pressure (Harvey, Mazzoleni, Melone 2025) traded on US100 intraday. Prereg: REBAL01_PREREG.md (frozen and pushed before this script read any market data; sha256 89c280c0458bb310a9c793a1836221cabfbfea0fff46f17761db57ae6a066766). Tested by: Claude (2026-10-10). Signal on daily closes of day t (SPY / IEF adjusted), trade on t+1: entry 09:35 ET open, exit 15:45 ET open, stop 1.0 x daily ATR(14) known at t, R = |W| x direction x move / ATR - cost. Run: python -X utf8 prop_eur2k_lab/_rebal01.py """ import io import math import os import sys import urllib.request import numpy as np import pandas as pd LAB = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, LAB) import _bar_boundary_audit_20261009 as B # noqa: E402 NY = "America/New_York" DATA = os.path.join(LAB, "_rebal01_data") LSE = os.path.join(os.path.dirname(LAB), "quantlab", "data", "inbox", "lse", "NQ_F_1m_2016-06-01_2026-10-09.parquet") CFD_PTS, FUT_PTS = 1.70, 2 * 0.25 + 1.50 / 2.0 # fallback CFD model (scaled by price); futures: 1 tick/side + $1.50 per MNQ START, OOS0, END = "2005-03-01", "2023-03-18", "2026-10-09" BANDS = np.arange(26) / 1000.0 RNG = np.random.default_rng(20261010) OUT = [] def pr(s=""): OUT.append(s) print(s, flush=True) def tstat(x): x = np.asarray(x, float) x = x[~np.isnan(x)] return float(x.mean() / x.std(ddof=1) * math.sqrt(len(x))) if len(x) > 2 and x.std(ddof=1) > 0 else float("nan") def row(lab, x): x = np.asarray(x, float) x = x[~np.isnan(x)] if len(x) < 3: return f" {lab:<44s} n={len(x):5d}" return f" {lab:<44s} n={len(x):5d} mean={x.mean():+.4f}R t={tstat(x):+5.2f} win={100 * (x > 0).mean():4.1f}% sum={x.sum():+7.1f}R" # ------------------------------------------------------------------------------------------------ daily signal def daily_closes(): os.makedirs(DATA, exist_ok=True) f = os.path.join(DATA, "spy_ief_adj.csv") if not os.path.exists(f): import yfinance as yf px = yf.download(["SPY", "IEF"], start="2002-07-01", end="2026-10-10", auto_adjust=True, progress=False)["Close"] px.dropna().to_csv(f) px = pd.read_csv(f, index_col=0, parse_dates=True).dropna() px = px[px.index <= END] assert px.index.is_unique and px.index.is_monotonic_increasing g = os.path.join(DATA, "DGS10.csv") if not os.path.exists(g): try: raw = urllib.request.urlopen("https://fred.stlouisfed.org/graph/fredgraph.csv?id=DGS10", timeout=60).read() open(g, "wb").write(raw) except Exception as e: # sanity check only pr(f" FRED DGS10 not available ({type(e).__name__}); sanity check skipped") y = None if os.path.exists(g): y = pd.read_csv(g, index_col=0, parse_dates=True).iloc[:, 0] y = pd.to_numeric(y, errors="coerce").dropna() return px, y def signal(px): """W_t for every trading day t (information: closes through t).""" rs, rb = px["SPY"].pct_change().to_numpy(), px["IEF"].pct_change().to_numpy() n = len(px) month = px.index.to_period("M") pos = pd.Series(np.arange(n), index=px.index) from_end = (pos.groupby(month).transform("max") - pos).to_numpy() first = (pos.groupby(month).transform("min") == pos).to_numpy() complete = np.ones(n, bool) # the last month in the file is not finished: no month-end flags there last_m = month == month[-1] if px.index[-1] != (px.index[-1] + pd.offsets.BMonthEnd(0)): complete[last_m] = False last5 = (from_end <= 4) & complete month_end = (from_end == 0) & complete w = np.full(len(BANDS), 0.60) wc = 0.60 thr, cal = np.zeros(n), np.zeros(n) for i in range(1, n): # path-dependent recursion: 5,500 days x 26 bands w = w * (1 + rs[i]) / (w * (1 + rs[i]) + (1 - w) * (1 + rb[i])) thr[i] = (w - 0.60).mean() w = np.where(np.abs(w - 0.60) > BANDS, 0.60, w) wc = wc * (1 + rs[i]) / (wc * (1 + rs[i]) + (1 - wc) * (1 + rb[i])) cal[i] = wc - 0.60 if month_end[i]: wc = 0.60 cal_s = pd.Series(cal) C = np.where(last5, np.sign(-cal), np.where(first, np.sign(cal_s.shift(4).fillna(0.0).to_numpy()), 0.0)) S = pd.DataFrame({"thr": thr, "cal": cal, "C": C}, index=px.index) S["W_thr"] = np.clip(-S["thr"] / 0.015, -1, 1) S["W"] = np.clip(0.5 * (-S["thr"] / 0.015) + 0.5 * S["C"], -1, 1) nxt = list(px.index[1:]) + [px.index[-1] + pd.offsets.BDay(1)] S["trade_day"] = pd.DatetimeIndex(nxt) return S.iloc[1:] # ------------------------------------------------------------------------------------------------ intraday book def book(m, spread=None, entry_mod=575, cost_pts=None, ref=None): """m: tz-aware NY M1 mid bars stamped at minute START. One row per usable day, per-unit-size long/short results in ATR units.""" m = m[m.index.dayofweek < 5] mod = np.asarray(m.index.hour * 60 + m.index.minute) k = (mod >= 570) & (mod <= 959) rth, rmod = m[k], mod[k] assert not rth.index.duplicated().any() day = rth.index.normalize().tz_localize(None) g = rth.groupby(day) D = pd.DataFrame({"n": g.size(), "o": g["open"].first(), "h": g["high"].max(), "l": g["low"].min(), "c": g["close"].last()}) D = D[D["n"] >= 30] pc = D["c"].shift(1) tr = np.maximum(D["h"] - D["l"], np.maximum((D["h"] - pc).abs(), (D["l"] - pc).abs())) atr = tr.ewm(alpha=1 / 14, adjust=False, min_periods=14).mean() D["atr"] = atr.shift(1) # ATR through the previous session: known before the trade day D["prev_c"] = pc def at(col, mm): q = rmod == mm return pd.Series(rth[col].to_numpy()[q], index=day[q]) D["entry"], D["exit"] = at("open", entry_mod), at("open", 945) q = (rmod >= entry_mod) & (rmod <= 944) D["lo"] = pd.Series(rth["low"].to_numpy()[q], index=day[q]).groupby(level=0).min() D["hi"] = pd.Series(rth["high"].to_numpy()[q], index=day[q]).groupby(level=0).max() ets = pd.Series(rth.index[rmod == entry_mod], index=day[rmod == entry_mod]) D["entry_ts"] = ets D["usable"] = (D["n"] >= 360) & D[["entry", "exit", "atr", "lo", "hi"]].notna().all(axis=1) & (D["atr"] > 0) pts = D["exit"] - D["entry"] D["RL"] = np.where(D["lo"] <= D["entry"] - D["atr"], -1.0, pts / D["atr"]) D["RS"] = np.where(D["hi"] >= D["entry"] + D["atr"], -1.0, -pts / D["atr"]) if spread is not None: sp = spread[spread.index.dayofweek < 5] smod = np.asarray(sp.index.hour * 60 + sp.index.minute) sday = sp.index.normalize().tz_localize(None) s_in = pd.Series(sp.to_numpy()[smod == entry_mod], index=sday[smod == entry_mod]).reindex(D.index) s_out = pd.Series(sp.to_numpy()[smod == 945], index=sday[smod == 945]).reindex(D.index) real = 0.5 * (s_in + s_out) model = cost_pts * D["entry"] / ref use_real = (D.index >= "2012-03-01") & real.notna() & (s_in > 0) & (s_out > 0) D["cost_pts"] = np.where(use_real, real, model) D["cost_real"] = use_real else: D["cost_pts"] = cost_pts D["cost_real"] = False D["cu"] = D["cost_pts"] / D["atr"] # cost per unit size, in R return D def trades(D, S, wcol="W", lag=0): """Join the signal of day t to the instrument's day t+1. lag shifts the signal by extra trading days (0 = the spec).""" s = S[[wcol, "trade_day"]].copy() s[wcol] = s[wcol].shift(lag) s = s.dropna().set_index("trade_day") assert s.index.is_unique and D.index.is_unique # join integrity T = D.join(s[wcol].rename("W"), how="inner", validate="1:1") T = T[(T.index >= START) & (T.index <= END)] n_sig = int((T["W"] != 0).sum()) T = T[T["usable"] & (T["W"] != 0)].copy() a = T["W"].abs() T["gross"] = a * np.where(T["W"] > 0, T["RL"], T["RS"]) T["cost"] = a * T["cu"] T["R"], T["R2"] = T["gross"] - T["cost"], T["gross"] - 2 * T["cost"] T["year"] = T.index.year T.attrs["skipped"] = n_sig - len(T) return T def load_cfd_raw(): df = pd.read_parquet(B.CFD) df.columns = [c.lower() for c in df.columns] df = df.set_index("date") idx = pd.DatetimeIndex(df.index).tz_localize("America/Chicago", ambiguous="NaT", nonexistent="NaT") ok = ~idx.isna() df = df[ok] df.index = idx[ok].tz_convert(NY) df = df[~df.index.duplicated(keep="last")].sort_index() mid = pd.DataFrame({k: (df[f"bid_{k}"] + df[f"ask_{k}"]) / 2 for k in ("open", "high", "low", "close")}) return mid, df["ask_open"] - df["bid_open"] def main(): pr("REBAL01 - 60/40 rebalancing pressure on US100 intraday. Prereg sha256 89c280c0...6766 (frozen before the run).") px, y = daily_closes() S = signal(px) pr(f"\n0. SIGNAL SPY/IEF adjusted closes {px.index[0].date()} -> {px.index[-1].date()} ({len(px)} days)") if y is not None: j = pd.concat([px["IEF"].pct_change().rename("ief"), (-y.diff()).rename("dy")], axis=1).dropna() pr(f" FRED DGS10 sanity: corr(IEF daily return, -change in 10Y yield) = {j['ief'].corr(j['dy']):+.3f} on {len(j)} days") s = S[(S["trade_day"] >= START)] pr(f" Threshold (weight deviation): sd {100 * s['thr'].std():.3f}% p1/p99 {100 * s['thr'].quantile(.01):+.2f}% / {100 * s['thr'].quantile(.99):+.2f}%" f" | Calendar: sd {100 * s['cal'].std():.3f}% | C != 0 on {100 * (s['C'] != 0).mean():.1f}% of days") pr(f" W: mean {s['W'].mean():+.3f} mean |W| {s['W'].abs().mean():.3f} long {100 * (s['W'] > 0).mean():.1f}% short {100 * (s['W'] < 0).mean():.1f}% " f"clipped {100 * (s['W'].abs() >= 1).mean():.1f}% autocorr(1) {s['W'].autocorr():+.2f}") mid, spr = load_cfd_raw() ref = float(mid["close"].iloc[-5000:].median()) sh = lambda x, ph: x.set_axis(x.index + pd.Timedelta(ph, "min")) D = book(sh(mid, -1), sh(spr, -1), 575, CFD_PTS, ref) T = trades(D, S) pr(f"\n1. PRIMARY: NAS100 CFD, true clock (file stamps - 1 min), entry 09:35 ET, exit 15:45 ET, stop 1.0 x daily ATR14") pr(f" trades {T.index.min().date()} -> {T.index.max().date()} (the CFD file ends 2026-08-14) | signal days skipped (short session / missing bar): {T.attrs['skipped']}") stop = np.where(T["W"] > 0, T["RL"], T["RS"]) == -1.0 pr(f" stop hit {100 * stop.mean():.2f}% | ATR median {1e4 * (T['atr'] / T['entry']).median():.0f} bp of price | cost per unit size median {T['cu'].median():.4f}R " f"(real bid/ask on {100 * T['cost_real'].mean():.0f}% of trades) | cost per trade median {T['cost'].median():.4f}R | SD of R {T['R'].std():.3f}") pr(row("gross", T["gross"])) pr(row("net 1x", T["R"])) pr(row("net 2x", T["R2"])) oos = T[T.index >= OOS0] h1, h2 = T[T.year <= 2015], T[T.year >= 2016] pr(row("authors' sample 2005-03 .. 2023-03-17 net", T[T.index < OOS0]["R"])) pr(row("out-of-authors OOS 2023-03-18 .. end net", oos["R"])) pr(row("half 2005-2015 net", h1["R"])) pr(row("half 2016-2026 net", h2["R"])) pr(row("long days net", T[T.W > 0]["R"])) pr(row("short days net", T[T.W < 0]["R"])) pr(" by year (net 1x, sum R / t): " + " ".join(f"{yv}:{g['R'].sum():+.1f}/{tstat(g['R']):+.1f}" for yv, g in T.groupby("year"))) T.assign(timestamp=T["entry_ts"].dt.tz_convert("UTC"))[["timestamp", "W", "R", "cost", "gross", "R2", "atr", "entry", "exit"]].to_csv( os.path.join(LAB, "_rebal01_trades.csv"), index_label="trade_day") pr("\n2. BUG, LEAK AND PLACEBO CHECKS (dataset 1)") flip = T["W"].abs() * np.where(T["W"] > 0, T["RS"], T["RL"]) pr(f" flip-all gross: mean {flip.mean():+.4f}R t={tstat(flip):+.2f} (original gross {T['gross'].mean():+.4f}R)") al, as_ = T["W"].abs() * T["RL"], T["W"].abs() * T["RS"] pr(f" always long, same days and sizes: gross {al.mean():+.4f}R t={tstat(al):+.2f} | always short: {as_.mean():+.4f}R t={tstat(as_):+.2f}") for lag, lab in [(-1, "lag 0 (same-day close = LOOKAHEAD, must not be traded)"), (0, "lag 1 (the spec)"), (1, "lag 2")]: X = trades(D, S, lag=lag) pr(f" signal {lab:<56s} gross mean {X['gross'].mean():+.4f}R t={tstat(X['gross']):+.2f} | net t={tstat(X['R']):+.2f}") a, RL, RS, cu = T["W"].to_numpy(), T["RL"].to_numpy(), T["RS"].to_numpy(), T["cu"].to_numpy() perm = np.array([RNG.permutation(a) for _ in range(200)]) rp = np.abs(perm) * (np.where(perm > 0, RL, RS) - cu) tp = rp.mean(1) / rp.std(1, ddof=1) * math.sqrt(len(a)) t1 = tstat(T["R"]) p_perm = (1 + int((tp >= t1).sum())) / 201 pr(f" random-200 (W shuffled across the trade dates): placebo t mean {tp.mean():+.2f}, p95 {np.percentile(tp, 95):+.2f}; observed {t1:+.2f}; p = {p_perm:.3f}") ph = {p: tstat(trades(book(sh(mid, p), sh(spr, p), 575, CFD_PTS, ref), S)["R"]) for p in range(-3, 4)} pr(" bar phase (net 1x t; -1 is the true clock): " + " ".join(f"{p:+d}m {v:+.2f}" for p, v in ph.items())) p1 = pd.read_csv(os.path.join(LAB, "_bar_boundary_trueclock_trades.csv")) p1["day"] = pd.to_datetime(p1["entry_ts"], utc=True).dt.tz_convert(NY).dt.tz_localize(None).dt.normalize() pd1 = p1.groupby("day")["net"].sum() j = pd.concat([T["R"].rename("x"), pd1.rename("p")], axis=1) j = j[(j.index >= T.index.min()) & (j.index <= T.index.max())].fillna(0.0) rho_all = j["x"].corr(j["p"]) pp = pd1[(pd1.index >= T.index.min()) & (pd1.index <= T.index.max())] worst = pp[pp <= pp.quantile(0.05)].index jw = j.loc[worst] rho_w = jw["x"].corr(jw["p"]) # [withheld: refers to the lab's private live strategy] # [withheld: refers to the lab's private live strategy] pr("\n3. SECOND SOURCE: NQ futures, LSE NQ.F 1m 2016-06 -> 2026-10-09, cost 1.25 pt round trip") d = pd.read_parquet(LSE) d["timestamp"] = pd.to_datetime(d["timestamp"], utc=True) m = d.drop_duplicates("timestamp").set_index("timestamp").sort_index()[["open", "high", "low", "close"]].astype(float) m.index = m.index.tz_convert(NY) DN = book(m, None, 575, FUT_PTS) N = trades(DN, S) pr(f" trades {N.index.min().date()} -> {N.index.max().date()} | skipped {N.attrs['skipped']} | cost per trade median {N['cost'].median():.4f}R") pr(row("gross", N["gross"])) pr(row("net futures 1x", N["R"])) pr(row("net futures 2x", N["R2"])) pr(row("OOS 2023-03-18 .. 2026-10-09 net", N[N.index >= OOS0]["R"])) c = pd.concat([T["gross"].rename("cfd"), N["gross"].rename("nq")], axis=1).dropna() pr(f" common days with dataset 1: {len(c)} | corr of daily gross R {c['cfd'].corr(c['nq']):.3f}") pr(row("dataset 1 on the same days, net CFD", T.loc[c.index, "R"])) pr(" by year (net, sum R / t): " + " ".join(f"{yv}:{g['R'].sum():+.1f}/{tstat(g['R']):+.1f}" for yv, g in N.groupby("year"))) pr("\n4. REPORT-ONLY") for lab, wcol in [("Threshold only", "W_thr"), ("Calendar only (overlaps X-TILTEOM / P102)", "C")]: X = trades(D, S, wcol=wcol) pr(row(f"{lab}, CFD net 1x", X["R"]) + f" | gross t={tstat(X['gross']):+.2f}") X = trades(DN, S, wcol=wcol) pr(row(f"{lab}, NQ 2016-26 net", X["R"])) X = trades(book(sh(mid, -1), sh(spr, -1), 611, CFD_PTS, ref), S) pr(row("entry 10:00 ET (10:11 by the news rule), CFD net", X["R"]) + f" | gross t={tstat(X['gross']):+.2f}") for lab, DD in [("CFD 2005-2026", D), ("NQ 2016-2026", DN)]: Q = DD.join(S.set_index("trade_day")["W"], how="inner", validate="1:1") Q = Q[(Q.index >= START) & (Q["W"] != 0)].dropna(subset=["prev_c", "c", "entry", "exit"]) Q = Q[Q["n"] >= 360] w = Q["W"] parts = {"close(t) -> close(t+1)": Q["c"] / Q["prev_c"] - 1, "overnight close(t) -> 09:35": Q["entry"] / Q["prev_c"] - 1, "09:35 -> 15:45 (the traded window, no stop)": Q["exit"] / Q["entry"] - 1} pr(f" {lab}: W x return, bp per day, gross, no cost (n={len(Q)}): " + " | ".join(f"{kk} {1e4 * (w * v).mean():+.2f} bp t={tstat(w * v):+.2f}" for kk, v in parts.items())) if "QQQ" not in px.columns: pass t2, to, tn = tstat(T["R2"]), tstat(oos["R"]), tstat(N["R"]) k = {"K1": t1 >= 2.0 and T["R"].mean() > 0, "K2": to > 1.0 and oos["R"].mean() > 0, "K3": t2 > 1.0, "K4": abs(rho_all) < 0.20 and abs(rho_w) < 0.20, "K5": p_perm < 0.05, "K6": h1["R"].mean() > 0 and h2["R"].mean() > 0, "K7": tn > 1.0 and N["R"].mean() > 0} pr(f"\nGATES K1 full CFD net t={t1:+.2f} (n={len(T)}) | K2 OOS t={to:+.2f} (n={len(oos)}) | K3 2x t={t2:+.2f} | K4 rho {rho_all:+.3f} / worst-5% {rho_w:+.3f} | " f"K5 p={p_perm:.3f} | K6 halves {h1['R'].mean():+.4f}R / {h2['R'].mean():+.4f}R | K7 NQ t={tn:+.2f} (n={len(N)})") pr(" " + " ".join(f"{a_} {'PASS' if b_ else 'FAIL'}" for a_, b_ in k.items())) pr(f"PROMOTION BAR t >= 3.43: {'REACHED (OOS + LEAN still required)' if t1 >= 3.43 else 'NOT REACHED'}") sd = T["R"].std() pr(f"POWER: at SD {sd:.3f}R and {len(T) / ((T.index.max() - T.index.min()).days / 365.25):.0f} trades/year, t = 2 on {len(T)} trades needs a mean of {2 * sd / math.sqrt(len(T)):+.4f}R; observed {T['R'].mean():+.4f}R") pr("VERDICT: " + ("FLAG (forward paper only)" if all(k.values()) else "KILL / REJECT (failed: " + ", ".join(a_ for a_, b_ in k.items() if not b_) + ")")) open(os.path.join(LAB, "_rebal01.out"), "w", encoding="utf-8").write("\n".join(OUT) + "\n") if __name__ == "__main__": main()