""" usdjpy_real_costs_v2.py ======================== Applies realistic live execution costs to the validated USDJPY model. Mirrors the structure of real_world_costs_v1.py used for EURUSD. EURUSD used a simple SPREAD_COST=0.0001 (% of notional). USDJPY was also modelled with SPREAD_COST=0.0001 but this is wrong: EURUSD: 0.0001 / 1.085 × 300k = $27.65/trade (correct order of magnitude) USDJPY: 0.0001 / 150 × 300k = $0.20/trade (severely undercosted) Real USDJPY execution costs: Spread: 2.0 pips avg (IC Markets raw = 1.5-2.5 pip typical) Entry slippage: 0.5 pip (Fibonacci pullback = near-limit entry) TP exit: 0.0 pips (limit order — no adverse slippage) Stop exit: 1.0 pip (market order on stop hit) Hold/z-exit: 0.5 pip (market order at bar close) 1 pip USDJPY = 0.01 JPY At USDJPY 150: 1 pip = 0.01/150 × 300,000 = $20/trade This script: 1. Re-runs the full backtest applying realistic per-trade cost model 2. Compares backtest vs real-costs: equity, Sharpe, WR, DD 3. Checks FTMO compliance survives real costs 4. Generates equity curve chart comparison 5. Produces usdjpy_trades_real_costs.csv LOCKED PARAMETERS: z30 t2.0 TP0.70% SL0.40% H24h fib0.786 london both scaled Run from project root: python src/research/usdjpy_real_costs_v2.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")) # ── Model constants ─────────────────────────────────────────────────────────── Z_WINDOW = 30 THRESHOLD = 2.0 TP = 0.0070 SL = 0.0040 HOLD = 24 FIB = 0.786 NOTIONAL = 300_000 WF_SPLIT = pd.Timestamp('2020-01-01') # ── Backtest spread cost (what the model was optimised with) ────────────────── SPREAD_COST_BT = 0.0001 # same as EURUSD — but wrong for USDJPY # ── Realistic live cost model ───────────────────────────────────────────────── # All costs in pips (0.01 JPY). Convert to % at execution time using live price. # At USDJPY 150: 1 pip = 0.01/150 = 0.0000667 = 0.00667% of price # At USDJPY 110: 1 pip = 0.01/110 = 0.0000909 = 0.00909% of price # We use actual entry price for conversion to be accurate across 23-year history. COST_TABLE = { # (exit_reason): (total_pips_cost) # Spread is always paid. Additional slippage depends on exit type. # # Entry: spread 2.0 pips + entry slip 0.5 pip = 2.5 pips # TP exit: 0 additional (limit order) total = 2.5 pips # Stop exit: stop slip 1.0 pip total = 3.5 pips # Hold/z-exit:market slip 0.5 pip total = 3.0 pips "tp": 2.5, "stop": 3.5, "hold_expiry": 3.0, "z_exit": 3.0, } PIP_SIZE = 0.01 # 1 pip = 0.01 JPY def pip_cost(pips, entry_price, notional): """Convert pip count to USD cost at given USDJPY entry price.""" return pips * PIP_SIZE / entry_price * notional 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): """Full backtest recording both backtest and real-cost P&L per trade.""" trades = [] last_exit = None for dt in pd.date_range('2003-01-01', pd.Timestamp.now().floor('h'), freq='h'): # v2: end date now dynamic — was hard-capped '2026-04-01', which froze the trade list at 2026-03-20 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 # ── Backtest P&L (simple spread cost) ──────────────────────────────── pnl_bt = round(((exit_price - entry_price) / entry_price * direction - SPREAD_COST_BT) * NOTIONAL * mult, 2) # ── Real-cost P&L (realistic pip-based cost model) ──────────────────── cost_pips = COST_TABLE.get(exit_reason, 3.0) cost_usd = pip_cost(cost_pips, entry_price, NOTIONAL * mult) gross_pnl = (exit_price - entry_price) / entry_price * direction \ * NOTIONAL * mult pnl_real = round(gross_pnl - cost_usd, 2) # ── MFE / MAE tracking ──────────────────────────────────────────────── mfe_pct = 0.0 mae_pct = 0.0 for hdt, hbar in hold_bars.iterrows(): if direction == 1: fav = (float(hbar['high']) - entry_price) / entry_price * 100 adv = (entry_price - float(hbar['low'])) / entry_price * 100 else: fav = (entry_price - float(hbar['low'])) / entry_price * 100 adv = (float(hbar['high']) - entry_price) / entry_price * 100 mfe_pct = max(mfe_pct, fav) mae_pct = max(mae_pct, adv) trades.append({ 'entry_time': entry_time, 'exit_reason': exit_reason, 'entry_price': round(entry_price, 4), 'exit_price': round(exit_price, 4), 'direction': direction, 'zscore': round(z, 4), 'zscore_abs': round(z_abs, 4), 'mult': mult, 'cost_pips': cost_pips, 'cost_usd': round(cost_usd, 2), 'pnl_bt': pnl_bt, 'pnl_real': pnl_real, 'mfe_pct': round(mfe_pct, 4), 'mae_pct': round(mae_pct, 4), 'year': entry_time.year, 'month': entry_time.month, }) last_exit = entry_time + pd.Timedelta(hours=HOLD) return pd.DataFrame(trades) def score(pnl, label, years): p = np.array(pnl) n = len(p) if n < 5: return {'label':label,'n':0,'wr':0,'sharpe':0,'dd':0,'total':0} 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 / years exc = p / 100_000 - 0.04 / pyr sh = exc.mean() / exc.std() * np.sqrt(pyr) if exc.std() > 0 else 0 gp = p[p > 0].sum(); gl = abs(p[p < 0].sum()) pf = round(gp / gl, 2) if gl > 0 else 999 return {'label':label,'n':n,'wr':round(wr,1),'sharpe':round(sh,2), 'dd':round(dd,2),'total':int(tot),'pf':pf,'per_yr':round(pyr,1)} def newey_west_t(pnl, years): p = np.array(pnl); n = len(p); mu = p.mean() lags = 5 nw_var = np.var(p, ddof=1) for lag in range(1, lags + 1): w = 1 - lag / (lags + 1) cov = np.cov(p[lag:], p[:-lag])[0, 1] nw_var += 2 * w * cov se = np.sqrt(max(nw_var, 1e-12) / n) t = mu / se if se > 0 else 0 pv = 2 * (1 - stats.t.cdf(abs(t), df=n-1)) return round(t, 2), round(pv, 6) def main(): print("="*72) print("USDJPY REAL COSTS v1") print("="*72) print(f"\nCost model:") print(f" Backtest (old): SPREAD_COST=0.0001 → ${0.0001/150*300000:.2f}/trade") print(f" Real costs:") print(f" Spread: 2.0 pips") print(f" Entry slippage: 0.5 pip (Fibonacci near-limit entry)") print(f" TP exit: 0.0 pips (limit order)") print(f" Stop exit: 1.0 pip (market order)") print(f" Hold/z exit: 0.5 pip (market order)") print(f" TP total: 2.5 pips = ${2.5*0.01/150*300000:.2f} at USDJPY 150") print(f" Stop total: 3.5 pips = ${3.5*0.01/150*300000:.2f} at USDJPY 150") print("\nLoading data...") z_hourly = load_rates() m15 = load_prices() print("Running backtest with real cost model...") df = run_backtest(z_hourly, m15) print(f" {len(df):,} trades generated") years = (df['entry_time'].max() - df['entry_time'].min()).days / 365.25 yrs_oos = (df['entry_time'].max() - WF_SPLIT).days / 365.25 # ── Core comparison ─────────────────────────────────────────────────────── pnl_bt = df['pnl_bt'].values pnl_real = df['pnl_real'].values r_bt = score(pnl_bt, 'Backtest (SPREAD_COST=0.0001)', years) r_real = score(pnl_real, 'Real costs (pip model)', years) oos_bt = df[df['entry_time'] >= WF_SPLIT]['pnl_bt'].values oos_real = df[df['entry_time'] >= WF_SPLIT]['pnl_real'].values r_oos_bt = score(oos_bt, 'OOS Backtest', yrs_oos) r_oos_real = score(oos_real, 'OOS Real costs', yrs_oos) print(f"\n{'='*72}") print("BACKTEST vs REAL COSTS COMPARISON") print(f"{'='*72}") print(f"\n {'Metric':<28}{'Backtest':>12}{'Real Costs':>12}{'Δ':>8}") print(f" {'─'*60}") metrics = [ ('Full Sharpe', r_bt['sharpe'], r_real['sharpe']), ('OOS Sharpe', r_oos_bt['sharpe'], r_oos_real['sharpe']), ('Win Rate %', r_bt['wr'], r_real['wr']), ('Max DD %', r_bt['dd'], r_real['dd']), ('PF', r_bt['pf'], r_real['pf']), ('Total $', r_bt['total']/1000, r_real['total']/1000), ('Trades/yr', r_bt['per_yr'], r_real['per_yr']), ] for label, vbt, vrc in metrics: unit = 'k' if label == 'Total $' else '' delta = vrc - vbt print(f" {label:<28}{vbt:>12.2f}{vrc:>12.2f}{delta:>+8.2f}") # ── Cost analysis ───────────────────────────────────────────────────────── print(f"\n{'='*72}") print("COST ANALYSIS") print(f"{'='*72}") total_cost_usd = df['cost_usd'].sum() avg_cost = df['cost_usd'].mean() cost_pct_tp = df['cost_usd'].mean() / (TP * NOTIONAL) * 100 print(f"\n Total cost over 23 years: ${total_cost_usd:,.0f}") print(f" Average cost per trade: ${avg_cost:.2f}") print(f" Cost as % of avg TP: {cost_pct_tp:.1f}%") print(f" Cost drag on total P&L: " f"{(r_bt['total']-r_real['total'])/r_bt['total']*100:.1f}%") print(f"\n Cost by exit type:") print(f" {'Exit':>14}{'N':>6}{'Pips':>7}{'$/trade':>10}{'WR%':>8}") print(f" {'─'*48}") for reason, pips in COST_TABLE.items(): sub = df[df['exit_reason'] == reason] if len(sub) == 0: continue wr = (sub['pnl_real'] > 0).mean() * 100 avg = sub['cost_usd'].mean() print(f" {reason:>14}{len(sub):>6}{pips:>7.1f}{avg:>10.2f}{wr:>8.1f}%") print(f"\n Cost vs TP=0.70% comparison:") avg_tp_win = df[df['pnl_real'] > 0]['pnl_real'].mean() avg_loss = abs(df[df['pnl_real'] < 0]['pnl_real'].mean()) print(f" Avg winner P&L (real): ${avg_tp_win:.2f}") print(f" Avg loser P&L (real): -${avg_loss:.2f}") print(f" Avg cost/trade: ${avg_cost:.2f} " f"({avg_cost/(TP*NOTIONAL)*100:.1f}% of gross TP)") # ── EURUSD comparison ───────────────────────────────────────────────────── print(f"\n{'='*72}") print("COMPARISON WITH EURUSD REAL COSTS") print(f"{'='*72}") print(f"\n {'Metric':<30}{'EURUSD':>12}{'USDJPY':>12}") print(f" {'─'*54}") comparisons = [ ('Full Sharpe (backtest)', '3.72', f'{r_bt["sharpe"]:.2f}'), ('Full Sharpe (real costs)','~3.72', f'{r_real["sharpe"]:.2f}'), ('OOS Sharpe (backtest)', '2.73', f'{r_oos_bt["sharpe"]:.2f}'), ('OOS Sharpe (real costs)', '~2.73', f'{r_oos_real["sharpe"]:.2f}'), ('Win Rate (real costs)', '~75%', f'{r_real["wr"]:.1f}%'), ('Cost/trade avg', '~$27', f'${avg_cost:.2f}'), ('Cost pips', '~1.7', f'{df["cost_pips"].mean():.1f}'), ('Cost drag', '~15%', f'{(r_bt["total"]-r_real["total"])/r_bt["total"]*100:.1f}%'), ('Trades/year', '182', f'{r_bt["per_yr"]:.0f}'), ('TP', '0.20%', '0.70%'), ('SL', '0.25%', '0.40%'), ] for label, eu, uj in comparisons: print(f" {label:<30}{eu:>12}{uj:>12}") # ── Yearly P&L comparison ───────────────────────────────────────────────── print(f"\n{'='*72}") print("YEARLY P&L — BACKTEST vs REAL COSTS") print(f"{'='*72}") print(f"\n {'Year':>6}{'N':>5}{'BT P&L':>12}{'Real P&L':>12}{'Cost $':>10}{'Real WR%':>10}") print(f" {'─'*55}") for yr in sorted(df['year'].unique()): sub = df[df['year'] == yr] bt_p = sub['pnl_bt'].sum() re_p = sub['pnl_real'].sum() cost = sub['cost_usd'].sum() wr = (sub['pnl_real'] > 0).mean() * 100 print(f" {yr:>6}{len(sub):>5}{bt_p:>12,.0f}{re_p:>12,.0f}" f"{cost:>10,.0f}{wr:>10.1f}%") # ── Newey-West significance ──────────────────────────────────────────────── t_bt, pv_bt = newey_west_t(pnl_bt, years) t_real,pv_real= newey_west_t(pnl_real, years) print(f"\n Newey-West t-stat (backtest): {t_bt} p={pv_bt:.2e}") print(f" Newey-West t-stat (real costs): {t_real} p={pv_real:.2e}") # ── Verdict ─────────────────────────────────────────────────────────────── print(f"\n{'='*72}") print("VERDICT") print(f"{'='*72}") sharpe_degradation = (r_bt['sharpe'] - r_real['sharpe']) / r_bt['sharpe'] * 100 oos_degradation = (r_oos_bt['sharpe'] - r_oos_real['sharpe']) / r_oos_bt['sharpe'] * 100 print(f"\n Sharpe degradation (full): {sharpe_degradation:.1f}%") print(f" Sharpe degradation (OOS): {oos_degradation:.1f}%") print(f" Real-cost OOS Sharpe: {r_oos_real['sharpe']:.2f}") if r_oos_real['sharpe'] > 2.0 and sharpe_degradation < 30: verdict = "EDGE SURVIVES REAL COSTS ✅" detail = (f"OOS Sharpe {r_oos_real['sharpe']:.2f} with realistic costs. " f"Cost drag {sharpe_degradation:.0f}% is acceptable.") elif r_oos_real['sharpe'] > 1.5: verdict = "EDGE SURVIVES — MARGINAL COST IMPACT ✅" detail = "Still investment grade. Monitor execution quality in live trading." else: verdict = "COSTS MATERIAL — REVIEW BROKER SELECTION ⚠️" detail = "Consider tighter-spread broker or session filter to reduce cost." print(f"\n {verdict}") print(f" {detail}") # ── Charts ──────────────────────────────────────────────────────────────── df_sorted = df.sort_values('entry_time').reset_index(drop=True) eq_bt = 100_000 + df_sorted['pnl_bt'].cumsum() eq_real = 100_000 + df_sorted['pnl_real'].cumsum() dates = pd.to_datetime(df_sorted['entry_time']) fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 10)) ax1.plot(dates, eq_bt, color='#4488ff', linewidth=1.2, label=f'Backtest | Sharpe {r_bt["sharpe"]:.2f}', alpha=0.8) ax1.plot(dates, eq_real, color='#00cc66', linewidth=1.5, label=f'Real Costs | Sharpe {r_real["sharpe"]:.2f}') ax1.axvline(WF_SPLIT, color='gold', linewidth=1.2, linestyle='--', label='IS/OOS split (2020)') ax1.axhline(100_000, color='#888', linewidth=0.5) ax1.set_title(f'USDJPY Equity Curve — Backtest vs Real Costs\n' f'Real: Sharpe {r_real["sharpe"]:.2f} OOS {r_oos_real["sharpe"]:.2f} ' f'WR {r_real["wr"]:.1f}% DD {r_real["dd"]:.2f}%') ax1.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, _: f'${x:,.0f}')) ax1.legend(); ax1.grid(alpha=0.2) # Cost breakdown by year yearly = df.groupby('year').agg( cost_usd=('cost_usd','sum'), pnl_bt=('pnl_bt','sum'), pnl_real=('pnl_real','sum') ) x = yearly.index w = 0.35 ax2.bar(x - w/2, yearly['pnl_bt']/1000, w, color='#4488ff', alpha=0.7, label='Backtest P&L ($k)') ax2.bar(x + w/2, yearly['pnl_real']/1000, w, color='#00cc66', alpha=0.7, label='Real Costs P&L ($k)') ax2.axhline(0, color='#888', linewidth=0.5) ax2.set_title('Yearly P&L — Backtest vs Real Costs ($k)') ax2.set_ylabel('$k'); ax2.legend(); ax2.grid(alpha=0.2, axis='y') fig.tight_layout() out_chart = OUT_PATH / 'usdjpy_real_costs.png' fig.savefig(out_chart, dpi=150, bbox_inches='tight') plt.close(fig) print(f"\n Chart saved: {out_chart.name}") # ── Save trades ─────────────────────────────────────────────────────────── out_csv = PROC_PATH / "usdjpy_trades_real_costs.csv" df.to_csv(out_csv, index=False) print(f" Trades saved: {out_csv.name}") print(f" Columns: {list(df.columns)}") if __name__ == "__main__": main()