""" usdjpy_signal_build_v1.py ========================== Builds the US-JP 2Y yield spread lag signal for USDJPY, analogous to spot_lag_v3.py for EURUSD. Then runs a full parameter sweep across: - Z-score threshold: 2.0, 2.5, 2.75, 3.0, 3.5 - Z-score window: 30, 60, 90, 120 bars - TP: 0.20%, 0.30%, 0.40%, 0.50% - SL: 0.25%, 0.35%, 0.50% - Hold: 24h, 36h, 52h, 72h Outputs the top 20 configs ranked by Sharpe with walk-forward validation on OOS period (2020+). Run from project root: python src/research/usdjpy_signal_build_v1.py Expected runtime: 30-60 minutes """ import pandas as pd import numpy as np from pathlib import Path 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" SRC_PATH = BASE_PATH / "src" if str(SRC_PATH) not in sys.path: sys.path.insert(0, str(SRC_PATH)) SYMBOL = "USDJPY" SPREAD_COST = 0.0001 # ~1 pip spread cost (USDJPY ~0.01 pip = 0.0001) FIB = 0.786 NOTIONAL = 300_000 # USD BASE_RISK = 0.0075 WF_SPLIT = pd.Timestamp('2020-01-01') # ── Parameter sweep grid ────────────────────────────────────────────────────── THRESHOLDS = [2.0, 2.5, 2.75, 3.0, 3.5] Z_WINDOWS = [30, 60, 90, 120] TPS = [0.0020, 0.0030, 0.0040, 0.0050] SLS = [0.0025, 0.0035, 0.0050] HOLDS = [24, 36, 52, 72] def load_data(): """Load and merge US2Y, JP2Y and build spread z-score.""" print("Loading rate data...") # US 2Y from FRED us2y_path = RATES_PATH / "us2y.csv" us2y = pd.read_csv(us2y_path) date_col = [c for c in us2y.columns if 'date' in c.lower()][0] val_col = [c for c in us2y.columns if c != date_col][0] us2y[date_col] = pd.to_datetime(us2y[date_col]) us2y = us2y.rename(columns={date_col:'date', val_col:'us2y'}) us2y = us2y[['date','us2y']].dropna().sort_values('date') print(f" US2Y: {len(us2y):,} rows {us2y['date'].min().date()} → {us2y['date'].max().date()}") # JP 2Y from MoF jp2y_path = RATES_PATH / "jp2y.csv" jp2y = pd.read_csv(jp2y_path) jp2y['date'] = pd.to_datetime(jp2y['date']) jp2y = jp2y[['date','jp2y']].dropna().sort_values('date') print(f" JP2Y: {len(jp2y):,} rows {jp2y['date'].min().date()} → {jp2y['date'].max().date()}") # Merge on date, forward-fill weekends/holidays 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'] print(f" Spread: mean={rates['spread'].mean():.3f}% std={rates['spread'].std():.3f}%") print(f" Range: {rates['spread'].min():.3f}% → {rates['spread'].max():.3f}%") return rates def build_zscore(rates, z_window): """Compute rolling z-score of the yield spread.""" s = rates['spread'] z = (s - s.rolling(z_window).mean()) / s.rolling(z_window).std() return z.rename('zscore') def load_price_data(): """Load USDJPY 15M bars from Dukascopy.""" print("\nLoading USDJPY price data...") try: from ingestion.price_loader_15m import load_15m df = load_15m(symbol='USDJPY').sort_values('datetime') df['datetime'] = pd.to_datetime(df['datetime']) print(f" USDJPY 15M: {len(df):,} bars") return df.set_index('datetime').sort_index() except Exception as e: print(f" ERROR: {e}") print(" Trying alternative loader...") try: # Try loading directly from CSV fpath = BASE_PATH / "data" / "raw" / "prices" / "USDJPY_15M_2003_2026.csv" if not fpath.exists(): fpath = next(BASE_PATH.rglob("USDJPY_15M*.csv"), None) if fpath is None: print(" USDJPY 15M data not found") print(" Need to download USDJPY 15M data from Dukascopy first") return None df = pd.read_csv(fpath) df.columns = df.columns.str.strip().str.lower() date_col = next(c for c in df.columns if 'date' in c or 'time' in c) df[date_col] = pd.to_datetime(df[date_col]) df = df.rename(columns={date_col:'datetime'}) df = df.set_index('datetime').sort_index() print(f" USDJPY 15M: {len(df):,} bars (from CSV)") return df except Exception as e2: print(f" Also failed: {e2}") return None def run_backtest(rates, zscore, m15_idx, threshold, tp_pct, sl_pct, hold_hours, session_start=7, session_end=17): """ Full backtest — same structure as EURUSD truth_baseline_backtest. Entry: Fibonacci 0.786 pullback within 6h of z-score crossing threshold. Exit: TP / SL / z-exit at ±1.5 / hold expiry. """ ZSCORE_EXIT = 1.5 # Build hourly z-score lookup from daily rates z_daily = zscore.copy() full_idx_h = pd.date_range(z_daily.index.min(), z_daily.index.max(), freq='h') z_hourly = z_daily.reindex(full_idx_h).ffill() trades = [] last_exit = None dates = pd.date_range('2003-01-01', '2026-04-01', freq='h') for dt in dates: if dt not in z_hourly.index: continue z = float(z_hourly[dt]) if abs(z) < threshold: continue # Session filter (EET hours) h = (dt.hour + 2) % 24 # UTC → EET approx if not (session_start <= h <= session_end): continue if last_exit is not None and dt <= last_exit: continue direction = 1 if z >= threshold else -1 # Get bar data for Fibonacci entry bar_end = dt + pd.Timedelta(hours=1) - pd.Timedelta(minutes=1) bar_slice = m15_idx.loc[dt:bar_end] if dt in m15_idx.index else pd.DataFrame() 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.0005: continue target = bc - FIB * pull else: pull = bh - bc if pull <= 0.0005: continue target = bc + FIB * pull # 6-hour entry window entry_window = m15_idx.loc[ dt + pd.Timedelta(minutes=1):dt + pd.Timedelta(hours=6)] entry_price = entry_time = None for edt, ebar in entry_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 = entry_price * (1+tp_pct) if direction==1 else entry_price*(1-tp_pct) sl = entry_price * (1-sl_pct) if direction==1 else entry_price*(1+sl_pct) hold = m15_idx.loc[ entry_time + pd.Timedelta(minutes=1): entry_time + pd.Timedelta(hours=hold_hours)] exit_price = None exit_reason = "hold_expiry" for hdt, hbar in hold.iterrows(): hr = hdt.replace(minute=0, second=0, microsecond=0) if hr in z_hourly.index: hz = float(z_hourly[hr]) if ((direction==1 and hz<=-ZSCORE_EXIT) or (direction==-1 and hz>=ZSCORE_EXIT)): exit_price = float(hbar['close']) exit_reason = "z_exit"; break if direction == 1: if float(hbar['high']) >= tp: exit_price = tp; exit_reason = "tp"; break if float(hbar['low']) <= sl: exit_price = sl; exit_reason = "stop"; break else: if float(hbar['low']) <= tp: exit_price = tp; exit_reason = "tp"; break if float(hbar['high']) >= sl: exit_price = sl; exit_reason = "stop"; break if exit_price is None: exit_price = float(hold['close'].iloc[-1]) if not hold.empty else entry_price exit_reason = "hold_expiry" pnl = round(((exit_price - entry_price) / entry_price * direction - SPREAD_COST) * NOTIONAL, 2) trades.append({ 'entry_time': entry_time, 'direction': direction, 'zscore': round(z, 4), 'exit_reason': exit_reason, 'pnl': pnl, }) last_exit = entry_time + pd.Timedelta(hours=hold_hours) return pd.DataFrame(trades) def score(df, label, years): if df is None or len(df) == 0: return {"label":label,"n":0,"wr":0,"sharpe":0,"pf":0,"dd":0,"total":0} pnl = df['pnl'].values n = len(pnl) wr = (pnl>0).mean()*100 total = pnl.sum() eq = 100_000 + np.cumsum(pnl) peak = np.maximum.accumulate(eq) max_dd = ((eq-peak)/peak*100).min() per_yr = n/years exc = pnl/100_000 - 0.04/per_yr sharpe = exc.mean()/exc.std()*np.sqrt(per_yr) if exc.std()>0 else 0 gp = pnl[pnl>0].sum(); gl = abs(pnl[pnl<0].sum()) pf = round(gp/gl,2) if gl>0 else 999 return {"label":label,"n":n,"wr":round(wr,1),"sharpe":round(sharpe,2), "pf":pf,"dd":round(max_dd,2),"total":int(total), "per_yr":round(per_yr,1)} def main(): print("="*72) print("USDJPY MACRO SIGNAL — Parameter Sweep v1") print("US-JP 2Y yield spread z-score lead-lag model") print("="*72) # ── Load data ───────────────────────────────────────────────────────────── rates = load_data() m15 = load_price_data() if m15 is None: print("\nCannot proceed without USDJPY 15M data.") print("Download USDJPY 15M bars from Dukascopy and save as:") print(" data/raw/prices/USDJPY_15M_2003_2026.csv") sys.exit(1) years = (m15.index.max() - m15.index.min()).days / 365.25 yrs_oos = (m15.index.max() - WF_SPLIT).days / 365.25 print(f"\nData coverage: {years:.1f} years") print(f"OOS (2020+): {yrs_oos:.1f} years") # ── Parameter sweep ─────────────────────────────────────────────────────── print(f"\n{'='*72}") print("PARAMETER SWEEP") total_configs = len(THRESHOLDS)*len(Z_WINDOWS)*len(TPS)*len(SLS)*len(HOLDS) print(f"Testing {total_configs:,} configurations...") print(f"{'='*72}") results = [] done = 0 for z_win in Z_WINDOWS: zscore = build_zscore(rates, z_win) for thresh in THRESHOLDS: for tp in TPS: for sl in SLS: for hold in HOLDS: done += 1 if done % 20 == 0: print(f" {done:>4}/{total_configs} " f"z{z_win} t{thresh} tp{tp*100:.2f} " f"sl{sl*100:.2f} h{hold}h", end="\r") trades = run_backtest( rates, zscore, m15, thresh, tp, sl, hold) if len(trades) < 20: continue is_trades = trades[trades['entry_time'] < WF_SPLIT] oos_trades = trades[trades['entry_time'] >= WF_SPLIT] r_full = score(trades, "full", years) r_oos = score(oos_trades, "oos", yrs_oos) if r_full['sharpe'] < 0.5: continue label = (f"z{z_win} t{thresh} tp{tp*100:.2f}% " f"sl{sl*100:.2f}% h{hold}h") results.append({ 'label': label, 'z_win': z_win, 'threshold': thresh, 'tp': tp, 'sl': sl, 'hold': hold, 'n_full': r_full['n'], 'wr_full': r_full['wr'], 'sh_full': r_full['sharpe'], 'dd_full': r_full['dd'], 'tot_full': r_full['total'], 'n_oos': r_oos['n'], 'sh_oos': r_oos['sharpe'], 'wr_oos': r_oos['wr'], }) print(f"\n\nSweep complete. {len(results)} configs with Sharpe > 0.5") if not results: print("\nNo configs passed the minimum threshold.") print("USDJPY may require different signal construction.") sys.exit(0) # Sort by OOS Sharpe results.sort(key=lambda x: x['sh_oos'], reverse=True) print(f"\n{'='*72}") print("TOP 20 CONFIGS — Ranked by OOS Sharpe (2020+)") print(f"{'='*72}") print(f" {'Config':<40}{'N':>5}{'WR%':>6}{'Sh(full)':>9}" f"{'Sh(OOS)':>9}{'DD%':>7}{'Total$':>10}") print(f" {'─'*75}") for r in results[:20]: print(f" {r['label']:<40}{r['n_full']:>5}{r['wr_full']:>6.1f}" f"{r['sh_full']:>9.2f}{r['sh_oos']:>9.2f}" f"{r['dd_full']:>7.2f}{r['tot_full']:>10,.0f}") # Save results out = pd.DataFrame(results) out_path = PROC_PATH / "usdjpy_sweep_results.csv" out.to_csv(out_path, index=False) print(f"\nFull results saved: {out_path.name}") # Best config best = results[0] print(f"\n{'='*72}") print("BEST CONFIG (highest OOS Sharpe)") print(f"{'='*72}") print(f" {best['label']}") print(f" Full period: Sharpe {best['sh_full']:.2f} WR {best['wr_full']:.1f}% " f"DD {best['dd_full']:.2f}%") print(f" OOS 2020+: Sharpe {best['sh_oos']:.2f} WR {best['wr_oos']:.1f}% " f"N={best['n_oos']}") if best['sh_oos'] >= 1.0: print(f"\n VERDICT: PROMISING — OOS Sharpe {best['sh_oos']:.2f} ≥ 1.0") print(f" Next step: full validation on best config (walk-forward, MC)") elif best['sh_oos'] >= 0.5: print(f"\n VERDICT: MARGINAL — OOS Sharpe {best['sh_oos']:.2f}") print(f" Some signal present but below investment grade") else: print(f"\n VERDICT: NO EDGE — OOS Sharpe below 0.5") print(f" USDJPY does not respond to this signal in the same way as EURUSD") if __name__ == "__main__": main()