""" live_vs_backtest_zscore_v1.py ============================== Compares the z-score the live system is computing against what the historical backtest pipeline would show for the same recent dates. If they match: the signal is genuinely quiet — market regime explanation. If they diverge: there is a pipeline difference between live and backtest. Also shows: - Historical signal frequency by month to show natural clustering - Current z-score context vs last 12 months of backtest z-scores - How many signals fired in equivalent periods of the backtest Run on HOME machine (has full historical data): python src/research/live_vs_backtest_zscore_v1.py """ import pandas as pd import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib as mpl mpl.rcParams["text.usetex"] = False import matplotlib.pyplot as plt import matplotlib.gridspec as gridspec from pathlib import Path import sys BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" if str(SRC_PATH) not in sys.path: sys.path.append(str(SRC_PATH)) RATES_DIR = BASE_PATH / "data" / "raw" / "rates" TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" CHARTS_DIR = BASE_PATH / "data" / "processed" THRESHOLD = 2.75 def main(): print("=" * 70) print("LIVE vs BACKTEST Z-SCORE COMPARISON") print("=" * 70) # ── Load backtest signal series ─────────────────────────────────────────── print("\nLoading backtest signal series...") 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) print(f" Loaded {len(signal_df):,} hourly z-score observations") z_col = "lag_zscore_24h_v3" except Exception as e: print(f" Could not load backtest signal series: {e}") signal_df = None z_col = None # ── Load validated 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","exit_time"]) print(f" Trade log: {fname} ({len(trade_log):,} trades)") break # ── PART 1: Historical signal frequency by month ────────────────────────── print(f"\n{'─'*70}") print("PART 1: HISTORICAL SIGNAL FREQUENCY BY MONTH") print(f" Shows natural clustering — some months have 0 signals, others 10+") print(f"{'─'*70}") if signal_df is not None: # Count signals per month signals_only = signal_df[signal_df[z_col].abs() >= THRESHOLD].copy() signals_only["month"] = signals_only["datetime"].dt.to_period("M") signals_only["hour"] = signals_only["datetime"].dt.hour # Only count session hours (7-16 EET = roughly 5-14 UTC) session_signals = signals_only[ signals_only["hour"].between(5, 16) ] monthly_signals = session_signals.groupby("month").size() monthly_signals = monthly_signals.reset_index() monthly_signals.columns = ["month", "signals"] print(f"\n Monthly signal counts (threshold crossings >= 2.75):") print(f"\n {'Month':<12}{'Signals':>10} {'Bar chart'}") print(f" {'─'*50}") for _, row in monthly_signals.tail(24).iterrows(): bar = "█" * int(row["signals"]) print(f" {str(row['month']):<12}{int(row['signals']):>10} {bar}") avg_monthly = monthly_signals["signals"].mean() avg_weekly = avg_monthly / 4.33 zero_months = (monthly_signals["signals"] == 0).sum() print(f"\n Avg signals per month : {avg_monthly:.1f}") print(f" Avg signals per week : {avg_weekly:.1f}") print(f" Months with 0 signals : {zero_months}/{len(monthly_signals)}") print(f" Max in one month : {monthly_signals['signals'].max()}") print(f" Min in one month : {monthly_signals['signals'].min()}") # ── PART 2: Recent z-score vs historical context ────────────────────────── print(f"\n{'─'*70}") print("PART 2: CURRENT Z-SCORE IN HISTORICAL CONTEXT") print(f"{'─'*70}") if signal_df is not None: recent_90d = signal_df[ signal_df["datetime"] >= signal_df["datetime"].max() - pd.Timedelta(days=90) ] current_z = float(signal_df[z_col].iloc[-1]) current_dt = signal_df["datetime"].iloc[-1] hist_mean = float(signal_df[z_col].mean()) hist_std = float(signal_df[z_col].std()) pct_above = (signal_df[z_col].abs() >= THRESHOLD).mean() * 100 recent_max = float(recent_90d[z_col].abs().max()) recent_mean = float(recent_90d[z_col].abs().mean()) print(f"\n Current z-score : {current_z:.3f} (as of {current_dt})") print(f" 22yr historical mean : {hist_mean:.3f}") print(f" 22yr historical std : {hist_std:.3f}") print(f" % of hours >= 2.75 : {pct_above:.1f}% of all hours") print(f" Last 90 days max |z| : {recent_max:.3f}") print(f" Last 90 days mean |z| : {recent_mean:.3f}") # How does this week compare to typical quiet periods week_ago = signal_df["datetime"].max() - pd.Timedelta(days=5) this_week = signal_df[signal_df["datetime"] >= week_ago] week_max_z = float(this_week[z_col].abs().max()) week_sigs = (this_week[z_col].abs() >= THRESHOLD).sum() print(f"\n This week max |z| : {week_max_z:.3f}") print(f" This week signals : {week_sigs}") # Find comparable quiet weeks in history signal_df["week"] = signal_df["datetime"].dt.to_period("W") weekly_max = signal_df.groupby("week")[z_col].apply(lambda x: x.abs().max()) quiet_weeks = (weekly_max < week_max_z + 0.5).sum() total_weeks = len(weekly_max) print(f"\n Weeks in history with similar or lower peak z : " f"{quiet_weeks}/{total_weeks} ({quiet_weeks/total_weeks*100:.0f}%)") print(f" This week is {'UNUSUALLY QUIET' if quiet_weeks/total_weeks < 0.2 else 'WITHIN NORMAL RANGE'}") # ── PART 3: What would have happened to LIVE system this week ───────────── print(f"\n{'─'*70}") print("PART 3: EQUIVALENT BACKTEST PERIOD") print(f" What did the backtest show for the same calendar week in past years?") print(f"{'─'*70}") if trade_log is not None and signal_df is not None: # Current week of year current_week_num = pd.Timestamp.now().week current_month = pd.Timestamp.now().month # Find same week in each historical year trade_log["week_num"] = trade_log["entry_time"].dt.isocalendar().week.astype(int) trade_log["year"] = trade_log["entry_time"].dt.year same_week_hist = trade_log[ trade_log["week_num"].between(current_week_num - 1, current_week_num + 1) ] print(f"\n Current week number: {current_week_num}") print(f"\n Trades in same week (±1) across all historical years:") print(f"\n {'Year':<8}{'Trades':>8}{'WR%':>8} Notes") print(f" {'─'*40}") ret_col = "return_real" if "return_real" in trade_log.columns else "return" for year, grp in same_week_hist.groupby("year"): wr = grp[ret_col].gt(0).mean() * 100 if len(grp) > 0 else 0 print(f" {year:<8}{len(grp):>8}{wr:>8.0f}%") avg_trades_this_week = same_week_hist.groupby("year").size().mean() print(f"\n Avg trades in this week historically: {avg_trades_this_week:.1f}") if avg_trades_this_week < 2: print(f" This week is HISTORICALLY QUIET — low signal count is expected") elif avg_trades_this_week >= 3: print(f" This week should historically have {avg_trades_this_week:.0f} trades") print(f" If live system has 0, this warrants investigation") # ── PART 4: Chart ───────────────────────────────────────────────────────── if signal_df is not None: print(f"\nGenerating chart...") BG = "#0d1117" PANEL = "#161b22" GRID = "#21262d" TEXT = "#e6edf3" MUTED = "#8b949e" GREEN = "#00ff88" RED = "#ff4444" GOLD = "#ffd700" CYAN = "#00d4ff" fig = plt.figure(figsize=(20, 14), facecolor=BG) gs = gridspec.GridSpec(2, 2, figure=fig, hspace=0.45, wspace=0.30, left=0.07, right=0.97, top=0.92, bottom=0.05) def style(ax, title): ax.set_facecolor(PANEL) ax.set_title(title, color=TEXT, fontsize=10, fontweight="bold", pad=10) ax.tick_params(colors=TEXT, labelsize=8) for sp in ax.spines.values(): sp.set_color(GRID) ax.grid(True, color=GRID, linewidth=0.5, alpha=0.6) ax.yaxis.label.set_color(TEXT) # Panel 1: Z-score last 90 days ax1 = fig.add_subplot(gs[0, :]) style(ax1, "Z-Score Last 90 Days vs Threshold ±2.75") r90 = signal_df[ signal_df["datetime"] >= signal_df["datetime"].max() - pd.Timedelta(days=90) ].copy() ax1.plot(r90["datetime"], r90[z_col], color=CYAN, linewidth=0.8, alpha=0.8) ax1.fill_between(r90["datetime"], r90[z_col], 0, where=r90[z_col] >= THRESHOLD, color=GREEN, alpha=0.4, label="Long signal") ax1.fill_between(r90["datetime"], r90[z_col], 0, where=r90[z_col] <= -THRESHOLD, color=RED, alpha=0.4, label="Short signal") ax1.axhline(y=THRESHOLD, color=GREEN, linewidth=1.2, linestyle="--", alpha=0.8) ax1.axhline(y=-THRESHOLD, color=RED, linewidth=1.2, linestyle="--", alpha=0.8) ax1.axhline(y=0, color=MUTED, linewidth=0.6, alpha=0.5) # Mark current z ax1.scatter([r90["datetime"].iloc[-1]], [r90[z_col].iloc[-1]], color=GOLD, s=80, zorder=8) ax1.annotate(f"Now: {r90[z_col].iloc[-1]:.3f}", xy=(r90["datetime"].iloc[-1], r90[z_col].iloc[-1]), xytext=(-80, 15), textcoords="offset points", color=GOLD, fontsize=9, fontweight="bold") ax1.set_ylabel("Z-Score", fontsize=9) ax1.legend(facecolor=PANEL, edgecolor=GRID, labelcolor=TEXT, fontsize=8) # Panel 2: Monthly signal count ax2 = fig.add_subplot(gs[1, 0]) style(ax2, "Monthly Signal Count (last 3 years)") last3yr = monthly_signals[ monthly_signals["month"] >= monthly_signals["month"].max() - 35 ] colors2 = [GREEN if v > 0 else RED for v in last3yr["signals"]] ax2.bar(range(len(last3yr)), last3yr["signals"], color=colors2, alpha=0.8) ax2.axhline(y=avg_monthly, color=GOLD, linewidth=1.2, linestyle="--", alpha=0.8, label=f"Avg {avg_monthly:.1f}/month") ax2.set_xticks(range(len(last3yr))) ax2.set_xticklabels( [str(m)[-7:] for m in last3yr["month"]], rotation=45, fontsize=6 ) ax2.set_ylabel("Signals", fontsize=9) ax2.legend(facecolor=PANEL, edgecolor=GRID, labelcolor=TEXT, fontsize=8) # Panel 3: Z-score distribution ax3 = fig.add_subplot(gs[1, 1]) style(ax3, "Z-Score Distribution — How Rare is a Signal?") z_vals = signal_df[z_col].dropna().values ax3.hist(z_vals, bins=100, color=CYAN, alpha=0.7, edgecolor="none", density=True) ax3.axvline(x=THRESHOLD, color=GREEN, linewidth=1.5, linestyle="--", label=f"Long threshold +{THRESHOLD}") ax3.axvline(x=-THRESHOLD, color=RED, linewidth=1.5, linestyle="--", label=f"Short threshold -{THRESHOLD}") ax3.axvline(x=float(signal_df[z_col].iloc[-1]), color=GOLD, linewidth=2, label="Current z-score") ax3.set_xlabel("Z-Score", fontsize=9) ax3.set_ylabel("Density", fontsize=9) ax3.legend(facecolor=PANEL, edgecolor=GRID, labelcolor=TEXT, fontsize=8) ax3.text(0.02, 0.95, f"Signals = {pct_above:.1f}% of all hours", transform=ax3.transAxes, color=TEXT, fontsize=8, va="top") fig.suptitle( f"Live vs Backtest Z-Score Analysis | Current z = " f"{float(signal_df[z_col].iloc[-1]):.3f} | " f"Threshold ±{THRESHOLD} | " f"Avg {avg_weekly:.1f} signals/week historically", color=TEXT, fontsize=12, fontweight="bold", y=0.97 ) chart_path = CHARTS_DIR / "zscore_context_chart.png" plt.savefig(chart_path, dpi=150, bbox_inches="tight", facecolor=fig.get_facecolor()) plt.close() print(f" Chart saved: {chart_path}") # ── PART 5: Diagnosis decision tree ─────────────────────────────────────── print(f"\n{'='*70}") print("DIAGNOSIS GUIDE — If Consistently < 2 Trades/Week After 4 Weeks") print(f"{'='*70}") print(f""" Step 1: Check if z-scores are computing correctly Run this script — if monthly chart shows signals firing in backtset for the same period but not live, the pipeline has diverged. Fix: compare live z-score log vs backtest z-score for same dates. Step 2: Check if threshold is too tight for current regime If z-scores are reaching 2.0-2.5 but not 2.75, consider testing a lower threshold e.g. 2.50. Run threshold_regime_filter_v1.py. Step 3: Check rate data freshness If DE rates are stale by more than 5 days the spread signal will lag reality. Run: python src/ingestion/auto_rates_loader.py Step 4: Check if current macro regime is structurally quiet Low Fed/ECB divergence = fewer dislocations = fewer signals. This is not a model failure — it is the model being selective. Check: is the US-DE 2Y spread compressed vs historical levels? Step 5: Check signal definition has not drifted Compare BETA_WINDOW (120), SMOOTH_SPAN (20), ZSCORE_WINDOW (60) in live script vs what the backtest used. Any difference changes the z-score distribution. """) if __name__ == "__main__": main()