""" usdjpy_tp_sl_sweep_v5.py ========================= Targeted TP/SL sweep based on MFE/MAE findings. MFE/MAE told us: - TP 0.60% is well calibrated — test small range around it - SL 0.65% is too wide — winners need only 0.30% - Test SL range 0.20%-0.45% tightly All other params locked from v4: z30 t2.0 hold24h fib0.786 london both scaled Run from project root: python src/research/usdjpy_tp_sl_sweep_v5.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 ──────────────────────────────────────────────────────────────────── Z_WINDOW = 30 THRESHOLD = 2.0 HOLD = 24 FIB = 0.786 # ── Sweep — tight range based on MFE/MAE ───────────────────────────────────── # TP: confirmed ~0.60%, test small range TPS = [0.0050, 0.0055, 0.0060, 0.0065, 0.0070] # SL: MAE says 0.30% covers 90% of winners — test 0.20%-0.45% SLS = [0.0020, 0.0025, 0.0030, 0.0035, 0.0040, 0.0045, 0.0050] SWEEP = list(product(TPS, SLS)) print(f"Configs: {len(SWEEP)}") def load_data(): 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') z_h = z.reindex(full_h).ffill() fpath = next(BASE_PATH.rglob("USDJPY_15M*.csv"), None) df = pd.read_csv(fpath) df.columns = df.columns.str.strip().str.lower() dc2 = next(c for c in df.columns if 'date' in c or 'time' in c) df[dc2] = pd.to_datetime(df[dc2]) m15 = df.rename(columns={dc2:'datetime'}).set_index('datetime').sort_index() return z_h, m15 def run_backtest(z_hourly, m15, tp, sl): 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, 'pnl': pnl}) 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 wr = (pnl>0).mean()*100 eq = 100_000+np.cumsum(pnl) pk = np.maximum.accumulate(eq) dd = ((eq-pk)/pk*100).min() pyr = len(pnl)/years exc = pnl/100_000 - 0.04/pyr sh = exc.mean()/exc.std()*np.sqrt(pyr) if exc.std()>0 else 0 return {'n':len(pnl),'wr':round(wr,1),'sharpe':round(sh,2), 'dd':round(dd,2),'total':int(pnl.sum())} def main(): print("="*72) print("USDJPY TP/SL SWEEP v5 — MFE/MAE Guided") print(f"Locked: z{Z_WINDOW} t{THRESHOLD} h{HOLD}h fib{FIB} london both scaled") print(f"TP range: {[f'{t*100:.2f}%' for t in TPS]}") print(f"SL range: {[f'{s*100:.2f}%' for s in SLS]}") print("="*72) z_hourly, m15 = load_data() years = (m15.index.max() - m15.index.min()).days / 365.25 yrs_oos = (m15.index.max() - WF_SPLIT).days / 365.25 results = [] for i, (tp, sl) in enumerate(SWEEP, 1): print(f" [{i:>2}/{len(SWEEP)}] TP={tp*100:.2f}% SL={sl*100:.2f}%", end=" ... ", flush=True) trades = run_backtest(z_hourly, m15, tp, sl) 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}% " f"dd={rf['dd']:>6.2f}%") results.append({'tp':tp,'sl':sl,'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) pd.DataFrame(results).to_csv(PROC_PATH/"usdjpy_tp_sl_sweep_v5.csv", index=False) print(f"\n{'='*72}") print("RESULTS — Ranked by OOS Sharpe") print(f"{'='*72}") print(f" {'TP%':>6}{'SL%':>6}{'N':>5}{'WR%':>6}" f"{'Sh_full':>9}{'Sh_OOS':>8}{'DD%':>7}") print(f" {'─'*50}") for r in results: marker = " ← BEST" if r == results[0] else "" print(f" {r['tp']*100:>6.2f}{r['sl']*100:>6.2f}{r['n']:>5}" f"{r['wr']:>6.1f}{r['sh_full']:>9.2f}" f"{r['sh_oos']:>8.2f}{r['dd']:>7.2f}{marker}") best = results[0] print(f"\n{'='*72}") print("FINAL LOCKED PARAMETERS — USDJPY MODEL") print(f"{'='*72}") print(f" z_window: {Z_WINDOW}") print(f" threshold: ±{THRESHOLD}") print(f" tp: {best['tp']*100:.2f}%") print(f" sl: {best['sl']*100:.2f}%") print(f" hold: {HOLD}h") print(f" fib: {FIB}") print(f" session: london") print(f" direction: both") print(f" sizing: scaled (1x/1.5x/2x)") 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']}") v4_oos = 3.40 if best['sh_oos'] > v4_oos: improvement = best['sh_oos'] - v4_oos print(f"\n MFE/MAE improvement: +{improvement:.2f} Sharpe over v4 baseline") else: print(f"\n v4 TP/SL already optimal — no improvement from tightening SL") if __name__ == "__main__": main()