""" usdjpy_param_sweep_v4.py ========================= Finds the missing 3.40 OOS Sharpe config. LOCKED from v3 (best dimensions): fib = 0.786 direction = both sizing = scaled session = london SWEEPING (parameters locked in v3 that produced the 3.40): threshold: 2.0, 2.25, 2.5, 2.75, 3.0 z_window: 30, 60, 90 tp: 0.30%, 0.40%, 0.50%, 0.60% sl: 0.35%, 0.50%, 0.65% hold: 24h, 36h, 52h, 72h z_exit: 1.0, 1.5, 2.0, off Total: 2,880 configs — ~30 minutes Run from project root: python src/research/usdjpy_param_sweep_v4.py """ import pandas as pd import numpy as np from pathlib import Path import sys import warnings warnings.filterwarnings('ignore') from itertools import product BASE_PATH = Path(__file__).resolve().parents[2] PROC_PATH = BASE_PATH / "data" / "processed" RATES_PATH = BASE_PATH / "data" / "raw" / "rates" if str(BASE_PATH / "src") not in sys.path: sys.path.insert(0, str(BASE_PATH / "src")) NOTIONAL = 300_000 SPREAD_COST = 0.0001 WF_SPLIT = pd.Timestamp('2020-01-01') # ── Locked from v3 ──────────────────────────────────────────────────────────── FIB = 0.786 DIRECTION = 'both' SIZING = 'scaled' SESSION = 'london' # ── Sweep ───────────────────────────────────────────────────────────────────── THRESHOLDS = [2.0, 2.25, 2.5, 2.75, 3.0] Z_WINDOWS = [30, 60, 90] TPS = [0.0030, 0.0040, 0.0050, 0.0060] SLS = [0.0035, 0.0050, 0.0065] HOLDS = [24, 36, 52, 72] Z_EXITS = [1.0, 1.5, 2.0, 99.0] SWEEP = list(product(THRESHOLDS, Z_WINDOWS, TPS, SLS, HOLDS, Z_EXITS)) print(f"Configs: {len(SWEEP):,}") 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') full_idx = pd.date_range(rates.index.min(), rates.index.max(), freq='D') rates = rates.reindex(full_idx).ffill().dropna() rates['spread'] = rates['us2y'] - rates['jp2y'] return rates def build_zscores(rates): zscores = {} for w in Z_WINDOWS: s = rates['spread'] z = (s - s.rolling(w).mean()) / s.rolling(w).std() full_h = pd.date_range(z.index.min(), z.index.max(), freq='h') zscores[w] = z.reindex(full_h).ffill() return zscores 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, threshold, tp, sl, hold_h, z_exit): trades = [] last_exit = None hours = pd.date_range('2003-01-01', '2026-04-01', freq='h') for dt in hours: if dt not in z_hourly.index: continue z = float(z_hourly.at[dt]) if abs(z) < threshold: continue # London session filter h_eet = (dt.hour + 2) % 24 if not (7 <= h_eet <= 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) bar_end = dt + pd.Timedelta(hours=1) - pd.Timedelta(minutes=1) try: bar_slice = m15.loc[dt:bar_end] 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 window = m15.loc[dt + pd.Timedelta(minutes=1): dt + pd.Timedelta(hours=6)] for edt, ebar in window.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_h)] exit_price = None exit_reason= "hold_expiry" for hdt, hbar in hold_bars.iterrows(): if z_exit < 99: hr = hdt.replace(minute=0, second=0, microsecond=0) if hr in z_hourly.index: hz = float(z_hourly.at[hr]) if ((direction == 1 and hz <= -z_exit) or (direction == -1 and hz >= z_exit)): exit_price = float(hbar['close']) exit_reason = "z_exit"; break 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, 'pnl': pnl, 'direction': direction, 'exit_reason': exit_reason}) last_exit = entry_time + pd.Timedelta(hours=hold_h) return pd.DataFrame(trades) def score(df, years): if df is None or len(df) < 10: return {'n': 0, 'wr': 0, 'sharpe': 0, 'dd': 0, 'total': 0} pnl = df['pnl'].values n = len(pnl) wr = (pnl > 0).mean() * 100 eq = 100_000 + np.cumsum(pnl) pk = np.maximum.accumulate(eq) dd = ((eq - pk) / pk * 100).min() pyr = n / years exc = pnl / 100_000 - 0.04 / pyr sh = exc.mean() / exc.std() * np.sqrt(pyr) if exc.std() > 0 else 0 return {'n': n, 'wr': round(wr,1), 'sharpe': round(sh,2), 'dd': round(dd,2), 'total': int(pnl.sum())} def main(): print("="*72) print("USDJPY PARAM SWEEP v4 — Finding the 3.40") print(f"Locked: fib={FIB} direction={DIRECTION} sizing={SIZING} session={SESSION}") print(f"Sweeping: threshold × z_window × tp × sl × hold × z_exit") print("="*72) rates = load_rates() print("Pre-computing z-scores...") zscores = build_zscores(rates) print("Loading prices...") m15 = load_prices() years = (m15.index.max() - m15.index.min()).days / 365.25 yrs_oos = (m15.index.max() - WF_SPLIT).days / 365.25 results = [] total = len(SWEEP) for i, (thresh, z_win, tp, sl, hold, z_exit) in enumerate(SWEEP, 1): if i % 50 == 0: best = max((r['sh_oos'] for r in results), default=0) print(f" [{i:>4}/{total}] best OOS so far: {best:.2f}", end="\r") trades = run_backtest(zscores[z_win], m15, thresh, tp, sl, hold, z_exit) if len(trades) < 10: continue oos = trades[trades['entry_time'] >= WF_SPLIT] rf = score(trades, years) ro = score(oos, yrs_oos) if rf['sharpe'] < 0.5: continue results.append({ 'threshold': thresh, 'z_window': z_win, 'tp': tp, 'sl': sl, 'hold': hold, 'z_exit': z_exit, 'n': rf['n'], 'wr': rf['wr'], 'sh_full': rf['sharpe'], 'sh_oos': ro['sharpe'], 'dd': rf['dd'], 'total': rf['total'], 'n_oos': ro['n'], 'wr_oos': ro['wr'], }) results.sort(key=lambda x: x['sh_oos'], reverse=True) out = PROC_PATH / "usdjpy_param_sweep_v4.csv" pd.DataFrame(results).to_csv(out, index=False) print(f"\n\n{'='*72}") print("TOP 20 — Ranked by OOS Sharpe") print(f"{'='*72}") print(f" {'t':>5}{'zw':>4}{'tp%':>6}{'sl%':>6}{'h':>4}" f"{'ze':>5}{'N':>5}{'WR%':>6}{'Sh':>6}{'Sh_OOS':>8}{'DD%':>7}") print(f" {'─'*65}") for r in results[:20]: print(f" {r['threshold']:>5}{r['z_window']:>4}" f"{r['tp']*100:>6.2f}{r['sl']*100:>6.2f}" f"{r['hold']:>4}{r['z_exit']:>5}" f"{r['n']:>5}{r['wr']:>6.1f}" f"{r['sh_full']:>6.2f}{r['sh_oos']:>8.2f}{r['dd']:>7.2f}") best = results[0] print(f"\n{'='*72}") print("BEST CONFIG — FINAL LOCKED PARAMETERS") print(f"{'='*72}") print(f" z_window: {best['z_window']}") print(f" threshold: ±{best['threshold']}") print(f" tp: {best['tp']*100:.2f}%") print(f" sl: {best['sl']*100:.2f}%") print(f" hold: {best['hold']}h") print(f" z_exit: {best['z_exit']}") print(f" fib: {FIB} (locked)") print(f" session: {SESSION} (locked)") print(f" direction: {DIRECTION} (locked)") print(f" sizing: {SIZING} (locked)") print(f"\n Full: Sharpe {best['sh_full']:.2f} WR {best['wr']:.1f}% " f"DD {best['dd']:.2f}% N={best['n']}") print(f" OOS: Sharpe {best['sh_oos']:.2f} WR {best['wr_oos']:.1f}% " f"N={best['n_oos']}") print(f"\n Results saved: {out.name}") if __name__ == "__main__": main()