""" lag_instability_test_v1.py (v2 — signal-return correlation) ============================================================= TRUE lag instability test with the missing piece added: For each z-window / lag bucket: Correlation(signal, forward return at that horizon) Sharpe of trades triggered at that threshold Smooth degradation across windows = robust ✅ Sharp peak at one window only = overfit ❌ Data used: actual us2y.csv, de2y.csv, jp2y.csv from data/raw/rates/ """ import sys, warnings from pathlib import Path import numpy as np import pandas as pd from scipy import stats warnings.filterwarnings('ignore') BASE = Path(__file__).resolve().parents[2] RATES = BASE / 'data/raw/rates' EU_FILE = BASE / 'data/processed/trades/trades_real_costs.csv' UJ_FILE = BASE / 'data/processed/usdjpy_trades_real_costs.csv' ACCOUNT = 100_000 def load_rate(fname, col): p = RATES / fname if not p.exists(): return None df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] df['date'] = pd.to_datetime(df['date'], errors='coerce') df[col] = pd.to_numeric(df[col], errors='coerce') return df[['date',col]].dropna().sort_values('date').set_index('date') def load_price(fname, col='close'): """Try to find 1H price file.""" candidates = [ BASE / 'data/raw/prices' / fname, BASE / 'data/raw' / fname, BASE / 'data/processed' / fname, ] for p in candidates: if p.exists(): df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] for c in ['close','price','last','mid','open']: if c in df.columns: df['price'] = pd.to_numeric(df[c], errors='coerce') break for c in ['date','datetime','time','timestamp']: if c in df.columns: df['date'] = pd.to_datetime(df[c], errors='coerce') break return df[['date','price']].dropna().sort_values('date') return None def load_trades(path): df = pd.read_csv(path) df.columns = [c.lower().strip() for c in df.columns] for col in ['dollar_pnl_real','pnl_real','pnl','net_pnl']: if col in df.columns: df['pnl'] = pd.to_numeric(df[col], errors='coerce'); break for col in ['entry_time','date','entry_date']: if col in df.columns: df['date'] = pd.to_datetime(df[col], errors='coerce'); break df = df.dropna(subset=['pnl','date']).sort_values('date').reset_index(drop=True) if df['pnl'].abs().median() < 1.0: df['pnl'] *= ACCOUNT return df def daily_sharpe(pnl, dates): dfx = pd.DataFrame({'d': pd.to_datetime(dates).dt.date, 'p': pnl}) daily = dfx.groupby('d')['p'].sum() idx = pd.date_range(daily.index.min(), daily.index.max(), freq='B') daily = daily.reindex(idx, fill_value=0) ret = daily / ACCOUNT return (ret.mean() / ret.std() * np.sqrt(252)) if ret.std() > 0 else 0.0 def signal_return_correlation(signal_series, price_series, forward_days): """ Core test: Pearson correlation between z-score signal magnitude and forward FX return at 'forward_days' horizon. """ sig = signal_series.copy() px = price_series.copy() # Align on date both = pd.concat([sig.rename('z'), px.rename('p')], axis=1).dropna() both['fwd_ret'] = both['p'].pct_change(forward_days).shift(-forward_days) both = both.dropna() if len(both) < 50: return np.nan, np.nan, 0 # Correlation between |z| and direction-adjusted forward return # direction: positive z → expect positive FX return (mean reversion) both['signed_fwd'] = np.sign(both['z']) * both['fwd_ret'] # Only use rows where |z| >= threshold active = both[both['z'].abs() >= 2.0] if len(active) < 30: return np.nan, np.nan, len(active) corr, pval = stats.pearsonr(active['z'].abs(), active['signed_fwd']) return corr, pval, len(active) def build_zscore(spread, window): m = spread.rolling(window, min_periods=window).mean() s = spread.rolling(window, min_periods=window).std().replace(0, np.nan) return ((spread - m) / s).dropna() def signal_sharpe_by_window(spread, price_series, z_window, threshold, tp_pct, sl_pct, hold_days=2): """ Simulate trades triggered by z-score breaches and compute Sharpe. Simplified: long when z >= +threshold, short when z <= -threshold. Exit: after hold_days or when price moves TP/SL. """ z = build_zscore(spread, z_window) both = pd.concat([z.rename('z'), price_series.rename('p')], axis=1).dropna() if len(both) < 100: return np.nan, 0 trades = [] in_trade = False entry_i = 0 direction = 0 for i in range(len(both)): row = both.iloc[i] if not in_trade: if row['z'] >= threshold: direction = 1; in_trade = True; entry_i = i elif row['z'] <= -threshold: direction = -1; in_trade = True; entry_i = i else: elapsed = i - entry_i entry_p = both.iloc[entry_i]['p'] curr_p = row['p'] ret = direction * (curr_p - entry_p) / entry_p if ret >= tp_pct: trades.append(tp_pct * ACCOUNT) in_trade = False elif ret <= -sl_pct: trades.append(-sl_pct * ACCOUNT) in_trade = False elif elapsed >= hold_days: trades.append(ret * ACCOUNT) in_trade = False if len(trades) < 10: return np.nan, len(trades) t = np.array(trades) # Trade-level Sharpe proxy sh = (t.mean() / t.std() * np.sqrt(len(t))) if t.std() > 0 else 0 return sh, len(trades) # ── Load data ────────────────────────────────────────────────────────────────── us2y = load_rate('us2y.csv', 'us2y') de2y = load_rate('de2y.csv', 'de2y') jp2y = load_rate('jp2y.csv', 'jp2y') eu_trades = load_trades(EU_FILE) uj_trades = load_trades(UJ_FILE) eu_price = load_price('eurusd_1h.csv') or load_price('EURUSD_1h.csv') or load_price('eurusd.csv') uj_price = load_price('usdjpy_1h.csv') or load_price('USDJPY_1h.csv') or load_price('usdjpy.csv') print("=" * 75) print(" LAG INSTABILITY TEST v2 — Signal-Return Correlation Analysis") print("=" * 75) # ══════════════════════════════════════════════════════════════════════════════ # EURUSD # ══════════════════════════════════════════════════════════════════════════════ print("\n" + "─"*75) print(" EURUSD — Z-WINDOW SENSITIVITY WITH SIGNAL-RETURN CORRELATION") print("─"*75) if us2y is not None and de2y is not None: eu_rates = us2y.join(de2y, how='inner').dropna() eu_rates['spread'] = eu_rates['us2y'] - eu_rates['de2y'] eu_rates['spread_chg'] = eu_rates['spread'].diff(1) eu_spread = eu_rates['spread'].dropna() # Get daily price proxy from trades (use entry price if available) if eu_price is not None: eu_px = eu_price.set_index(pd.to_datetime(eu_price['date']))['price'] eu_px = eu_px.resample('D').last().dropna() else: # Build proxy from trade P&L (approximate daily level) eu_px = None print(f"\n Rate data: {len(eu_rates)} daily obs | " f"spread range {eu_rates['spread'].min():.2f} to {eu_rates['spread'].max():.2f}%") print(f"\n {'Z-win':>6} {'Sigs/yr':>8} {'ACF 1d':>8} {'ACF 5d':>8} " f"{'Corr(z,fwd1d)':>14} {'Corr(z,fwd5d)':>14} {'Trd Sharpe':>11} {'n':>5}") print(f" {'─'*90}") for z_win in [20, 30, 40, 50, 60, 75, 90, 120]: z = build_zscore(eu_spread, z_win) # ACF (spread persistence) acf1 = float(z.autocorr(lag=1)) acf5 = float(z.autocorr(lag=5)) # Signal frequency new_sig = (z.abs() >= 2.75) & ~(z.shift(1).abs() >= 2.75) n_yr = new_sig.sum() / (len(z) / 252) # Signal-return correlations if eu_px is not None: c1d, p1d, n1d = signal_return_correlation(z, eu_px, 1) c5d, p5d, n5d = signal_return_correlation(z, eu_px, 5) # Trade-level Sharpe simulation sh, n_trades = signal_sharpe_by_window( eu_spread, eu_px, z_win, 2.75, 0.002, 0.0025, hold_days=2) else: c1d = c5d = np.nan n1d = n5d = 0 sh, n_trades = np.nan, 0 marker = ' ◄' if z_win == 60 else '' print(f" {z_win:>6} {n_yr:>8.1f} {acf1:>8.4f} {acf5:>8.4f} " f"{c1d:>14.4f} {c5d:>14.4f} {sh:>11.2f} {n_trades:>5}{marker}") # Spread persistence: how long does a z-breach stay elevated? print(f"\n Spread persistence — how many days does z stay elevated after breach?") print(f" (Critical: if z collapses within 1 day → lag is compressed, edge weakens)") print(f"\n {'Days after breach':>18} {'Still |z|>1.5':>14} {'Still |z|>2.0':>14} {'Mean |z|':>10}") z60 = build_zscore(eu_spread, 60) breach_idx = z60.index[z60.abs() >= 2.75] for lag in [0, 1, 2, 3, 5]: lag_z = z60.shift(-lag) still_1p5 = (lag_z[z60.abs() >= 2.75].abs() >= 1.5).mean() * 100 still_2p0 = (lag_z[z60.abs() >= 2.75].abs() >= 2.0).mean() * 100 mean_z = lag_z[z60.abs() >= 2.75].abs().mean() print(f" Day +{lag:>2}: {still_1p5:>13.1f}% {still_2p0:>13.1f}% {mean_z:>10.4f}") else: print(" ⚠️ Rate CSVs not found — running from trades only") # Signal-return from trade data directly print(f"\n EURUSD trade performance by z-score band:") print(f" (Higher z = stronger conviction = should show higher WR and avg return)") if 'zscore' in eu_trades.columns: eu_z = eu_trades.dropna(subset=['zscore']) eu_z['z_abs'] = eu_z['zscore'].abs() print(f"\n {'Band':>12} {'n':>5} {'WR':>6} {'Avg $':>8} {'Sharpe':>8} {'Expected':>15}") print(f" {'─'*65}") for lo, hi, label in [(2.75,3.0,'2.75-3.00'),(3.0,3.5,'3.00-3.50'), (3.5,4.5,'3.50-4.50'),(4.5,99,'4.50+')]: sub = eu_z[(eu_z['z_abs']>=lo)&(eu_z['z_abs']0).mean() avg = sub['pnl'].mean() sh = daily_sharpe(sub['pnl'],sub['date']) exp = 'Improving ✅' if avg > eu_z[eu_z['z_abs']12} {len(sub):>5} {wr:5.1%} {avg:>+8.0f} {sh:>8.2f} {exp:>15}") print(f"\n A SMOOTH INCREASE in WR/Avg across bands = lag is real and stable ✅") print(f" A FLAT or NOISY pattern = z-score isn't measuring lag strength ⚠️") else: print(" (z-score column not in file — year-by-year shown instead)") # ══════════════════════════════════════════════════════════════════════════════ # USDJPY # ══════════════════════════════════════════════════════════════════════════════ print("\n" + "─"*75) print(" USDJPY — Z-WINDOW SENSITIVITY WITH SIGNAL-RETURN CORRELATION") print("─"*75) if us2y is not None and jp2y is not None: uj_rates = us2y.join(jp2y, how='inner').dropna() uj_rates['spread'] = uj_rates['us2y'] - uj_rates['jp2y'] uj_spread = uj_rates['spread'].dropna() if uj_price is not None: uj_px = uj_price.set_index(pd.to_datetime(uj_price['date']))['price'] uj_px = uj_px.resample('D').last().dropna() else: uj_px = None print(f"\n US-JP spread: {len(uj_rates)} obs | " f"range {uj_rates['spread'].min():.2f} to {uj_rates['spread'].max():.2f}%") print(f"\n {'Z-win':>6} {'Sigs/yr':>8} {'ACF 1d':>8} {'ACF 5d':>8} " f"{'Corr(z,fwd1d)':>14} {'Corr(z,fwd5d)':>14} {'Trd Sharpe':>11} {'n':>5}") print(f" {'─'*90}") for z_win in [10, 15, 20, 25, 30, 40, 50, 60, 90]: z = build_zscore(uj_spread, z_win) acf1 = float(z.autocorr(lag=1)) acf5 = float(z.autocorr(lag=5)) new_s = (z.abs() >= 2.0) & ~(z.shift(1).abs() >= 2.0) n_yr = new_s.sum() / (len(z) / 252) if uj_px is not None: c1d, _, n1 = signal_return_correlation(z, uj_px, 1) c5d, _, n5 = signal_return_correlation(z, uj_px, 5) sh, n_t = signal_sharpe_by_window( uj_spread, uj_px, z_win, 2.0, 0.007, 0.004, hold_days=1) else: c1d = c5d = np.nan; n1 = n5 = 0; sh = np.nan; n_t = 0 marker = ' ◄' if z_win == 30 else '' print(f" {z_win:>6} {n_yr:>8.1f} {acf1:>8.4f} {acf5:>8.4f} " f"{c1d:>14.4f} {c5d:>14.4f} {sh:>11.2f} {n_t:>5}{marker}") # BoJ era analysis — does signal-return correlation change post-YCC? print(f"\n BoJ era comparison — does the lag change post-YCC?") print(f" (If post-YCC correlation collapses → model structure needs updating)") print(f"\n {'Era':>20} {'ACF(1d)':>9} {'ACF(5d)':>9} {'Note':>30}") print(f" {'─'*75}") for era, y0, y1, note in [ ('Pre-YCC 2003-15', 2003, 2015, 'JP2Y free-floating'), ('YCC Era 2016-23', 2016, 2023, 'JP2Y pegged near 0'), ('Post-YCC 2024+', 2024, 2030, 'JP2Y now rising'), ]: sub = uj_spread.loc[f'{y0}':f'{y1}'] if len(sub) < 50: continue z = build_zscore(sub, 30) if len(z) < 30: continue a1 = float(z.autocorr(lag=1)) a5 = float(z.autocorr(lag=5)) print(f" {era:>20} {a1:>9.4f} {a5:>9.4f} {note:>30}") # Random lag distribution print(f"\n RANDOM Z-WINDOW TEST (1,000 simulations, uniform 10d-60d):") np.random.seed(42) acfs = [float(build_zscore(uj_spread, w).autocorr(lag=1)) for w in np.random.randint(10, 60, 1000)] acfs = [a for a in acfs if not np.isnan(a)] print(f" ACF(1d) under random window:") print(f" Min={min(acfs):.4f} p5={np.percentile(acfs,5):.4f} " f"Median={np.median(acfs):.4f} p95={np.percentile(acfs,95):.4f} Max={max(acfs):.4f}") print(f" % sims positive ACF: {(np.array(acfs)>0.05).mean()*100:.1f}%") print(f" → If consistently positive across all windows: lag is real, not a peak ✅") else: print(" ⚠️ Rate CSVs missing") print(f"\n{'='*75}") print(" VERDICT") print(f"{'='*75}") print(""" Smooth degradation across windows = robust timing assumption ✅ Sharp single-window peak = overfit to exact parameter ❌ Correlation(signal, forward_return) interpretation: Positive and stable across horizons = edge is real, lag is genuine ✅ Peaks at exactly 24h then drops = timing overfit ⚠️ Near zero across all horizons = signal has no predictive power ❌ The BoJ era table is the most important for USDJPY: If ACF degrades post-2024 → two-sided spread is weakening the z-score approach Monitor every quarter during paper trading. """)