""" portfolio_ftmo_v1.py ===================== Portfolio-level FTMO challenge simulation across EURUSD, GBPUSD, USDJPY, and AUDUSD simultaneously. Portfolio Rules --------------- - One trade at a time PER PAIR (4 concurrent max) - All pairs share the same FTMO $100k account - Daily DD tracks the SUM of all pair losses on each day - Each pair uses the same signal scaling (z-score bands) - Each pair uses the same 0.30% base risk Key Question ------------ Does running 4 pairs simultaneously: 1. Reduce avg challenge duration (more trades per year)? 2. Keep 95th pct combined DD below 10%? 3. Maintain adequate pass rate? Monte Carlo Approach -------------------- For each simulation: - Shuffle returns WITHIN each pair independently - Keep the timeline (entry/exit dates) fixed per pair - Combine all shuffled trade logs chronologically - Apply portfolio-level FTMO rules This captures: - Intra-pair sequence risk (own losing streaks) - Cross-pair overlap on same dates (combined daily DD risk) - Realistic multi-pair challenge dynamics Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\portfolio_ftmo_v1.py Run from project root (after multi_pair_signals_v1.py): python src/research/portfolio_ftmo_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" if str(SRC_PATH) not in sys.path: sys.path.append(str(SRC_PATH)) TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" OUTPUT_DIR = TRADES_DIR # ── FTMO Parameters ─────────────────────────────────────────────────────────── ACCOUNT_START = 100_000.0 PROFIT_TARGET = ACCOUNT_START * 0.10 # +$10,000 MAX_OVERALL_LOSS = ACCOUNT_START * 0.10 # floor $90,000 MAX_DAILY_LOSS = ACCOUNT_START * 0.05 # $5,000 per day # ── Sizing Parameters (validated) ───────────────────────────────────────────── BASE_RISK_PCT = 0.0030 # 0.30% per trade STOP_PCT = 0.0025 BASE_NOTIONAL = (ACCOUNT_START * BASE_RISK_PCT) / STOP_PCT # $120,000 NOTIONAL_HARD_CAP = BASE_NOTIONAL * 3.0 # ── Signal scaling bands (validated) ───────────────────────────────────────── ZSCORE_BANDS = [ (2.00, 2.75, 0.75), (2.75, 3.50, 1.00), (3.50, 4.50, 1.50), (4.50, 99.0, 2.00), ] N_SIMS = 2000 RANDOM_SEED = 42 YEARS = 22.0 # ── All pairs (EURUSD + new pairs) ─────────────────────────────────────────── ALL_PAIRS = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"] # EURUSD uses growth_52h trade log (best single-pair candidate) PAIR_FILE_MAP = { "EURUSD": "trades_growth_52h.csv", "GBPUSD": "trades_GBPUSD.csv", "USDJPY": "trades_USDJPY.csv", "AUDUSD": "trades_AUDUSD.csv", } # ── Signal scaling ──────────────────────────────────────────────────────────── def get_multiplier(zscore_abs: float) -> float: for min_z, max_z, mult in ZSCORE_BANDS: if min_z <= zscore_abs < max_z: return mult return ZSCORE_BANDS[-1][2] # ── Load all trade logs ─────────────────────────────────────────────────────── def load_all_trades() -> dict: """ Loads trade logs for all available pairs. Returns dict of {pair: DataFrame}. """ all_trades = {} missing = [] for pair in ALL_PAIRS: filename = PAIR_FILE_MAP[pair] csv_path = TRADES_DIR / filename if not csv_path.exists(): missing.append(pair) continue df = pd.read_csv(csv_path, parse_dates=["entry_time", "exit_time"]) df = df.sort_values("entry_time").reset_index(drop=True) # Ensure zscore_abs exists (EURUSD log may not have it natively) if "zscore_abs" not in df.columns: df["zscore_abs"] = 2.75 # default to base band all_trades[pair] = df if missing: print(f"\n [WARNING] Missing trade logs for: {missing}") if "EURUSD" in missing: print(" [ERROR] EURUSD trade log required. Run export_trade_logs_v1.py first.") return all_trades # ── Build portfolio trade event stream ─────────────────────────────────────── def build_portfolio_stream(all_trades: dict) -> pd.DataFrame: """ Combines all pair trade logs into a single chronological event stream. Each row = one trade event from one pair. """ dfs = [] for pair, df in all_trades.items(): t = df[["entry_time", "exit_time", "return", "zscore_abs"]].copy() t["pair"] = pair t["multiplier"]= t["zscore_abs"].apply(get_multiplier) t["notional"] = (BASE_NOTIONAL * t["multiplier"]).clip(upper=NOTIONAL_HARD_CAP) t["dollar_pnl"]= t["return"] * t["notional"] dfs.append(t) combined = pd.concat(dfs, ignore_index=True) combined = combined.sort_values("entry_time").reset_index(drop=True) return combined # ── Portfolio FTMO simulation (one pass) ───────────────────────────────────── def simulate_portfolio_ftmo( portfolio_stream : pd.DataFrame, shuffled_returns : dict | None = None, ) -> dict: """ Simulates FTMO challenge on a combined portfolio of trades. Each pair's trades run independently (one at a time per pair). Daily DD is the combined P&L across all pairs on each calendar day. shuffled_returns: dict of {pair: array} if running Monte Carlo. If None, uses original return sequence (sequential simulation). """ # Build event stream with optionally shuffled returns if shuffled_returns is not None: # Apply shuffled returns while keeping dates intact events = [] for pair, df in portfolio_stream.groupby("pair"): t = df.copy().reset_index(drop=True) if pair in shuffled_returns: sh_ret = shuffled_returns[pair] # Keep multipliers aligned with shuffled returns sh_mult = t["multiplier"].values[ np.random.default_rng().permutation(len(t)) ] t["dollar_pnl"] = sh_ret * (BASE_NOTIONAL * sh_mult).clip(max=NOTIONAL_HARD_CAP) events.append(t) events = pd.concat(events, ignore_index=True).sort_values("entry_time").reset_index(drop=True) else: events = portfolio_stream.copy() balance = ACCOUNT_START peak_balance = ACCOUNT_START max_dd_reached = 0.0 daily_pnl = {} outcome = "INCOMPLETE" trading_days_set = set() trades_taken = 0 for _, row in events.iterrows(): date_key = str(pd.Timestamp(row["exit_time"]).date()) if date_key not in daily_pnl: daily_pnl[date_key] = 0.0 dollar_pnl = float(row["dollar_pnl"]) balance += dollar_pnl trades_taken += 1 trading_days_set.add(date_key) daily_pnl[date_key] += dollar_pnl if balance > peak_balance: peak_balance = balance current_dd = (peak_balance - balance) / ACCOUNT_START if current_dd > max_dd_reached: max_dd_reached = current_dd if daily_pnl[date_key] < -MAX_DAILY_LOSS: outcome = "BREACH_DAILY_DD" break if balance <= (ACCOUNT_START - MAX_OVERALL_LOSS): outcome = "BREACH_OVERALL_DD" break if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" break if outcome == "INCOMPLETE": if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" elif (peak_balance - balance) / ACCOUNT_START >= 0.10: outcome = "BREACH_OVERALL_DD" return { "outcome" : outcome, "final_balance" : balance, "max_dd_reached_pct" : max_dd_reached * 100, "total_pnl_pct" : (balance / ACCOUNT_START - 1) * 100, "trades_taken" : trades_taken, "trading_days" : len(trading_days_set), } # ── Monte Carlo ─────────────────────────────────────────────────────────────── def run_portfolio_mc( all_trades : dict, portfolio_stream: pd.DataFrame, n_sims : int = N_SIMS, ) -> pd.DataFrame: """ Shuffles returns within each pair independently and runs n_sims simulations. """ rng = np.random.default_rng(RANDOM_SEED) results = [] for sim_i in range(n_sims): # Build shuffled events — returns shuffled per pair, dates fixed events = [] for pair, df in portfolio_stream.groupby("pair"): t = df.copy().reset_index(drop=True) idx = rng.permutation(len(t)) sh_ret = t["return"].values[idx] sh_mult = t["multiplier"].values[idx] t["dollar_pnl"] = sh_ret * (BASE_NOTIONAL * sh_mult).clip(max=NOTIONAL_HARD_CAP) events.append(t) shuffled_events = ( pd.concat(events, ignore_index=True) .sort_values("exit_time") .reset_index(drop=True) ) # Run FTMO sim directly on pre-computed shuffled events balance = ACCOUNT_START peak_balance = ACCOUNT_START max_dd_reached = 0.0 daily_pnl = {} outcome = "INCOMPLETE" trading_days_set = set() trades_taken = 0 for _, row in shuffled_events.iterrows(): date_key = str(pd.Timestamp(row["exit_time"]).date()) if date_key not in daily_pnl: daily_pnl[date_key] = 0.0 dollar_pnl = float(row["dollar_pnl"]) balance += dollar_pnl trades_taken += 1 trading_days_set.add(date_key) daily_pnl[date_key] += dollar_pnl if balance > peak_balance: peak_balance = balance current_dd = (peak_balance - balance) / ACCOUNT_START if current_dd > max_dd_reached: max_dd_reached = current_dd if daily_pnl[date_key] < -MAX_DAILY_LOSS: outcome = "BREACH_DAILY_DD" break if balance <= (ACCOUNT_START - MAX_OVERALL_LOSS): outcome = "BREACH_OVERALL_DD" break if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" break if outcome == "INCOMPLETE": if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" elif (peak_balance - balance) / ACCOUNT_START >= 0.10: outcome = "BREACH_OVERALL_DD" results.append({ "sim" : sim_i, "outcome" : outcome, "final_balance" : balance, "max_dd_pct" : max_dd_reached * 100, "total_pnl_pct" : (balance / ACCOUNT_START - 1) * 100, "trades_taken" : trades_taken, }) return pd.DataFrame(results) # ── Single-pair MC benchmark ────────────────────────────────────────────────── def run_single_pair_mc(df: pd.DataFrame, pair: str, n_sims: int = N_SIMS) -> pd.DataFrame: """Runs single-pair FTMO MC for comparison against portfolio.""" rng = np.random.default_rng(RANDOM_SEED) returns = df["return"].values zscores = df["zscore_abs"].values results = [] for _ in range(n_sims): idx = rng.permutation(len(returns)) sh_ret = returns[idx] sh_mult = np.array([get_multiplier(z) for z in zscores[idx]]) balance = ACCOUNT_START peak = ACCOUNT_START max_dd = 0.0 daily_pnl = {} outcome = "INCOMPLETE" trading_days = set() trades_taken = 0 for ret, mult, ex_dt in zip(sh_ret, sh_mult, df["exit_time"].values): date_key = str(pd.Timestamp(ex_dt).date()) if date_key not in daily_pnl: daily_pnl[date_key] = 0.0 notional = min(BASE_NOTIONAL * mult, NOTIONAL_HARD_CAP) pnl = ret * notional balance += pnl trades_taken += 1 trading_days.add(date_key) daily_pnl[date_key] += pnl if balance > peak: peak = balance dd = (peak - balance) / ACCOUNT_START if dd > max_dd: max_dd = dd if daily_pnl[date_key] < -MAX_DAILY_LOSS: outcome = "BREACH_DAILY_DD"; break if balance <= ACCOUNT_START - MAX_OVERALL_LOSS: outcome = "BREACH_OVERALL_DD"; break if balance >= ACCOUNT_START + PROFIT_TARGET: outcome = "PASS"; break if outcome == "INCOMPLETE": outcome = ("PASS" if balance >= ACCOUNT_START + PROFIT_TARGET else "BREACH_OVERALL_DD" if (peak - balance) / ACCOUNT_START >= 0.10 else "INCOMPLETE") results.append({ "outcome" : outcome, "max_dd" : max_dd * 100, "trades" : trades_taken, }) mc = pd.DataFrame(results) passing = mc[mc["outcome"] == "PASS"] return { "pass_rate" : (mc["outcome"] == "PASS").mean() * 100, "breach_rate" : mc["outcome"].str.startswith("BREACH").mean() * 100, "dd_95" : mc["max_dd"].quantile(0.95), "avg_trades_pass": passing["trades"].mean() if len(passing) > 0 else np.nan, "med_trades_pass": passing["trades"].median() if len(passing) > 0 else np.nan, } # ── Summarise MC ────────────────────────────────────────────────────────────── def summarise_mc(mc: pd.DataFrame) -> dict: passing = mc[mc["outcome"] == "PASS"] return { "pass_rate" : (mc["outcome"] == "PASS").mean() * 100, "breach_rate" : mc["outcome"].str.startswith("BREACH").mean() * 100, "daily_breach" : (mc["outcome"] == "BREACH_DAILY_DD").mean() * 100, "dd_50" : mc["max_dd_pct"].quantile(0.50), "dd_75" : mc["max_dd_pct"].quantile(0.75), "dd_95" : mc["max_dd_pct"].quantile(0.95), "danger_pct" : (mc["max_dd_pct"] > 8.0).mean() * 100, "avg_trades_pass": passing["trades_taken"].mean() if len(passing) > 0 else np.nan, "med_trades_pass": passing["trades_taken"].median() if len(passing) > 0 else np.nan, } # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 70) print("PORTFOLIO FTMO V1 — 4-Pair Portfolio") print(f" Pairs : {', '.join(ALL_PAIRS)}") print(f" Base risk : {BASE_RISK_PCT:.2%} per trade per pair") print(f" Notional : ${BASE_NOTIONAL:,.0f} base (signal-scaled)") print(f" MC sims : {N_SIMS:,}") print(f" DD limit : 10% (95th pct)") print("=" * 70) # Load all trade logs print("\nLoading trade logs...") all_trades = load_all_trades() if not all_trades: print("[ERROR] No trade logs found. Run export_trade_logs_v1.py and multi_pair_signals_v1.py first.") return available_pairs = list(all_trades.keys()) print(f" Loaded: {available_pairs}") # Per-pair summary print(f"\n{'─'*70}") print("PER-PAIR TRADE LOG SUMMARY") print(f"{'─'*70}") total_annual = 0 print(f"\n {'Pair':<12}{'Trades':>8}{'Per/yr':>8}{'WR':>8}" f"{'AvgRet':>10}{'AvgPnL$':>10}") print(f" {'─'*56}") for pair, df in all_trades.items(): n = len(df) per_yr= n / YEARS total_annual += per_yr avg_pnl = df["return"].mean() * BASE_NOTIONAL print(f" {pair:<12}{n:>8}{per_yr:>8.1f}" f"{(df['return']>0).mean():>7.2%}" f"{df['return'].mean():>10.6f}" f"{avg_pnl:>+10.2f}") print(f"\n Total portfolio trades/year: {total_annual:.1f}") # Build combined portfolio stream portfolio_stream = build_portfolio_stream(all_trades) print(f"\n Combined portfolio events: {len(portfolio_stream):,}") # Max simultaneous open trades analysis print(f"\n Daily DD risk analysis:") print(f" Max pairs trading same day: if all 4 stop out simultaneously") max_daily_loss_scenario = 4 * BASE_NOTIONAL * STOP_PCT * 2.0 # 4 pairs × max notional × stop print(f" Worst case daily loss (4× max notional × stop): " f"${max_daily_loss_scenario:,.0f}") print(f" Daily DD limit: ${MAX_DAILY_LOSS:,.0f}") safe = max_daily_loss_scenario <= MAX_DAILY_LOSS print(f" Safe: {'✓' if safe else '✗ RISK'}") # ── Sequential simulation ───────────────────────────────────────────────── print(f"\n{'─'*70}") print("SEQUENTIAL SIMULATION (historical order)") print(f"{'─'*70}") seq = simulate_portfolio_ftmo(portfolio_stream) print(f"\n Portfolio ({len(available_pairs)} pairs):") print(f" Outcome : {seq['outcome']}") print(f" Final balance: ${seq['final_balance']:>12,.2f}") print(f" Total P&L : {seq['total_pnl_pct']:>+.2f}%") print(f" Max DD : {seq['max_dd_reached_pct']:.3f}%") print(f" Trades taken : {seq['trades_taken']}") print(f" Trading days : {seq['trading_days']}") # ── Monte Carlo ─────────────────────────────────────────────────────────── print(f"\n{'─'*70}") print(f"MONTE CARLO ({N_SIMS:,} simulations)") print(f"{'─'*70}") print("\n Running portfolio MC...") mc_portfolio = run_portfolio_mc(all_trades, portfolio_stream) s_port = summarise_mc(mc_portfolio) # Single-pair benchmark (EURUSD only, validated) print(" Running EURUSD single-pair MC for comparison...") s_single = run_single_pair_mc(all_trades["EURUSD"], "EURUSD") # ── Print comparison ────────────────────────────────────────────────────── per_yr_single = len(all_trades["EURUSD"]) / YEARS per_yr_portfolio = total_annual avg_mths_single = (s_single["avg_trades_pass"] / per_yr_single * 12 if not np.isnan(s_single["avg_trades_pass"]) else np.nan) med_mths_single = (s_single["med_trades_pass"] / per_yr_single * 12 if not np.isnan(s_single["med_trades_pass"]) else np.nan) avg_mths_port = (s_port["avg_trades_pass"] / per_yr_portfolio * 12 if not np.isnan(s_port["avg_trades_pass"]) else np.nan) med_mths_port = (s_port["med_trades_pass"] / per_yr_portfolio * 12 if not np.isnan(s_port["med_trades_pass"]) else np.nan) print(f"\n {'Metric':<32}{'EURUSD only':>14}{'Portfolio':>14}") print(f" {'─'*60}") metrics = [ ("Pass rate", f"{s_single['pass_rate']:.2f}%", f"{s_port['pass_rate']:.2f}%"), ("Breach rate", f"{s_single['breach_rate']:.2f}%", f"{s_port['breach_rate']:.2f}%"), ("Daily DD breach", f"{s_single.get('daily_breach',0.0):.2f}%", f"{s_port['daily_breach']:.2f}%"), ("Median max DD", f"{s_single['dd_95']*0:.3f}%", f"{s_port['dd_50']:.3f}%"), ("95th pct max DD", f"{s_single['dd_95']:.3f}%", f"{s_port['dd_95']:.3f}% {'✓' if s_port['dd_95']<10 else '✗'}"), ("Danger >8% DD", f"{s_single.get('danger_pct',0):.1f}%", f"{s_port['danger_pct']:.1f}%"), ("Avg trades/pass", f"{s_single['avg_trades_pass']:.0f}", f"{s_port['avg_trades_pass']:.0f}"), ("Med trades/pass", f"{s_single['med_trades_pass']:.0f}", f"{s_port['med_trades_pass']:.0f}"), ("Trades/year", f"{per_yr_single:.1f}", f"{per_yr_portfolio:.1f}"), ("Avg months/pass", f"{avg_mths_single:.1f}", f"{avg_mths_port:.1f}"), ("Med months/pass", f"{med_mths_single:.1f}", f"{med_mths_port:.1f}"), ] for label, v1, v2 in metrics: print(f" {label:<32}{v1:>14}{v2:>14}") # Months saved if not np.isnan(avg_mths_single) and not np.isnan(avg_mths_port): months_saved = avg_mths_single - avg_mths_port print(f"\n Challenge duration improvement: {months_saved:+.1f} months average") print(f" Challenge duration improvement: " f"{med_mths_single - med_mths_port:+.1f} months median") # ── Trades-to-pass distribution ─────────────────────────────────────────── passing = mc_portfolio[mc_portfolio["outcome"] == "PASS"] if len(passing) > 0: print(f"\n Portfolio trades-to-pass distribution:") print(f" 5th pct : {passing['trades_taken'].quantile(0.05):.0f}") print(f" 25th pct : {passing['trades_taken'].quantile(0.25):.0f}" f" (~{passing['trades_taken'].quantile(0.25)/per_yr_portfolio*12:.1f} months)") print(f" Median : {passing['trades_taken'].median():.0f}" f" (~{med_mths_port:.1f} months)") print(f" 75th pct : {passing['trades_taken'].quantile(0.75):.0f}" f" (~{passing['trades_taken'].quantile(0.75)/per_yr_portfolio*12:.1f} months)") print(f" 95th pct : {passing['trades_taken'].quantile(0.95):.0f}") # ── Save results ────────────────────────────────────────────────────────── mc_portfolio["scenario"] = "portfolio" out_path = OUTPUT_DIR / "portfolio_mc_results.csv" mc_portfolio.to_csv(out_path, index=False) print(f"\n MC results saved: {out_path}") # ── Final assessment ────────────────────────────────────────────────────── print(f"\n{'='*70}") print("ASSESSMENT") print(f"{'='*70}") dd_ok = s_port["dd_95"] < 10.0 pass_ok = s_port["pass_rate"] >= 95.0 if dd_ok and pass_ok: print(f""" ✓ PORTFOLIO VALIDATED 95th pct DD: {s_port['dd_95']:.3f}% < 10% limit Pass rate : {s_port['pass_rate']:.2f}% ≥ 95% target Med months : {med_mths_port:.1f} (vs {med_mths_single:.1f} single-pair) The multi-pair portfolio improves challenge pacing while maintaining FTMO compliance. Freeze these parameters and proceed to execution. """) elif dd_ok and not pass_ok: print(f""" ~ PARTIAL — DD OK but pass rate below 95% 95th pct DD: {s_port['dd_95']:.3f}% ✓ Pass rate : {s_port['pass_rate']:.2f}% — below 95% target The new pairs may be adding noise. Review per-pair win rates. Consider using only pairs with positive avg return and >35% win rate. """) else: print(f""" ✗ DD LIMIT EXCEEDED — REVIEW REQUIRED 95th pct DD: {s_port['dd_95']:.3f}% — above 10% limit The portfolio's combined daily loss risk is too high. Options: 1. Reduce base risk to 0.25% (from 0.30%) 2. Add only pairs with independently validated win rates >33% 3. Use a max-concurrent-trades cap (e.g. max 2 pairs at once) Run risk_calibration_v1.py with the portfolio return stream to find the correct risk% for the combined portfolio. """) if __name__ == "__main__": main()