""" usdjpy_full_validation_v1.py ============================= Full validation suite for the locked USDJPY model. Runs every check we ran on EURUSD: 1. Equity curve vs buy & hold 2. Yearly P&L 3. Monthly heatmap 4. Drawdown 5. Walk-forward IS/OOS split 6. Yearly Sharpe (all positive?) 7. Bootstrap Sharpe CI 8. Permutation test (p-value) 9. Win rate by z-score band 10. Direction analysis (LONG vs SHORT) 11. BoJ regime analysis (pre-YCC / YCC / post-YCC) 12. Exit breakdown 13. Newey-West t-stat 14. Monthly return distribution LOCKED PARAMETERS: z_window=30, threshold=±2.0, tp=0.70%, sl=0.40% hold=24h, fib=0.786, session=london, direction=both sizing=scaled (1x/1.5x/2x) Run from project root: python src/research/usdjpy_full_validation_v1.py """ import pandas as pd import numpy as np import matplotlib.pyplot as plt from pathlib import Path from scipy import stats import sys import warnings warnings.filterwarnings('ignore') BASE_PATH = Path(__file__).resolve().parents[2] PROC_PATH = BASE_PATH / "data" / "processed" RATES_PATH = BASE_PATH / "data" / "raw" / "rates" OUT_PATH = BASE_PATH / "data" / "output" / "usdjpy" OUT_PATH.mkdir(parents=True, exist_ok=True) if str(BASE_PATH / "src") not in sys.path: sys.path.insert(0, str(BASE_PATH / "src")) # ── Locked parameters ───────────────────────────────────────────────────────── Z_WINDOW = 30 THRESHOLD = 2.0 TP = 0.0070 SL = 0.0040 HOLD = 24 FIB = 0.786 NOTIONAL = 300_000 SPREAD_COST = 0.0001 WF_SPLIT = pd.Timestamp('2020-01-01') YCC_START = pd.Timestamp('2016-01-01') YCC_END = pd.Timestamp('2024-01-01') def load_rates(): us2y = pd.read_csv(RATES_PATH / "us2y.csv") dc = [c for c in us2y.columns if 'date' in c.lower()][0] vc = [c for c in us2y.columns if c != dc][0] us2y[dc] = pd.to_datetime(us2y[dc]) us2y = us2y.rename(columns={dc:'date', vc:'us2y'})[['date','us2y']].dropna() jp2y = pd.read_csv(RATES_PATH / "jp2y.csv") jp2y['date'] = pd.to_datetime(jp2y['date']) jp2y = jp2y[['date','jp2y']].dropna() rates = pd.merge(us2y, jp2y, on='date', how='outer').sort_values('date') rates = rates.set_index('date') rates = rates.reindex( pd.date_range(rates.index.min(), rates.index.max(), freq='D') ).ffill().dropna() rates['spread'] = rates['us2y'] - rates['jp2y'] s = rates['spread'] z = (s - s.rolling(Z_WINDOW).mean()) / s.rolling(Z_WINDOW).std() return z.reindex( pd.date_range(z.index.min(), z.index.max(), freq='h') ).ffill() def load_prices(): fpath = next(BASE_PATH.rglob("USDJPY_15M*.csv"), None) df = pd.read_csv(fpath) df.columns = df.columns.str.strip().str.lower() dc = next(c for c in df.columns if 'date' in c or 'time' in c) df[dc] = pd.to_datetime(df[dc]) return df.rename(columns={dc:'datetime'}).set_index('datetime').sort_index() def run_backtest(z_hourly, m15): trades = [] last_exit = None for dt in pd.date_range('2003-01-01', '2026-04-01', freq='h'): if dt not in z_hourly.index: continue z = float(z_hourly.at[dt]) if abs(z) < THRESHOLD: continue if not (7 <= (dt.hour + 2) % 24 <= 17): continue if last_exit is not None and dt <= last_exit: continue direction = 1 if z >= THRESHOLD else -1 z_abs = abs(z) mult = 1.0 if z_abs < 2.5 else (1.5 if z_abs < 3.5 else 2.0) try: bar_slice = m15.loc[dt:dt + pd.Timedelta(hours=1) - pd.Timedelta(minutes=1)] except Exception: continue if bar_slice.empty: continue bh = float(bar_slice['high'].max()) bl = float(bar_slice['low'].min()) bc = float(bar_slice['close'].iloc[-1]) if direction == 1: pull = bc - bl if pull <= 0.05: continue target = bc - FIB * pull else: pull = bh - bc if pull <= 0.05: continue target = bc + FIB * pull entry_price = entry_time = None for edt, ebar in m15.loc[ dt + pd.Timedelta(minutes=1): dt + pd.Timedelta(hours=6)].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: continue tp_px = entry_price * (1 + TP) if direction == 1 else entry_price * (1 - TP) sl_px = entry_price * (1 - SL) if direction == 1 else entry_price * (1 + SL) hold_bars = m15.loc[ entry_time + pd.Timedelta(minutes=1): entry_time + pd.Timedelta(hours=HOLD)] exit_price = None exit_reason = "hold_expiry" for hdt, hbar in hold_bars.iterrows(): if direction == 1: if float(hbar['high']) >= tp_px: exit_price = tp_px; exit_reason = "tp"; break if float(hbar['low']) <= sl_px: exit_price = sl_px; exit_reason = "stop"; break else: if float(hbar['low']) <= tp_px: exit_price = tp_px; exit_reason = "tp"; break if float(hbar['high']) >= sl_px: exit_price = sl_px; exit_reason = "stop"; break if exit_price is None: exit_price = float(hold_bars['close'].iloc[-1]) \ if not hold_bars.empty else entry_price pnl = round(((exit_price - entry_price) / entry_price * direction - SPREAD_COST) * NOTIONAL * mult, 2) trades.append({ 'entry_time': entry_time, 'direction': direction, 'zscore': round(z, 4), 'zscore_abs': round(z_abs, 4), 'exit_reason': exit_reason, 'pnl': pnl, 'year': entry_time.year, 'month': entry_time.month, 'regime': ('ycc' if YCC_START <= entry_time < YCC_END else ('post_ycc' if entry_time >= YCC_END else 'pre_ycc')), }) last_exit = entry_time + pd.Timedelta(hours=HOLD) return pd.DataFrame(trades) def sharpe(pnl_arr, years): pnl = np.array(pnl_arr) if len(pnl) < 5 or pnl.std() == 0: return 0 pyr = len(pnl) / years exc = pnl / 100_000 - 0.04 / pyr return exc.mean() / exc.std() * np.sqrt(pyr) def run_validation(df): years = (df['entry_time'].max() - df['entry_time'].min()).days / 365.25 is_df = df[df['entry_time'] < WF_SPLIT] oos_df = df[df['entry_time'] >= WF_SPLIT] yrs_is = (WF_SPLIT - df['entry_time'].min()).days / 365.25 yrs_oos = (df['entry_time'].max() - WF_SPLIT).days / 365.25 pnl = df['pnl'].values print(f"\n{'='*72}") print("USDJPY FULL VALIDATION — LOCKED PARAMETERS") print(f"z{Z_WINDOW} t{THRESHOLD} TP{TP*100:.2f}% SL{SL*100:.2f}% " f"H{HOLD}h fib{FIB} london both scaled") print(f"{'='*72}") # ── Core metrics ────────────────────────────────────────────────────────── def metrics(d, yrs, label): p = d['pnl'].values n = len(p) wr = (p > 0).mean() * 100 tot = p.sum() eq = 100_000 + np.cumsum(p) pk = np.maximum.accumulate(eq) dd = ((eq - pk) / pk * 100).min() pyr = n / yrs sh = sharpe(p, yrs) gp = p[p > 0].sum(); gl = abs(p[p < 0].sum()) pf = gp / gl if gl > 0 else 999 return n, round(wr,1), round(sh,2), round(dd,2), int(tot), round(pf,2), round(pyr,1) nf,wf,sf,df2,tf,pff,pyrf = metrics(df, years, "Full") ni,wi,si,di,ti,pfi,pyri = metrics(is_df, yrs_is, "IS") no,wo,so,do,to,pfo,pyro = metrics(oos_df,yrs_oos, "OOS") print(f"\n {'Period':<22}{'N':>6}{'N/yr':>6}{'WR%':>6}{'Sharpe':>8}" f"{'DD%':>8}{'PF':>6}{'Total$':>12}") print(f" {'─'*70}") print(f" {'Full (2003-2026)':<22}{nf:>6}{pyrf:>6.0f}{wf:>6.1f}" f"{sf:>8.2f}{df2:>8.2f}{pff:>6.2f}{tf:>12,.0f}") print(f" {'IS (2003-2020)':<22}{ni:>6}{pyri:>6.0f}{wi:>6.1f}" f"{si:>8.2f}{di:>8.2f}{pfi:>6.2f}{ti:>12,.0f}") print(f" {'OOS (2020-2026)':<22}{no:>6}{pyro:>6.0f}{wo:>6.1f}" f"{so:>8.2f}{do:>8.2f}{pfo:>6.2f}{to:>12,.0f}") # ── T1: Yearly Sharpe — all positive? ───────────────────────────────────── print(f"\n{'='*72}") print("T1: YEARLY PERFORMANCE") print(f"{'='*72}") yearly = df.groupby('year').apply( lambda x: pd.Series({ 'n': len(x), 'pnl': x['pnl'].sum(), 'wr': (x['pnl'] > 0).mean() * 100, }) ).reset_index() pos_years = (yearly['pnl'] > 0).sum() total_years = len(yearly) print(f"\n {'Year':>6}{'N':>5}{'WR%':>7}{'P&L$':>12}{'':>5}") print(f" {'─'*35}") for _, row in yearly.iterrows(): marker = "✓" if row['pnl'] > 0 else "✗" print(f" {int(row['year']):>6}{int(row['n']):>5}" f"{row['wr']:>7.1f}%{int(row['pnl']):>12,} {marker}") print(f"\n Positive years: {pos_years}/{total_years} " f"({pos_years/total_years*100:.0f}%)") if pos_years == total_years: print(f" ✅ ALL years positive") else: neg = yearly[yearly['pnl'] <= 0]['year'].tolist() print(f" ⚠️ Negative years: {neg}") # ── T2: Bootstrap Sharpe CI ─────────────────────────────────────────────── print(f"\n{'='*72}") print("T2: BOOTSTRAP SHARPE CONFIDENCE INTERVAL (1000 resamples)") print(f"{'='*72}") np.random.seed(42) boot_sharpes = [] for _ in range(1000): sample = np.random.choice(pnl, size=len(pnl), replace=True) boot_sharpes.append(sharpe(sample, years)) ci_lo = np.percentile(boot_sharpes, 2.5) ci_hi = np.percentile(boot_sharpes, 97.5) print(f"\n Observed Sharpe: {sf:.3f}") print(f" 95% CI: [{ci_lo:.3f}, {ci_hi:.3f}]") if ci_lo > 0: print(f" ✅ Entire CI positive — edge confirmed") else: print(f" ⚠️ CI crosses zero") # ── T3: Permutation test ────────────────────────────────────────────────── print(f"\n{'='*72}") print("T3: PERMUTATION TEST (1000 shuffles)") print(f"{'='*72}") perm_sharpes = [] for _ in range(1000): shuffled = np.random.permutation(pnl) perm_sharpes.append(sharpe(shuffled, years)) p_val = (np.array(perm_sharpes) >= sf).mean() z_perm = (sf - np.mean(perm_sharpes)) / np.std(perm_sharpes) print(f"\n Observed Sharpe: {sf:.3f}") print(f" Permutation mean: {np.mean(perm_sharpes):.4f}") print(f" Z-score: {z_perm:.1f}σ") print(f" p-value: {p_val:.6f}") if p_val < 0.05: print(f" ✅ Significant (p < 0.05) — not random") else: print(f" ⚠️ Not significant") # ── T4: Walk-forward OOS ───────────────────────────────────────────────── print(f"\n{'='*72}") print("T4: WALK-FORWARD VALIDATION") print(f"{'='*72}") degradation = (sf - so) / sf * 100 print(f"\n IS Sharpe: {si:.3f}") print(f" OOS Sharpe: {so:.3f}") print(f" Degradation: {degradation:+.1f}%") if so > si: print(f" ✅ OOS BEATS IS — model improving out of sample") elif degradation < 35: print(f" ✅ Acceptable degradation (< 35%)") else: print(f" ⚠️ High degradation — possible overfitting") # ── T5: Newey-West t-stat ───────────────────────────────────────────────── print(f"\n{'='*72}") print("T5: NEWEY-WEST T-STATISTIC") print(f"{'='*72}") returns = pnl / 100_000 n = len(returns) mean_r = returns.mean() # Newey-West with 5 lags lags = 5 nw_var = np.var(returns, ddof=1) for lag in range(1, lags + 1): w = 1 - lag / (lags + 1) cov = np.cov(returns[lag:], returns[:-lag])[0, 1] nw_var += 2 * w * cov nw_se = np.sqrt(max(nw_var, 1e-12) / n) t_stat = mean_r / nw_se p_nw = 2 * (1 - stats.t.cdf(abs(t_stat), df=n-1)) print(f"\n Mean return/trade: {mean_r*100:.4f}%") print(f" Newey-West SE: {nw_se*100:.4f}%") print(f" t-statistic: {t_stat:.2f}") print(f" p-value: {p_nw:.2e}") if p_nw < 0.05: print(f" ✅ Significant — edge is real") else: print(f" ⚠️ Not significant") # ── Direction analysis ──────────────────────────────────────────────────── print(f"\n{'='*72}") print("DIRECTION ANALYSIS") print(f"{'='*72}") for d, dlabel in [(1, 'LONG (Short JPY)'), (-1, 'SHORT (Long JPY)')]: sub = df[df['direction'] == d] wr = (sub['pnl'] > 0).mean() * 100 tot = sub['pnl'].sum() print(f" {dlabel:<22}: n={len(sub):>4} WR={wr:.1f}% " f"Total=${tot:>10,.0f}") # ── Z-score band breakdown ──────────────────────────────────────────────── print(f"\n{'='*72}") print("Z-SCORE BAND BREAKDOWN") print(f"{'='*72}") bands = [(2.0, 2.5, '2.0-2.5 (1x)'), (2.5, 3.5, '2.5-3.5 (1.5x)'), (3.5, 9.9, '3.5+ (2x)')] for lo, hi, label in bands: sub = df[(df['zscore_abs'] >= lo) & (df['zscore_abs'] < hi)] if len(sub) == 0: continue wr = (sub['pnl'] > 0).mean() * 100 avg = sub['pnl'].mean() print(f" z {label:<18}: n={len(sub):>4} WR={wr:.1f}% Avg=${avg:.0f}") # ── BoJ regime analysis ─────────────────────────────────────────────────── print(f"\n{'='*72}") print("BoJ REGIME ANALYSIS") print(f"{'='*72}") for reg, rlabel in [ ('pre_ycc', 'Pre-YCC (<2016) '), ('ycc', 'YCC Era (2016-2024)'), ('post_ycc', 'Post-YCC (2024+) '), ]: sub = df[df['regime'] == reg] if len(sub) == 0: continue wr = (sub['pnl'] > 0).mean() * 100 tot = sub['pnl'].sum() sh_r = sharpe(sub['pnl'].values, (sub['entry_time'].max()-sub['entry_time'].min()).days/365.25) print(f" {rlabel}: n={len(sub):>4} WR={wr:.1f}% " f"Sharpe={sh_r:.2f} Total=${tot:>9,.0f}") # ── Exit breakdown ──────────────────────────────────────────────────────── print(f"\n{'='*72}") print("EXIT BREAKDOWN") print(f"{'='*72}") for reason, n_r in df['exit_reason'].value_counts().items(): pct = n_r / len(df) * 100 sub_wr = (df[df['exit_reason'] == reason]['pnl'] > 0).mean() * 100 print(f" {reason:<15}: {n_r:>4} ({pct:.1f}%) WR={sub_wr:.1f}%") # ── Comparison with EURUSD ──────────────────────────────────────────────── print(f"\n{'='*72}") print("COMPARISON WITH EURUSD MODEL") print(f"{'='*72}") print(f"\n {'Metric':<30}{'EURUSD':>12}{'USDJPY':>12}") print(f" {'─'*54}") comparisons = [ ('Full Sharpe', '3.72', f'{sf:.2f}'), ('OOS Sharpe', '2.73', f'{so:.2f}'), ('Win Rate', '75.2%', f'{wf:.1f}%'), ('Max Drawdown', '-4.28%', f'{df2:.2f}%'), ('Trades/year', '182', f'{pyrf:.0f}'), ('Threshold', '±2.75', f'±{THRESHOLD}'), ('TP', '0.20%', f'{TP*100:.2f}%'), ('SL', '0.25%', f'{SL*100:.2f}%'), ('Hold', '52h', f'{HOLD}h'), ('Fib entry', '0.786', f'{FIB}'), ('Session', 'London', 'London'), ('Z-exit', '±1.5', 'Not needed'), ] for label, eu, uj in comparisons: print(f" {label:<30}{eu:>12}{uj:>12}") # ── Verdict ─────────────────────────────────────────────────────────────── print(f"\n{'='*72}") print("VERDICT") print(f"{'='*72}") tests_passed = sum([ pos_years == total_years, ci_lo > 0, p_val < 0.05, so > 0, p_nw < 0.05, ]) print(f"\n Tests passed: {tests_passed}/5") print(f"\n T1 Yearly: {'✅ ALL positive' if pos_years==total_years else f'⚠️ {total_years-pos_years} negative years'}") print(f" T2 Bootstrap: {'✅ CI entirely positive' if ci_lo>0 else '⚠️ CI crosses zero'}") print(f" T3 Permutation: {'✅ p < 0.05' if p_val<0.05 else '⚠️ p >= 0.05'}") print(f" T4 Walk-fwd: {'✅ OOS beats IS' if so>si else ('✅ Acceptable degradation' if degradation<35 else '⚠️ High degradation')}") print(f" T5 Newey-West: {'✅ Significant' if p_nw<0.05 else '⚠️ Not significant'}") if tests_passed >= 4: print(f"\n RESULT: MODEL VALIDATED ✅") print(f" USDJPY model passes the same validation standard as EURUSD.") print(f" Ready to paper trade alongside EURUSD model.") else: print(f"\n RESULT: INSUFFICIENT EVIDENCE ⚠️") return df def make_charts(df): print(f"\n{'='*72}") print("GENERATING CHARTS") print(f"{'='*72}") df = df.sort_values('entry_time').reset_index(drop=True) df['equity'] = 100_000 + df['pnl'].cumsum() df['peak'] = df['equity'].cummax() df['drawdown'] = (df['equity'] - df['peak']) / df['peak'] * 100 dates = pd.to_datetime(df['entry_time']) years = (df['entry_time'].max() - df['entry_time'].min()).days / 365.25 yrs_oos = (df['entry_time'].max() - WF_SPLIT).days / 365.25 pnl = df['pnl'].values sf = sharpe(pnl, years) oos_pnl = df[df['entry_time'] >= WF_SPLIT]['pnl'].values so = sharpe(oos_pnl, yrs_oos) wr = (pnl > 0).mean() * 100 dd = df['drawdown'].min() # 1. Equity curve fig, ax = plt.subplots(figsize=(14, 5)) ax.plot(dates, df['equity'], color='#00cc66', linewidth=1.5, label='USDJPY Strategy') ax.axvline(WF_SPLIT, color='gold', linewidth=1.2, linestyle='--', label='IS/OOS split (2020)') ax.axhline(100_000, color='#888', linewidth=0.5) ax.fill_between(dates, 100_000, df['equity'], where=df['equity'] >= 100_000, alpha=0.08, color='#00cc66') ax.fill_between(dates, 100_000, df['equity'], where=df['equity'] < 100_000, alpha=0.08, color='red') ax.set_title( f"USDJPY Equity Curve | Sharpe {sf:.2f} | OOS Sharpe {so:.2f} | " f"WR {wr:.1f}% | DD {dd:.2f}%") ax.yaxis.set_major_formatter( plt.FuncFormatter(lambda x, _: f'${x:,.0f}')) ax.legend(); ax.grid(alpha=0.2) fig.tight_layout() fig.savefig(OUT_PATH / 'usdjpy_equity.png', dpi=150) plt.close(fig) print(" Saved: usdjpy_equity.png") # 2. Drawdown fig, ax = plt.subplots(figsize=(14, 4)) ax.fill_between(dates, df['drawdown'], 0, color='#ff4444', alpha=0.6) ax.axhline(0, color='#888', linewidth=0.5) ax.set_title(f"Drawdown | Max: {dd:.2f}%") ax.set_ylabel('%'); ax.grid(alpha=0.2) fig.tight_layout() fig.savefig(OUT_PATH / 'usdjpy_drawdown.png', dpi=150) plt.close(fig) print(" Saved: usdjpy_drawdown.png") # 3. Yearly P&L fig, ax = plt.subplots(figsize=(14, 5)) yearly = df.groupby('year')['pnl'].sum() colors = ['#00cc66' if v >= 0 else '#ff4444' for v in yearly.values] bars = ax.bar(yearly.index, yearly.values / 1000, color=colors, alpha=0.8) for bar, val in zip(bars, yearly.values): ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + (0.3 if val >= 0 else -1.5), f'${val/1000:.0f}k', ha='center', fontsize=7) ax.axhline(0, color='#888', linewidth=0.5) ax.set_title('Yearly P&L ($k)') ax.set_ylabel('$k'); ax.grid(alpha=0.2, axis='y') fig.tight_layout() fig.savefig(OUT_PATH / 'usdjpy_yearly.png', dpi=150) plt.close(fig) print(" Saved: usdjpy_yearly.png") # 4. Monthly heatmap hmap = df.pivot_table( values='pnl', index='year', columns='month', aggfunc='sum') / 1000 month_names = ['Jan','Feb','Mar','Apr','May','Jun', 'Jul','Aug','Sep','Oct','Nov','Dec'] hmap.columns = [month_names[m-1] for m in hmap.columns] fig, ax = plt.subplots(figsize=(16, max(6, len(hmap) * 0.45 + 2))) vmax = max(abs(hmap.values[~np.isnan(hmap.values)]).max(), 1) im = ax.imshow(hmap.values, cmap='RdYlGn', aspect='auto', vmin=-vmax, vmax=vmax) ax.set_xticks(range(len(hmap.columns))) ax.set_xticklabels(hmap.columns, fontsize=9) ax.set_yticks(range(len(hmap.index))) ax.set_yticklabels(hmap.index.astype(int), fontsize=9) for i in range(len(hmap.index)): for j in range(len(hmap.columns)): val = hmap.values[i, j] if not np.isnan(val): ax.text(j, i, f'{val:.1f}', ha='center', va='center', fontsize=7, color='black' if abs(val) < vmax * 0.6 else 'white') plt.colorbar(im, ax=ax, label='P&L $k') ax.set_title('Monthly P&L Heatmap ($k)') fig.tight_layout() fig.savefig(OUT_PATH / 'usdjpy_heatmap.png', dpi=150) plt.close(fig) print(" Saved: usdjpy_heatmap.png") # 5. Monthly distribution fig, ax = plt.subplots(figsize=(10, 5)) monthly = df.groupby(['year', 'month'])['pnl'].sum().values / 1000 ax.hist(monthly, bins=25, color='#4488ff', alpha=0.7, edgecolor='white', linewidth=0.3) ax.axvline(monthly.mean(), color='gold', linewidth=1.5, linestyle='--', label=f'Mean ${monthly.mean():.1f}k') ax.axvline(0, color='#888', linewidth=1) ax.set_title('Monthly P&L Distribution ($k)') ax.set_xlabel('P&L $k'); ax.legend(); ax.grid(alpha=0.2) fig.tight_layout() fig.savefig(OUT_PATH / 'usdjpy_monthly_dist.png', dpi=150) plt.close(fig) print(" Saved: usdjpy_monthly_dist.png") print(f"\n All charts saved to: {OUT_PATH}") def main(): print("="*72) print("USDJPY FULL VALIDATION v1") print("="*72) print("\nLoading data...") z_hourly = load_rates() m15 = load_prices() print("Running backtest...") df = run_backtest(z_hourly, m15) print(f" {len(df):,} trades generated") df = run_validation(df) make_charts(df) out = PROC_PATH / "usdjpy_validated_trades.csv" df.to_csv(out, index=False) print(f"\n Trades saved: {out.name}") if __name__ == "__main__": main()