""" usdjpy_risk_sensitivity_v1.py ============================== Tests USDJPY risk % sensitivity against the combined EU+UJ portfolio. EURUSD risk is LOCKED at 0.75% — only USDJPY risk varies. Research question: "What is the optimal USDJPY risk % that maximises returns while keeping combined FTMO breach rate at 0%?" Tests USDJPY risk at: 0.25%, 0.50%, 0.75%, 1.00%, 1.25%, 1.50%, 2.00% Metrics per level: - Combined daily DD breach rate (must stay 0%) - Combined overall DD breach rate (must stay 0%) - Median FTMO 1-Step challenge time - Combined Sharpe ratio - Combined P&L - Max single-day loss Run from project root: python src/research/usdjpy_risk_sensitivity_v1.py Output: data/output/usdjpy_risk_sensitivity/ sensitivity_results.csv sensitivity_charts.png summary_report.txt """ import sys import json import numpy as np import pandas as pd import matplotlib.pyplot as plt import matplotlib.gridspec as gridspec from pathlib import Path from datetime import datetime BASE_PATH = Path(__file__).resolve().parents[2] sys.path.insert(0, str(BASE_PATH / "src")) OUTPUT_DIR = BASE_PATH / "data" / "output" / "usdjpy_risk_sensitivity" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # ═══════════════════════════════════════════════════════════════════════════════ # LOCKED PARAMETERS — DO NOT CHANGE # ═══════════════════════════════════════════════════════════════════════════════ EU_RISK_PCT = 0.0075 # LOCKED — validated, never touch EU_TP_PCT = 0.0020 EU_SL_PCT = 0.0025 EU_NOTIONAL = 300_000 EU_BANDS = [(2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0)] UJ_TP_PCT = 0.0070 UJ_SL_PCT = 0.0040 UJ_NOTIONAL = 300_000 UJ_BANDS = [(2.0, 2.5, 1.0), (2.5, 3.5, 1.5), (3.5, 99.0, 2.0)] # FTMO limits — tested against both challenge types ACCOUNT_SIZE = 100_000 OVERALL_LIMIT = 10_000 # 10% — same for both 1-Step and 2-Step TARGET_PROFIT = 10_000 # 10% to pass (Phase 1 for 2-Step, full for 1-Step) # Daily DD limits differ between challenge types — tested separately DAILY_LIMIT_1STEP = 3_000 # 3% — 1-Step (stricter) DAILY_LIMIT_2STEP = 5_000 # 5% — 2-Step (more room) DAILY_LIMIT = DAILY_LIMIT_1STEP # default for MC (1-Step = harder constraint) N_SIMS = 5_000 UJ_RISK_LEVELS = [0.0025, 0.0050, 0.0075, 0.0100, 0.0125, 0.0150, 0.0200] RANDOM_SEED = 42 # ═══════════════════════════════════════════════════════════════════════════════ # LOAD TRADE DATA # ═══════════════════════════════════════════════════════════════════════════════ def load_eu_trades(): """Load EURUSD validated trades.""" # Try multiple possible paths for fname in [ "eurusd_validated_trades.csv", "eurusd_trades_real_costs.csv", "validated_trades.csv", ]: path = BASE_PATH / "data" / "processed" / fname if path.exists(): df = pd.read_csv(path) print(f" EU trades: {path.name} ({len(df)} rows)") return df print(" ⚠️ EU trade file not found — using synthetic data") return None def load_uj_trades(): """Load USDJPY validated trades.""" for fname in [ "usdjpy_validated_trades.csv", "usdjpy_trades_real_costs.csv", ]: path = BASE_PATH / "data" / "processed" / fname if path.exists(): df = pd.read_csv(path) print(f" UJ trades: {path.name} ({len(df)} rows)") return df print(" ⚠️ UJ trade file not found — using synthetic data") return None def build_daily_pnl_series(df, pair, risk_pct): """ Convert trade-level returns to daily P&L series. Applies risk % and z-score sizing bands. Returns Series indexed by date. """ if df is None: return None df = df.copy() df.columns = [c.lower() for c in df.columns] # Find return column ret_col = next((c for c in df.columns if c in ['return_real','return','ret','pnl_pct']), None) if ret_col is None: print(f" ⚠️ No return column found for {pair}") return None # Find date column date_col = next((c for c in df.columns if 'exit' in c or 'close' in c or 'date' in c), None) if date_col is None: date_col = df.columns[0] # Find z-score column z_col = next((c for c in df.columns if 'zscore' in c or 'z_score' in c or 'z_abs' in c), None) bands = EU_BANDS if pair == 'EURUSD' else UJ_BANDS notional = EU_NOTIONAL if pair == 'EURUSD' else UJ_NOTIONAL def get_multiplier(z): for lo, hi, mult in bands: if lo <= abs(z) < hi: return mult return bands[-1][2] df['date'] = pd.to_datetime(df[date_col]).dt.normalize() df['multiplier'] = df[z_col].apply(get_multiplier) if z_col else 1.0 df['trade_pnl'] = ( df[ret_col].astype(float) * notional * df['multiplier'] * (risk_pct / 0.0075) # scale from base 0.75% to target risk % ) daily = df.groupby('date')['trade_pnl'].sum() return daily # ═══════════════════════════════════════════════════════════════════════════════ # MONTE CARLO ENGINE # ═══════════════════════════════════════════════════════════════════════════════ def run_mc_simulation(eu_daily, uj_daily, uj_risk_pct, n_sims=N_SIMS, seed=RANDOM_SEED, sim_years=3): """ Run Monte Carlo simulation for combined EU+UJ portfolio. Method: - Bootstrap sample daily P&L from historical data - For each 3-year simulation, track daily and overall DD - Check FTMO breach conditions - Measure challenge completion time Returns dict of metrics. """ rng = np.random.default_rng(seed) # Build combined daily P&L from aligned history if eu_daily is not None and uj_daily is not None: # Align on common dates (outer join, fill missing with 0) combined = pd.DataFrame({'eu': eu_daily, 'uj': uj_daily}).fillna(0) combined['total'] = combined['eu'] + combined['uj'] daily_returns = combined['total'].values eu_only_returns = combined['eu'].values uj_only_returns = combined['uj'].values elif eu_daily is not None: daily_returns = eu_daily.values eu_only_returns = daily_returns uj_only_returns = np.zeros(len(daily_returns)) else: print(" ⚠️ No trade data — cannot run MC") return None n_days_hist = len(daily_returns) n_days_sim = int(sim_years * 252) # trading days # Metrics accumulators daily_breaches = 0 overall_breaches = 0 challenge_days = [] final_balances = [] max_dds = [] for sim in range(n_sims): # Bootstrap: resample daily P&L with replacement idx = rng.integers(0, n_days_hist, size=n_days_sim) sim_pnl = daily_returns[idx] balance = ACCOUNT_SIZE peak = ACCOUNT_SIZE max_dd = 0.0 daily_breach_hit = False overall_breach_hit = False challenge_done = False challenge_day = None # Simulate one day at a time day_start_balance = balance days_in_month = 0 for d in range(n_days_sim): # Reset daily tracking at start of each day if d % 1 == 0: # daily day_start = balance day_pnl = sim_pnl[d] balance += day_pnl peak = max(peak, balance) # Daily DD check — tested against 1-Step limit (stricter) # If it passes 1-Step it automatically passes 2-Step daily limit day_dd = day_start - balance if day_dd >= DAILY_LIMIT_1STEP: daily_breach_hit = True # Overall DD check overall_dd = peak - balance max_dd = max(max_dd, overall_dd) if overall_dd >= OVERALL_LIMIT: overall_breach_hit = True # Challenge completion if not challenge_done and (balance - ACCOUNT_SIZE) >= TARGET_PROFIT: challenge_done = True challenge_day = d + 1 if daily_breach_hit: daily_breaches += 1 if overall_breach_hit: overall_breaches += 1 if challenge_day: challenge_days.append(challenge_day) max_dds.append(max_dd) final_balances.append(balance) n = n_sims challenge_days_arr = np.array(challenge_days) if challenge_days else np.array([n_days_sim]) # Sharpe from historical combined returns mean_daily = np.mean(daily_returns) std_daily = np.std(daily_returns, ddof=1) sharpe = (mean_daily / std_daily * np.sqrt(252)) if std_daily > 0 else 0 return { 'uj_risk_pct': uj_risk_pct, 'daily_breach_rate': round(daily_breaches / n * 100, 2), 'overall_breach_rate': round(overall_breaches / n * 100, 2), 'pass_rate': round((1 - overall_breaches / n) * 100, 2), 'median_challenge_days':int(np.median(challenge_days_arr)), 'median_challenge_months': round(np.median(challenge_days_arr) / 21, 1), 'p25_months': round(np.percentile(challenge_days_arr, 25) / 21, 1), 'p75_months': round(np.percentile(challenge_days_arr, 75) / 21, 1), 'pct_challenges_done': round(len(challenge_days) / n * 100, 1), 'mean_max_dd': round(np.mean(max_dds), 0), 'p95_max_dd': round(np.percentile(max_dds, 95), 0), 'combined_sharpe': round(sharpe, 3), 'annual_pnl_mean': round(mean_daily * 252, 0), 'n_sims': n, } # ═══════════════════════════════════════════════════════════════════════════════ # SYNTHETIC DATA FALLBACK # ═══════════════════════════════════════════════════════════════════════════════ def build_synthetic_daily(pair, risk_pct): """ Build synthetic daily P&L from known backtest statistics. Used when actual trade CSV is not available. """ rng = np.random.default_rng(99) if pair == 'EURUSD': # 182 trades/yr, 75.2% WR, avg win $537 avg loss $723 at 1× trades_per_year = 182 win_rate = 0.752 avg_win = 600 * (risk_pct / 0.0075) avg_loss = -750 * (risk_pct / 0.0075) years = 23 else: # 31 trades/yr, 72.1% WR, avg win $1312 avg loss $750 at 1× trades_per_year = 31 win_rate = 0.721 avg_win = 1312 * (risk_pct / 0.0075) avg_loss = -750 * (risk_pct / 0.0075) years = 23 n_trading_days = int(years * 252) n_trades = int(trades_per_year * years) # Generate trade outcomes outcomes = rng.choice([1, 0], size=n_trades, p=[win_rate, 1 - win_rate]) pnl_per_trade = np.where(outcomes == 1, rng.normal(avg_win, abs(avg_win) * 0.3, n_trades), rng.normal(avg_loss, abs(avg_loss) * 0.2, n_trades)) # Spread trades randomly across trading days trade_days = np.sort(rng.choice(n_trading_days, size=n_trades, replace=True)) daily_pnl = np.zeros(n_trading_days) for i, d in enumerate(trade_days): daily_pnl[d] += pnl_per_trade[i] dates = pd.date_range('2003-01-01', periods=n_trading_days, freq='B') return pd.Series(daily_pnl, index=dates) # ═══════════════════════════════════════════════════════════════════════════════ # MAIN # ═══════════════════════════════════════════════════════════════════════════════ def main(): print("=" * 65) print("USDJPY RISK SENSITIVITY — COMBINED PORTFOLIO ANALYSIS") print("Daily DD tested vs 1-Step ($3k) — if 0% breach, 2-Step ($5k) is safe") print(f"EURUSD risk: LOCKED at {EU_RISK_PCT:.2%}") print(f"USDJPY risk: testing {[f'{r:.2%}' for r in UJ_RISK_LEVELS]}") print(f"Simulations: {N_SIMS:,} × 3-year Monte Carlo") print("=" * 65) # Load trade data print("\nLoading trade data...") eu_raw = load_eu_trades() uj_raw = load_uj_trades() results = [] for uj_risk in UJ_RISK_LEVELS: print(f"\n Testing UJ risk = {uj_risk:.2%}...") # Build daily P&L for EURUSD (locked at 0.75%) if eu_raw is not None: eu_daily = build_daily_pnl_series(eu_raw, 'EURUSD', EU_RISK_PCT) else: eu_daily = build_synthetic_daily('EURUSD', EU_RISK_PCT) # Build daily P&L for USDJPY at this risk level if uj_raw is not None: uj_daily = build_daily_pnl_series(uj_raw, 'USDJPY', uj_risk) else: uj_daily = build_synthetic_daily('USDJPY', uj_risk) # Scale UJ daily by risk ratio relative to base 0.75% if uj_raw is None: uj_daily = uj_daily * (uj_risk / 0.0075) result = run_mc_simulation(eu_daily, uj_daily, uj_risk) if result: results.append(result) r = result print(f" Pass rate: {r['pass_rate']}%") print(f" Daily breach: {r['daily_breach_rate']}%") print(f" Median time: {r['median_challenge_months']}m " f"(P25={r['p25_months']}m)") print(f" Sharpe: {r['combined_sharpe']}") if not results: print("No results — check trade data paths") return df = pd.DataFrame(results) # ── Find optimal ───────────────────────────────────────────────────────── safe = df[(df['daily_breach_rate'] == 0) & (df['overall_breach_rate'] == 0)] if not safe.empty: optimal = safe.loc[safe['median_challenge_months'].idxmin()] optimal_risk = optimal['uj_risk_pct'] else: # Find last safe level safe_any = df[df['overall_breach_rate'] == 0] optimal = safe_any.iloc[-1] if not safe_any.empty else df.iloc[2] optimal_risk = optimal['uj_risk_pct'] # ── Save CSV ────────────────────────────────────────────────────────────── df.to_csv(OUTPUT_DIR / "sensitivity_results.csv", index=False) # ── Summary report ──────────────────────────────────────────────────────── report_lines = [ "USDJPY RISK SENSITIVITY — SUMMARY REPORT", f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", f"EURUSD risk: LOCKED at {EU_RISK_PCT:.2%}", f"Simulations: {N_SIMS:,} × 3-year Monte Carlo", "=" * 65, "", f"{'UJ Risk':>8} {'Pass%':>6} {'Daily%':>7} {'Median':>7} " f"{'P25':>5} {'P75':>5} {'Sharpe':>7}", "-" * 65, ] for _, r in df.iterrows(): star = " ← OPTIMAL" if abs(r['uj_risk_pct'] - optimal_risk) < 0.0001 else "" report_lines.append( f"{r['uj_risk_pct']:>8.2%} " f"{r['pass_rate']:>6.1f}% " f"{r['daily_breach_rate']:>6.2f}% " f"{r['median_challenge_months']:>6.1f}m " f"{r['p25_months']:>4.1f}m " f"{r['p75_months']:>4.1f}m " f"{r['combined_sharpe']:>7.3f}" f"{star}" ) report_lines += [ "", "=" * 65, f"CONCLUSION: Optimal USDJPY risk = {optimal_risk:.2%}", f" Pass rate: {optimal['pass_rate']}%", f" Daily breach: {optimal['daily_breach_rate']}%", f" Median challenge: {optimal['median_challenge_months']}m", f" Combined Sharpe: {optimal['combined_sharpe']}", "", "Note: daily breach tested vs 1-Step ($3k) — the stricter limit.", "If 0% breach on 1-Step, 2-Step ($5k daily) is automatically safe.", "EURUSD risk remains locked at 0.75%.", "Do not re-optimise EURUSD risk — this is a USDJPY-only decision.", "", "Next step: if optimal > 0.75%, consider increasing USDJPY risk", "after confirming with 6+ months of live paper trading data.", "=" * 65, ] report_text = "\n".join(report_lines) (OUTPUT_DIR / "summary_report.txt").write_text(report_text, encoding="utf-8") print("\n" + report_text) # ── Charts ──────────────────────────────────────────────────────────────── fig = plt.figure(figsize=(16, 10)) fig.patch.set_facecolor('#0d1117') gs = gridspec.GridSpec(2, 3, figure=fig, hspace=0.4, wspace=0.35) risk_labels = [f"{r:.2%}" for r in df['uj_risk_pct']] opt_idx = df.index[df['uj_risk_pct'] == optimal_risk].tolist() def ax_style(ax, title): ax.set_facecolor('#161b22') ax.tick_params(colors='#7d8590', labelsize=8) ax.set_title(title, color='#e6edf3', fontsize=10, fontweight='bold', pad=8) ax.set_xlabel('USDJPY Risk %', color='#7d8590', fontsize=8) for spine in ax.spines.values(): spine.set_edgecolor('#21262d') ax.set_xticks(range(len(risk_labels))) ax.set_xticklabels(risk_labels, rotation=45, ha='right') if opt_idx: ax.axvline(x=opt_idx[0], color='#ffb830', alpha=0.3, linestyle='--', linewidth=1.5) # 1 — Pass rate ax1 = fig.add_subplot(gs[0, 0]) colors = ['#00e87a' if r == 0 else '#ff3d5a' for r in df['overall_breach_rate']] ax1.bar(range(len(df)), df['pass_rate'], color=colors, alpha=0.8) ax1.axhline(y=99.9, color='#ffb830', linestyle='--', linewidth=1) ax1.set_ylabel('Pass Rate %', color='#7d8590', fontsize=8) ax_style(ax1, 'FTMO Pass Rate') # 2 — Daily breach rate ax2 = fig.add_subplot(gs[0, 1]) ax2.bar(range(len(df)), df['daily_breach_rate'], color='#ff3d5a', alpha=0.7) ax2.axhline(y=0, color='#00e87a', linestyle='-', linewidth=1) ax2.set_ylabel('Daily Breach %', color='#7d8590', fontsize=8) ax_style(ax2, 'Daily DD Breach Rate (target: 0%)') # 3 — Median challenge time ax3 = fig.add_subplot(gs[0, 2]) ax3.bar(range(len(df)), df['median_challenge_months'], color='#3d9fff', alpha=0.8) ax3.errorbar(range(len(df)), df['median_challenge_months'], yerr=[df['median_challenge_months'] - df['p25_months'], df['p75_months'] - df['median_challenge_months']], fmt='none', color='#7d8590', capsize=4, linewidth=1) ax3.set_ylabel('Months', color='#7d8590', fontsize=8) ax_style(ax3, 'Median Challenge Time (with P25/P75)') # 4 — Combined Sharpe ax4 = fig.add_subplot(gs[1, 0]) ax4.plot(range(len(df)), df['combined_sharpe'], color='#a371f7', marker='o', linewidth=2, markersize=6) ax4.set_ylabel('Sharpe Ratio', color='#7d8590', fontsize=8) ax_style(ax4, 'Combined Portfolio Sharpe') # 5 — P95 Max Drawdown ax5 = fig.add_subplot(gs[1, 1]) ax5.bar(range(len(df)), df['p95_max_dd'], color='#ff3d5a', alpha=0.6) ax5.axhline(y=OVERALL_LIMIT, color='#ffb830', linestyle='--', linewidth=1.5, label='$10k limit') ax5.axhline(y=DAILY_LIMIT_1STEP, color='#ff8c00', linestyle='--', linewidth=1, label='$3k daily (1-Step)') ax5.axhline(y=DAILY_LIMIT_2STEP, color='#ffb830', linestyle=':', linewidth=1, label='$5k daily (2-Step)') ax5.set_ylabel('Max DD $ (P95)', color='#7d8590', fontsize=8) ax5.legend(fontsize=7, labelcolor='#7d8590', facecolor='#161b22', edgecolor='#21262d') ax_style(ax5, 'P95 Maximum Drawdown vs FTMO Limits') # 6 — Summary table ax6 = fig.add_subplot(gs[1, 2]) ax6.axis('off') table_data = [] for _, r in df.iterrows(): star = "★" if abs(r['uj_risk_pct'] - optimal_risk) < 0.0001 else "" table_data.append([ f"{r['uj_risk_pct']:.2%}{star}", f"{r['pass_rate']:.0f}%", f"{r['median_challenge_months']:.1f}m", f"{r['combined_sharpe']:.2f}", ]) tbl = ax6.table( cellText=table_data, colLabels=['UJ Risk', 'Pass%', 'Time', 'Sharpe'], cellLoc='center', loc='center', bbox=[0, 0, 1, 1] ) tbl.auto_set_font_size(False) tbl.set_fontsize(8) for (r, c), cell in tbl.get_celld().items(): cell.set_facecolor('#161b22' if r % 2 == 0 else '#0d1117') cell.set_text_props(color='#e6edf3') cell.set_edgecolor('#21262d') if r == 0: cell.set_facecolor('#21262d') ax6.set_title('Results Summary ★=optimal', color='#e6edf3', fontsize=10, fontweight='bold', pad=8) fig.suptitle( f'USDJPY Risk Sensitivity | EURUSD locked at 0.75% | ' f'{N_SIMS:,} MC sims | Optimal: {optimal_risk:.2%}', color='#e6edf3', fontsize=12, fontweight='bold', y=0.98 ) chart_path = OUTPUT_DIR / "sensitivity_charts.png" plt.savefig(chart_path, dpi=150, bbox_inches='tight', facecolor='#0d1117') plt.close() print(f"\n Charts saved: {chart_path}") print(f" Results CSV: {OUTPUT_DIR / 'sensitivity_results.csv'}") print(f" Report: {OUTPUT_DIR / 'summary_report.txt'}") print(f"\n OPTIMAL USDJPY RISK: {optimal_risk:.2%}") if __name__ == "__main__": main()