""" usdjpy_focused_sweep_v3.py =========================== Focused sweep — only tests the dimensions NOT covered in v1/v2. LOCKED from v1 (already validated): z_window = 30 threshold = 2.0 tp = 0.50% sl = 0.50% hold = 52h TESTING (genuinely new dimensions): z_exit: 1.0, 1.5, 2.0, OFF (4) fib: 0.618, 0.786, market (3) session: london, tokyo, all (3) direction: both, short_only, long_only(3) sizing: flat, scaled (2) Total: 216 configs — runs in ~10 minutes Run from project root: python src/research/usdjpy_focused_sweep_v3.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 v1 ──────────────────────────────────────────────────────────── Z_WINDOW = 30 THRESHOLD = 2.0 TP = 0.0050 SL = 0.0050 HOLD = 52 # ── New dimensions to test ──────────────────────────────────────────────────── Z_EXITS = [1.0, 1.5, 2.0, 99.0] FIBS = [0.618, 0.786, 0.0] SESSIONS = ['london', 'tokyo', 'all'] DIRECTIONS = ['both', 'short_only', 'long_only'] SIZING = ['flat', 'scaled'] SWEEP = list(product(Z_EXITS, FIBS, SESSIONS, DIRECTIONS, SIZING)) print(f"Configs to test: {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'] s = rates['spread'] z = (s - s.rolling(Z_WINDOW).mean()) / s.rolling(Z_WINDOW).std() full_h = pd.date_range(z.index.min(), z.index.max(), freq='h') return z.reindex(full_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 in_session(hour_utc, session): if session == 'london': return 7 <= (hour_utc + 2) % 24 <= 17 elif session == 'tokyo': return 0 <= hour_utc <= 9 return True def run_backtest(z_hourly, m15, z_exit, fib, session, direction_filter, sizing): 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 if not in_session(dt.hour, session): continue if last_exit is not None and dt <= last_exit: continue direction = 1 if z >= THRESHOLD else -1 if direction_filter == 'short_only' and direction != -1: continue if direction_filter == 'long_only' and direction != 1: continue mult = 1.0 if sizing == 'scaled': 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 fib == 0.0: entry_price = bc entry_time = bar_slice.index[-1] else: 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)] 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) 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 FOCUSED SWEEP v3") print(f"Locked: z{Z_WINDOW} t{THRESHOLD} tp{TP*100:.2f}% sl{SL*100:.2f}% h{HOLD}h") print(f"Testing: z_exit × fib × session × direction × sizing") print("="*72) print("Loading data...") z_hourly = load_rates() m15 = load_prices() years = (m15.index.max() - m15.index.min()).days / 365.25 yrs_oos = (m15.index.max() - WF_SPLIT).days / 365.25 print(f"Z-score range: {z_hourly.min():.2f} → {z_hourly.max():.2f}") print(f"\nRunning {len(SWEEP)} configs...") results = [] for i, (z_exit, fib, session, direction, sizing) in enumerate(SWEEP, 1): print(f" [{i:>3}/{len(SWEEP)}] " f"ze{z_exit} fib{fib:.3f} {session:>6} {direction:>12} {sizing}", end=" ... ", flush=True) trades = run_backtest(z_hourly, m15, z_exit, fib, session, direction, sizing) oos = trades[trades['entry_time'] >= WF_SPLIT] if len(trades) else pd.DataFrame() rf = score(trades, years) ro = score(oos, yrs_oos) print(f"n={rf['n']:>4} sh={rf['sharpe']:>5.2f} " f"sh_oos={ro['sharpe']:>5.2f} wr={rf['wr']:>5.1f}%") fib_str = f"{fib:.3f}" if fib > 0 else "mkt" results.append({ 'z_exit': z_exit, 'fib': fib_str, 'session': session, 'direction': direction, 'sizing': sizing, '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) # Save out = PROC_PATH / "usdjpy_focused_sweep_v3.csv" pd.DataFrame(results).to_csv(out, index=False) print(f"\n{'='*72}") print("TOP 15 — Ranked by OOS Sharpe") print(f"{'='*72}") print(f" {'ze':>5}{'fib':>7}{'session':>8}{'direction':>13}" f"{'sizing':>7}{'N':>5}{'WR%':>6}{'Sh':>6}{'Sh_OOS':>8}{'DD%':>7}") print(f" {'─'*73}") for r in results[:15]: print(f" {str(r['z_exit']):>5}{r['fib']:>7}{r['session']:>8}" f"{r['direction']:>13}{r['sizing']:>7}{r['n']:>5}" f"{r['wr']:>6.1f}{r['sh_full']:>6.2f}" f"{r['sh_oos']:>8.2f}{r['dd']:>7.2f}") print(f"\n{'='*72}") print("PARAMETER INSIGHTS — what wins in the top 10") print(f"{'='*72}") top10 = pd.DataFrame(results[:10]) for col in ['z_exit','fib','session','direction','sizing']: best = top10[col].value_counts().index[0] n = top10[col].value_counts().iloc[0] print(f" {col:<12}: {best} ({n}/10 top configs)") # V1 baseline reminder print(f"\n V1 BASELINE: ze1.5 fib0.786 london both flat") print(f" Full Sharpe 1.93 OOS Sharpe 2.38 WR 73.6% DD -2.90%") print(f"\n Results saved: {out.name}") if __name__ == "__main__": main()