""" zscore_live_vs_backtest_v2.py ============================== Directly compares the z-score the live system is computing against what the validated backtest pipeline shows for the same dates (14 May - 4 Jul 2026, post-v3 upgrade). If they match → z-score is computing correctly, market is just quiet If they diverge → there is a bug in the live pipeline Also checks: 1. Are the rate files on the VPS current and correct? 2. Is the beta window computing the same as the backtest? 3. What is the historical percentile of current z-score levels? 4. Has there ever been a 2-week period with no signal in the backtest? Run from project root on HOME MACHINE: python src/research/zscore_live_vs_backtest_v2.py """ import pandas as pd import numpy as np from pathlib import Path import sys from datetime import datetime, timedelta BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" RATES_DIR = BASE_PATH / "data" / "raw" / "rates" TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" if str(SRC_PATH) not in sys.path: sys.path.append(str(SRC_PATH)) LIVE_START = pd.Timestamp("2026-05-14") LIVE_END = pd.Timestamp("2026-07-04") THRESHOLD = 2.75 BETA_WINDOW = 120 ZSCORE_WINDOW = 60 def main(): print("=" * 70) print("Z-SCORE VALIDATION — Live vs Backtest Pipeline") print(f" Checking period: {LIVE_START.date()} to {LIVE_END.date()}") print("=" * 70) # ── PART 1: Load backtest z-score series ───────────────────────────────── print(f"\n{'─'*70}") print("PART 1: BACKTEST Z-SCORE FOR LIVE PERIOD") print(f"{'─'*70}") try: from features.spot_lag_v3 import get_model_ready_spot_lag_v3 signal_df = get_model_ready_spot_lag_v3().copy() signal_df["datetime"] = pd.to_datetime(signal_df["datetime"]) signal_df = signal_df.sort_values("datetime").reset_index(drop=True) z_col = "lag_zscore_24h_v3" print(f" Loaded backtest signal series: {len(signal_df):,} rows") except Exception as e: print(f" ERROR: {e}"); return # Filter to live period live_period = signal_df[ (signal_df["datetime"] >= LIVE_START) & (signal_df["datetime"] <= LIVE_END) ].copy() if live_period.empty: print(" WARNING: No backtest data for live period dates") print(" This means the backtest data ends before 16 Mar 2026") print(" The live z-score cannot be directly compared") else: print(f"\n Backtest z-scores for {LIVE_START.date()} to {LIVE_END.date()}:") print(f"\n {'Datetime':<22}{'Z-score':>10}{'Signal?':>10}") print(f" {'─'*44}") for _, row in live_period.iterrows(): z = float(row[z_col]) sig = "LONG" if z >= THRESHOLD else ("SHORT" if z <= -THRESHOLD else "—") flag = " ← SIGNAL" if abs(z) >= THRESHOLD else "" print(f" {str(row['datetime']):<22}{z:>10.3f}{sig:>10}{flag}") max_z = live_period[z_col].abs().max() mean_z = live_period[z_col].abs().mean() signals_in_period = (live_period[z_col].abs() >= THRESHOLD).sum() print(f"\n Max |z| in period : {max_z:.3f}") print(f" Mean |z| in period : {mean_z:.3f}") print(f" Signals fired : {signals_in_period}") if signals_in_period == 0: print(f"\n ✓ BACKTEST ALSO SHOWS NO SIGNALS in this period") print(f" This confirms the quiet period is REAL — not a live pipeline bug") else: print(f"\n ⚠ BACKTEST SHOWS {signals_in_period} SIGNAL(S) that live system missed") print(f" This suggests a pipeline divergence — investigate further") # ── PART 2: Historical context for current z-score levels ───────────────── print(f"\n{'─'*70}") print("PART 2: HOW RARE IS THE CURRENT Z-SCORE LEVEL?") print(f"{'─'*70}") # Current z-score from live log — approximate from recent readings # (user reported: -0.87 to +1.18 range over past 2 weeks) current_z_approx = 0.5 # conservative midpoint of observed range all_z = signal_df[z_col].dropna() pct_below_1 = (all_z.abs() < 1.0).mean() * 100 pct_below_175 = (all_z.abs() < 1.75).mean() * 100 pct_below_275 = (all_z.abs() < THRESHOLD).mean() * 100 pct_above_275 = (all_z.abs() >= THRESHOLD).mean() * 100 print(f""" Historical z-score distribution (22yr backtest): |z| < 1.0 (very quiet) : {pct_below_1:.1f}% of hours |z| < 1.75 (below watch) : {pct_below_175:.1f}% of hours |z| < 2.75 (no signal) : {pct_below_275:.1f}% of hours |z| >= 2.75 (signal!) : {pct_above_275:.1f}% of hours The current range of 0.5-1.2 z-score puts us in the bottom {pct_below_175:.0f}% of all historical readings. This IS a quiet period by historical standards. """) # ── PART 3: Two-week zero-signal periods in history ─────────────────────── print(f"\n{'─'*70}") print("PART 3: HOW OFTEN HAS THE MODEL GONE 2 WEEKS WITHOUT A SIGNAL?") print(f"{'─'*70}") # Load trade log trade_log = None for fname in ["trades_real_costs.csv", "trades_eurusd_final.csv"]: p = TRADES_DIR / fname if p.exists(): trade_log = pd.read_csv(p, parse_dates=["entry_time"]) break if trade_log is not None: trade_log = trade_log.sort_values("entry_time").reset_index(drop=True) # Find gaps between consecutive trades gaps = [] for i in range(1, len(trade_log)): gap_days = (trade_log.iloc[i]["entry_time"] - trade_log.iloc[i-1]["entry_time"]).total_seconds() / 86400 if gap_days >= 14: gaps.append({ "from" : trade_log.iloc[i-1]["entry_time"], "to" : trade_log.iloc[i]["entry_time"], "days" : gap_days, "weeks" : gap_days / 7, }) print(f"\n Gaps of 14+ days between consecutive trades:") print(f"\n {'From':<14}{'To':<14}{'Days':>6}{'Weeks':>7} Notes") print(f" {'─'*55}") if gaps: for g in sorted(gaps, key=lambda x: x["days"], reverse=True): note = "← LONGEST" if g["days"] == max(x["days"] for x in gaps) else "" print(f" {str(g['from'].date()):<14}{str(g['to'].date()):<14}" f"{g['days']:>6.0f}{g['weeks']:>7.1f} {note}") else: print(" None found — no 14+ day gaps in 22yr backtest") # Current gap last_trade = trade_log["entry_time"].max() current_gap = (pd.Timestamp.now() - last_trade).days print(f"\n Last backtest trade : {last_trade.date()}") print(f" Current gap (live) : ~10 days since live start (16 Mar)") print(f"\n Note: Live system started 16 Mar — no live trades yet") print(f" The backtest shows the model's last trade was {last_trade.date()}") # ── PART 4: Rate file freshness check ───────────────────────────────────── print(f"\n{'─'*70}") print("PART 4: RATE FILE VALIDATION") print(f"{'─'*70}") all_ok = True for fname, col in [("us2y.csv","us2y"),("us10y.csv","us10y"), ("de2y.csv","de2y"),("de10y.csv","de10y")]: path = RATES_DIR / fname if not path.exists(): print(f" ✗ {fname} MISSING"); all_ok = False; continue df = pd.read_csv(path) df["date"] = pd.to_datetime(df.iloc[:,0]) last = df["date"].max() days_old = (pd.Timestamp.now() - last).days last_val = float(df.iloc[-1,1]) status = "✓" if days_old <= 5 else "⚠" if days_old > 5: all_ok = False print(f" {status} {fname:<12} Latest: {last.date()} " f"({days_old}d old) Last value: {last_val:.3f}%") if all_ok: print(f"\n ✓ All rate files are current and valid") else: print(f"\n ⚠ Some rate files may be stale — run auto_rates_loader.py") # ── PART 5: Quick z-score recompute from scratch ────────────────────────── print(f"\n{'─'*70}") print("PART 5: INDEPENDENT Z-SCORE RECOMPUTE (last 30 days)") print(f"{'─'*70}") print(" Recomputing z-score from raw rate files + price data...") try: from ingestion.price_loader_15m import load_eurusd_15m m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) # Resample to 1H m15["datetime"] = pd.to_datetime(m15["datetime"]) m15 = m15.set_index("datetime") h1 = m15["close"].resample("1h").last().dropna().reset_index() h1.columns = ["datetime","close"] h1["date"] = h1["datetime"].dt.normalize() # Load rates us2y = pd.read_csv(RATES_DIR/"us2y.csv") de2y = pd.read_csv(RATES_DIR/"de2y.csv") us2y["date"] = pd.to_datetime(us2y.iloc[:,0]) de2y["date"] = pd.to_datetime(de2y.iloc[:,0]) us2y_col = us2y.columns[1] de2y_col = de2y.columns[1] # Daily merge daily = h1.groupby("date", as_index=False)["close"].last() daily["date"] = pd.to_datetime(daily["date"]) daily = daily.merge(us2y[["date",us2y_col]], on="date", how="left") daily = daily.merge(de2y[["date",de2y_col]], on="date", how="left") daily = daily.ffill().dropna() daily["spread"] = daily[us2y_col] - daily[de2y_col] daily["spread_chg"] = daily["spread"].diff() daily["ret_1d"] = daily["close"].pct_change() # Rolling beta from scipy import stats betas = [] for i in range(len(daily)): if i < BETA_WINDOW: betas.append(np.nan); continue window = daily.iloc[i-BETA_WINDOW:i].dropna( subset=["ret_1d","spread_chg"]) if len(window) < 30: betas.append(np.nan); continue slope,_,_,_,_ = stats.linregress( window["spread_chg"], window["ret_1d"]) betas.append(slope) daily["beta"] = betas daily["predicted"] = daily["beta"] * daily["spread_chg"] daily["return_24h"] = daily["close"].pct_change(1) daily["lag_gap"] = daily["predicted"] - daily["return_24h"] # Merge back to hourly h1["date"] = pd.to_datetime(h1["date"]) h1m = h1.merge(daily[["date","lag_gap"]], on="date", how="left").ffill() h1m["zscore"] = ((h1m["lag_gap"] - h1m["lag_gap"].rolling(ZSCORE_WINDOW).mean()) / h1m["lag_gap"].rolling(ZSCORE_WINDOW).std()) # Show last 14 days hourly z-scores during session recent = h1m[ (h1m["datetime"] >= LIVE_START) & (h1m["datetime"] <= LIVE_END) & (h1m["datetime"].dt.hour.between(7,16)) ].dropna(subset=["zscore"]) if not recent.empty: print(f"\n Independently computed z-scores (session hours only):") print(f"\n {'Datetime':<22}{'Z-score (indep)':>17}{'Signal?':>10}") print(f" {'─'*52}") for _, row in recent.iterrows(): z = float(row["zscore"]) sig = "LONG" if z >= THRESHOLD else ("SHORT" if z <= -THRESHOLD else "—") flag = " ← SIGNAL" if abs(z) >= THRESHOLD else "" print(f" {str(row['datetime']):<22}{z:>17.3f}{sig:>10}{flag}") max_iz = recent["zscore"].abs().max() print(f"\n Max |z| independently : {max_iz:.3f}") # Compare with backtest if available if not live_period.empty: max_bz = live_period[z_col].abs().max() diff = abs(max_iz - max_bz) print(f" Max |z| from backtest : {max_bz:.3f}") print(f" Difference : {diff:.3f}") if diff < 0.5: print(f"\n ✓ PIPELINES MATCH — z-scores are consistent") print(f" The live system is computing correctly") print(f" Market is genuinely quiet — no pipeline bug") else: print(f"\n ⚠ PIPELINES DIVERGE — investigate live_signal_monitor.py") else: print(" No data for live period in independent computation") except Exception as e: print(f" ERROR in independent recompute: {e}") # ── SUMMARY ─────────────────────────────────────────────────────────────── print(f"\n{'='*70}") print("SUMMARY") print(f"{'='*70}") print(f""" Run this checklist mentally: 1. Did backtest show any signals in 16-25 Mar? → If NO: market is genuinely quiet, live system is correct → If YES: live system missed signals, investigate pipeline 2. Are rate files current (< 5 days old)? → If NO: update rates, z-score may be stale 3. Do independent and backtest z-scores match (diff < 0.5)? → If YES: pipeline is healthy → If NO: live computation has a bug 4. Is current z-score in the bottom quartile historically? → If YES: quiet macro regime, be patient → If NO: z-score should be higher, investigate If all four are green → sit tight, the model is working correctly. If any are red → investigate that specific component. """) if __name__ == "__main__": main()