""" kalman_zscore_backtest_v1.py ============================== Tests whether applying a Kalman filter to the yield spread or z-score improves the validated EURUSD model. Unlike the Markov backtest (which filtered existing trades), this runs a FULL fresh backtest with Kalman-smoothed signals — generating a new trade set from scratch and comparing against the validated 4,019 trades. Three approaches tested: 1. Validated baseline (raw z-score, no Kalman) 2. Kalman on spread — smooth the US-DE 2Y spread BEFORE z-score calc 3. Kalman on z-score — smooth the z-score series directly For each approach, multiple Q/R tuning ratios are tested: Q = process noise (how fast true state changes) R = observation noise (how noisy the measurements are) Q/R high → light smoothing (tracks signal closely) Q/R low → heavy smoothing (slower, cleaner signal) Key question: Does Kalman reduce noise-triggered trades enough to improve Sharpe and win rate on the validated holdout? Run from project root (takes ~30 mins): python src/research/kalman_zscore_backtest_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys import warnings warnings.filterwarnings('ignore') BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" PROC_PATH = BASE_PATH / "data" / "processed" if str(SRC_PATH) not in sys.path: sys.path.insert(0, str(SRC_PATH)) # ── Validated model parameters — DO NOT CHANGE ─────────────────────────────── THRESHOLD = 2.75 FIB = 0.786 TP_PCT = 0.0020 STOP_PCT = 0.0025 ZSCORE_EXIT = 1.5 SPREAD_COST = 0.0001 HOLD_HOURS = 52 NOTIONAL = 300_000 Z_WINDOW = 60 # rolling z-score window (bars) BETA_WINDOW = 120 # rolling beta window (bars) ZSCORE_BANDS = [(2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0)] def get_mult(z_abs): for lo, hi, m in ZSCORE_BANDS: if lo <= z_abs < hi: return m return 2.0 # ── Kalman Filter ───────────────────────────────────────────────────────────── class KalmanFilter1D: """ 1D Kalman filter (random walk state model). Parameters ---------- Q : process noise variance How quickly the true underlying state can change. Higher Q = filter tracks observations more closely. R : observation noise variance How noisy the measurements are. Higher R = measurements are less trusted, more smoothing. The Q/R ratio determines smoothing intensity: Q/R = 1.0 → minimal smoothing Q/R = 0.1 → light smoothing Q/R = 0.01 → heavy smoothing """ def __init__(self, Q=0.01, R=1.0): self.Q = Q self.R = R def smooth(self, observations): obs = np.array(observations, dtype=float) n = len(obs) x = np.zeros(n) P = np.zeros(n) x[0] = obs[0] P[0] = 1.0 for t in range(1, n): xp = x[t-1] Pp = P[t-1] + self.Q K = Pp / (Pp + self.R) x[t] = xp + K * (obs[t] - xp) P[t] = (1 - K) * Pp return x # ── Signal generation ───────────────────────────────────────────────────────── def compute_zscore_raw(spread_series, window=Z_WINDOW): """Validated raw z-score computation — matches live model exactly.""" s = pd.Series(spread_series) mu = s.rolling(window).mean() sd = s.rolling(window).std() z = (s - mu) / sd return z.fillna(0).values def compute_zscore_kalman_spread(spread_series, Q, R, window=Z_WINDOW): """Apply Kalman to spread first, then compute z-score.""" kf = KalmanFilter1D(Q=Q, R=R) spread_smooth = kf.smooth(spread_series) return compute_zscore_raw(spread_smooth, window) def compute_zscore_kalman_direct(spread_series, Q, R, window=Z_WINDOW): """Compute raw z-score, then apply Kalman to smooth the z-score.""" z_raw = compute_zscore_raw(spread_series, window) kf = KalmanFilter1D(Q=Q, R=R) return kf.smooth(z_raw) # ── Backtest execution ──────────────────────────────────────────────────────── def run_backtest_with_zscore(z_series, signal_df, m15_idx, label): """ Run full trade simulation using provided z-score series. Uses the exact same entry/exit logic as the validated model. """ trades = [] last_exit = None no_entry = 0 # Build fast z-score lookup indexed by datetime z_lookup = pd.Series(z_series, index=signal_df["datetime"]).sort_index() for _, row in signal_df.iterrows(): dt = row["datetime"] z = float(z_lookup.get(dt, 0.0)) if abs(z) < THRESHOLD: continue if last_exit is not None and dt <= last_exit: continue direction = 1 if z >= THRESHOLD else -1 # Get signal bar OHLC bar_end = dt + pd.Timedelta(hours=1) - pd.Timedelta(minutes=1) bar_slice = m15_idx.loc[dt:bar_end] if bar_slice.empty: no_entry += 1 continue bh = float(bar_slice["high"].max()) bl = float(bar_slice["low"].min()) bc = float(bar_slice["close"].iloc[-1]) # Fibonacci entry target if direction == 1: pull = bc - bl if pull <= 0.00005: no_entry += 1; continue target = bc - FIB * pull else: pull = bh - bc if pull <= 0.00005: no_entry += 1; continue target = bc + FIB * pull # Search 6-hour entry window entry_slice = m15_idx.loc[ dt + pd.Timedelta(minutes=1):dt + pd.Timedelta(hours=6)] entry_price = entry_time = None for edt, ebar in entry_slice.iterrows(): if direction == 1 and float(ebar["low"]) <= target: entry_price = target; entry_time = edt; break elif direction == -1 and float(ebar["high"]) >= target: entry_price = target; entry_time = edt; break if entry_price is None: no_entry += 1; continue # TP / SL tp = entry_price*(1+TP_PCT) if direction==1 else entry_price*(1-TP_PCT) sl = entry_price*(1-STOP_PCT) if direction==1 else entry_price*(1+STOP_PCT) # Hold period exit hold = m15_idx.loc[ entry_time + pd.Timedelta(minutes=1): entry_time + pd.Timedelta(hours=HOLD_HOURS)] exit_price = None exit_reason = "hold" for hdt, hbar in hold.iterrows(): hr = hdt.replace(minute=0, second=0, microsecond=0) if hr in z_lookup.index: hz = float(z_lookup[hr]) if ((direction==1 and hz<=-ZSCORE_EXIT) or (direction==-1 and hz>=ZSCORE_EXIT)): exit_price = float(hbar["close"]) exit_reason = "z_exit"; break if direction == 1: if float(hbar["high"]) >= tp: exit_price = tp; exit_reason = "tp"; break if float(hbar["low"]) <= sl: exit_price = sl; exit_reason = "sl"; break else: if float(hbar["low"]) <= tp: exit_price = tp; exit_reason = "tp"; break if float(hbar["high"]) >= sl: exit_price = sl; exit_reason = "sl"; break if exit_price is None: exit_price = float(hold["close"].iloc[-1]) if not hold.empty else entry_price exit_reason = "hold_expiry" mult = get_mult(abs(z)) pnl = round(((exit_price - entry_price)/entry_price * direction - SPREAD_COST) * mult * NOTIONAL, 2) trades.append({"entry_time": entry_time, "exit_reason": exit_reason, "pnl": pnl, "win": int(pnl > 0), "mult": mult}) last_exit = entry_time + pd.Timedelta(hours=HOLD_HOURS) return trades, no_entry def score(trades, label): """Compute performance metrics from list of trades.""" if not trades: return {"label": label, "n": 0, "wr_pct": 0, "sharpe": 0, "pf": 0, "max_dd": 0, "total": 0, "per_yr": 0, "tp_pct": 0, "sl_pct": 0} df = pd.DataFrame(trades) pnl = df["pnl"].values n = len(pnl) wr = (pnl > 0).mean() * 100 eq = 100_000 + np.cumsum(pnl) peak = np.maximum.accumulate(eq) dd = ((eq - peak) / peak * 100).min() per_yr = n / 22 excess = pnl / 100_000 - 0.04 / per_yr sharpe = (excess.mean() / excess.std() * np.sqrt(per_yr) if excess.std() > 0 else 0) gp = pnl[pnl > 0].sum() gl = abs(pnl[pnl < 0].sum()) pf = round(gp / gl, 2) if gl > 0 else 999 tp_pct = (df["exit_reason"]=="tp").mean()*100 if "exit_reason" in df else 0 sl_pct = (df["exit_reason"]=="sl").mean()*100 if "exit_reason" in df else 0 return {"label": label, "n": n, "wr_pct": round(wr,1), "sharpe": round(sharpe,2), "pf": pf, "max_dd": round(dd,2), "total": round(pnl.sum()), "per_yr": round(per_yr,1), "tp_pct": round(tp_pct,1), "sl_pct": round(sl_pct,1)} def print_results(results, title): print(f"\n{'='*75}") print(title) print(f"{'='*75}") print(f"{'Approach':<42}{'N':>5}{'WR%':>6}{'Sharpe':>8}" f"{'PF':>6}{'MaxDD%':>8}{'Total$':>10}{'Tr/yr':>7}") print(f"{'─'*75}") for r in results: if r.get("n", 0) == 0: continue marker = " ◄ BEST" if r == max(results, key=lambda x: x.get("sharpe",0)) else "" print(f" {r['label']:<40}{r['n']:>5}{r['wr_pct']:>6.1f}" f"{r['sharpe']:>8.2f}{r['pf']:>6.2f}{r['max_dd']:>8.2f}" f"{r['total']:>10,.0f}{r['per_yr']:>7.1f}{marker}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 75) print("KALMAN FILTER Z-SCORE BACKTEST v1") print("Full re-simulation using Kalman-smoothed signals vs validated model") print("=" * 75) # ── Load data ───────────────────────────────────────────────────────────── print("\nStep 1: Loading signal data (1H z-scores)...") 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" Signal rows: {len(signal_df):,}") except Exception as e: print(f" ERROR: {e}"); sys.exit(1) print("\nStep 2: Loading yield spread data for Kalman...") try: # The spread is in the signal_df already — extract it # Also try to load raw rate files rates_dir = BASE_PATH / "data" / "raw" / "rates" us2y_path = rates_dir / "us2y.csv" de2y_path = rates_dir / "de2y.csv" if us2y_path.exists() and de2y_path.exists(): us2y = pd.read_csv(us2y_path, header=None, names=["date","us2y"], parse_dates=["date"]).set_index("date")["us2y"] de2y = pd.read_csv(de2y_path, header=None, names=["date","de2y"], parse_dates=["date"]).set_index("date")["de2y"] spread_daily = (us2y - de2y).dropna() print(f" Daily spread: {len(spread_daily):,} observations") # Reindex to hourly using forward fill signal_df["date"] = signal_df["datetime"].dt.date.astype(str) spread_map = spread_daily.reset_index() spread_map.columns = ["date","spread"] spread_map["date"] = spread_map["date"].astype(str) signal_df = signal_df.merge(spread_map, on="date", how="left") signal_df["spread"] = signal_df["spread"].ffill() spread_series = signal_df["spread"].values print(f" Spread aligned to {len(spread_series):,} hourly bars") else: # Use z-score series as proxy for spread print(" Rate CSV not found — using z-score as spread proxy") spread_series = signal_df[z_col].values except Exception as e: print(f" WARNING: {e} — using z-score as proxy") spread_series = signal_df[z_col].values print("\nStep 3: Loading 15M price data...") try: from ingestion.price_loader_15m import load_eurusd_15m m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) m15["datetime"] = pd.to_datetime(m15["datetime"]) m15_idx = m15.set_index("datetime").sort_index() print(f" 15M bars: {len(m15):,}") except Exception as e: print(f" ERROR: {e}"); sys.exit(1) # ── Baseline: validated raw z-score ────────────────────────────────────── print("\nStep 4: Running BASELINE (validated raw z-score)...") z_raw = signal_df[z_col].values trades_base, _ = run_backtest_with_zscore(z_raw, signal_df, m15_idx, "Baseline") r_base = score(trades_base, "Baseline (validated raw z-score)") print(f" Trades: {r_base['n']:,} WR: {r_base['wr_pct']:.1f}% " f"Sharpe: {r_base['sharpe']:.2f}") print(f" TP: {r_base['tp_pct']:.1f}% SL: {r_base['sl_pct']:.1f}%") # ── Approach A: Kalman on spread, multiple Q/R ratios ──────────────────── print("\nStep 5: Testing Kalman on SPREAD (multiple Q/R ratios)...") results_spread = [r_base] qr_spread = [ (0.0001, 0.01, "Kalman-Spread Q/R=0.01 (heavy smooth)"), (0.001, 0.01, "Kalman-Spread Q/R=0.10 (medium smooth)"), (0.005, 0.01, "Kalman-Spread Q/R=0.50 (light smooth)"), (0.01, 0.01, "Kalman-Spread Q/R=1.00 (minimal smooth)"), ] for Q, R, label in qr_spread: print(f" Testing {label}...") z_kal = compute_zscore_kalman_spread(spread_series, Q, R) trades, _ = run_backtest_with_zscore(z_kal, signal_df, m15_idx, label) r = score(trades, label) results_spread.append(r) print(f" Trades: {r['n']:,} WR: {r['wr_pct']:.1f}% Sharpe: {r['sharpe']:.2f}") # ── Approach B: Kalman on z-score directly ──────────────────────────────── print("\nStep 6: Testing Kalman on Z-SCORE directly...") results_zscore = [r_base] qr_zscore = [ (0.001, 1.0, "Kalman-Zscore Q/R=0.001 (heavy smooth)"), (0.01, 1.0, "Kalman-Zscore Q/R=0.010 (medium smooth)"), (0.1, 1.0, "Kalman-Zscore Q/R=0.100 (light smooth)"), (0.5, 1.0, "Kalman-Zscore Q/R=0.500 (minimal smooth)"), ] for Q, R, label in qr_zscore: print(f" Testing {label}...") z_kal = compute_zscore_kalman_direct(spread_series, Q, R) trades, _ = run_backtest_with_zscore(z_kal, signal_df, m15_idx, label) r = score(trades, label) results_zscore.append(r) print(f" Trades: {r['n']:,} WR: {r['wr_pct']:.1f}% Sharpe: {r['sharpe']:.2f}") # ── Results ─────────────────────────────────────────────────────────────── print_results(results_spread, "KALMAN ON SPREAD — vs Validated Baseline") print_results(results_zscore, "KALMAN ON Z-SCORE — vs Validated Baseline") all_results = results_spread + results_zscore[1:] # skip duplicate baseline print_results(all_results, "ALL APPROACHES — FINAL COMPARISON") # ── Conclusion ──────────────────────────────────────────────────────────── best = max(all_results, key=lambda r: r.get("sharpe", 0)) base = r_base print(f"\n{'='*75}") print("CONCLUSION") print(f"{'='*75}") print(f"\n Baseline (validated) : {base['n']:,} trades " f"WR {base['wr_pct']:.1f}% Sharpe {base['sharpe']:.2f}") print(f" Best Kalman approach : {best['n']:,} trades " f"WR {best['wr_pct']:.1f}% Sharpe {best['sharpe']:.2f}") print(f" Label : {best['label']}") sharpe_gain = best['sharpe'] - base['sharpe'] trade_pct = best['n'] / base['n'] * 100 if sharpe_gain >= 0.3 and trade_pct >= 80: verdict = "IMPLEMENT — Kalman meaningfully improves signal quality" detail = (f"Sharpe +{sharpe_gain:.2f}, keeps {trade_pct:.0f}% of trades") elif sharpe_gain >= 0.1: verdict = "MARGINAL — small improvement, adds complexity, not recommended" else: verdict = "NO IMPROVEMENT — keep validated raw z-score" print(f"\n Verdict: {verdict}") if sharpe_gain >= 0.1: print(f" Detail : {detail if sharpe_gain >= 0.3 else 'Sharpe gain below threshold'}") print(f""" INTERPRETATION ───────────────────────────────────────────────────────────────────── Kalman smoothing reduces noise at the cost of signal lag. If spread Kalman improves results → noise in yield data is hiding cleaner entry points. Worth implementing in live_signal_monitor. If z-score Kalman improves results → z-score is oversensitive to short-term fluctuations. Threshold may need recalibration. If neither improves → current z-score already well-calibrated and Kalman smoothing removes genuine signal along with noise. ───────────────────────────────────────────────────────────────────── """) # Save results out_df = pd.DataFrame(all_results) out_path = PROC_PATH / "kalman_results.csv" out_df.to_csv(out_path, index=False) print(f"Results saved: {out_path.name}") if __name__ == "__main__": main()