""" mae_crossover_analysis_v1.py ============================= Finds the empirically optimal stop loss level by analysing the Maximum Adverse Excursion (MAE) distributions of winning vs losing trades. The core question: at what adverse excursion level does a trade become "more likely to be a loser than a winner"? That crossover point = the optimal stop. Setting the stop there maximises the number of trades allowed to recover while cutting true losers early. Methodology ----------- For each trade in both candidates: - MAE is already computed (worst adverse move during hold, as negative %) - We classify trades as winners (return > 0) or losers (return <= 0) - We compute the density of winners vs losers at each MAE threshold - The crossover = the MAE level where loser density exceeds winner density Outputs ------- 1. MAE distribution statistics for winners vs losers 2. Crossover point table at multiple thresholds 3. Stop ladder comparison — how does changing the stop affect: - Trade count (some losers cut early, freeing up capacity) - Win rate - Avg return - Final equity - Max drawdown 4. Recommendation for optimal stop Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\mae_crossover_analysis_v1.py Run from project root: python src/research/mae_crossover_analysis_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys # ── Path setup ──────────────────────────────────────────────────────────────── 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 CANDIDATES = ["growth_52h", "smoother_25h"] # Stop ladder to test (as positive decimals: 0.001 = 0.1%) STOP_LADDER = [0.0010, 0.0015, 0.0020, 0.0025, 0.0030, 0.0035, 0.0040, 0.0050, 0.0060, 0.0075, 0.0100] # ── MAE crossover analysis ──────────────────────────────────────────────────── def analyse_mae_crossover(df: pd.DataFrame, name: str) -> dict: """ Computes winner vs loser MAE distributions and finds the crossover point. MAE is stored as negative (adverse move). We flip to positive for analysis. """ df = df.copy() df["mae_abs"] = df["mae"].abs() # 0.003 = moved 0.3% against us df["is_winner"] = (df["return"] > 0).astype(int) winners = df[df["is_winner"] == 1]["mae_abs"] losers = df[df["is_winner"] == 0]["mae_abs"] print(f"\n{'─'*60}") print(f"MAE ANALYSIS: {name.upper()}") print(f"{'─'*60}") print(f" Total trades : {len(df)}") print(f" Winners : {len(winners)} ({len(winners)/len(df):.1%})") print(f" Losers : {len(losers)} ({len(losers)/len(df):.1%})") # ── Percentile breakdown ────────────────────────────────────────────────── pcts = [10, 25, 50, 75, 90, 95, 99] print(f"\n MAE Percentile Table (% adverse excursion):") print(f" {'Pct':<8}{'Winners MAE':>14}{'Losers MAE':>14}{'Ratio W/L':>12}") print(f" {'─'*50}") for p in pcts: w_val = np.percentile(winners, p) * 100 l_val = np.percentile(losers, p) * 100 ratio = w_val / l_val if l_val > 0 else 0 print(f" {p:<8}{w_val:>13.4f}%{l_val:>13.4f}%{ratio:>12.3f}") print(f"\n Mean MAE — Winners: {winners.mean()*100:.4f}% " f"Losers: {losers.mean()*100:.4f}%") print(f" Median MAE — Winners: {winners.median()*100:.4f}% " f"Losers: {losers.median()*100:.4f}%") # ── Crossover analysis ──────────────────────────────────────────────────── # At each threshold, what fraction of winners and losers have MAE below it? # Crossover = threshold where cumulative loser density overtakes winner density print(f"\n MAE Threshold Crossover Table:") print(f" (At each stop level, what % of winners would be stopped out?)") print(f"\n {'Stop %':<10}{'Winners stopped':>18}{'Losers stopped':>16}" f"{'Survive ratio':>15}{'Edge score':>12}") print(f" {'─'*72}") crossover_data = [] for stop in STOP_LADDER: # Winners stopped = winners with MAE > stop (they dip past stop but recover) # We want to AVOID stopping these out w_stopped = (winners > stop).sum() / len(winners) # fraction of winners # that would have been stopped at this level (their MAE exceeded the stop) # Actually we want: winners whose MAE > stop = those that WOULD be stopped # if we set stop at this level. Lower is better. w_stopped_pct = (winners > stop).mean() * 100 l_stopped_pct = (losers > stop).mean() * 100 # Survive ratio = winners NOT stopped / losers NOT stopped # We want this to be HIGH — means most winners survive, most losers stopped w_survive = (winners <= stop).mean() l_survive = (losers <= stop).mean() survive_ratio = (w_survive / l_survive) if l_survive > 0 else 999.0 # Edge score = (losers stopped - winners stopped) — higher = better stop # Penalises stops that cut winners early edge_score = l_stopped_pct - w_stopped_pct crossover_data.append({ "stop" : stop, "w_stopped_pct" : w_stopped_pct, "l_stopped_pct" : l_stopped_pct, "survive_ratio" : survive_ratio, "edge_score" : edge_score, }) marker = " ◄ CURRENT" if abs(stop - 0.0025) < 0.00001 else "" print(f" {stop*100:>6.2f}% {w_stopped_pct:>14.1f}% " f"{l_stopped_pct:>12.1f}% {survive_ratio:>13.3f} " f"{edge_score:>9.1f}{marker}") cd = pd.DataFrame(crossover_data) # Best stop = highest edge score (maximum separation between losers stopped # and winners stopped) best_idx = cd["edge_score"].idxmax() best_stop = cd.loc[best_idx, "stop"] print(f"\n Optimal stop by edge score: {best_stop*100:.2f}%") print(f" (Maximises losers cut - winners saved)") # ── Winner MAE distribution summary ────────────────────────────────────── print(f"\n Key insight — of all WINNING trades:") for stop in [0.0010, 0.0015, 0.0020, 0.0025, 0.0030, 0.0040, 0.0050]: pct_exceed = (winners > stop).mean() * 100 print(f" {stop*100:.2f}% stop would eliminate {pct_exceed:.1f}% of winners" f" (they dipped past this level before winning)") return { "name" : name, "crossover_df": cd, "best_stop" : best_stop, "winners" : winners, "losers" : losers, } # ── Stop ladder backtest ────────────────────────────────────────────────────── def stop_ladder_backtest(df: pd.DataFrame, name: str, hold_hours: int) -> pd.DataFrame: """ Reruns the equity simulation at each stop level using the pre-computed MAE and MFE data — no need to reprocess price data. For each trade, if MAE_abs > stop_level → stop hit → return = -stop_level Otherwise → original return stands. This approximates the impact of different stops on the return series. Note: This does not recompute entries or Fibonacci pullbacks — it only changes the exit condition for existing trades. """ print(f"\n Stop Ladder Backtest: {name}") print(f" {'Stop %':<10}{'Trades':>8}{'Win rate':>10}{'Avg ret':>10}" f"{'Final eq':>12}{'Max DD':>10}{'Streak':>8}") print(f" {'─'*68}") rows = [] for stop in STOP_LADDER: t = df.copy() # Apply new stop: if MAE exceeded the stop, trade return = -stop - spread t["adj_return"] = t.apply( lambda r: (-stop - 0.0001) if r["mae_abs"] > stop else r["return"], axis=1, ) t["equity"] = (1 + t["adj_return"]).cumprod() t["peak"] = t["equity"].cummax() t["drawdown"] = t["equity"] / t["peak"] - 1 # Losing streak streak = max_streak = 0 for r in t["adj_return"]: if r <= 0: streak += 1 max_streak = max(max_streak, streak) else: streak = 0 n_stopped = (t["mae_abs"] > stop).sum() win_rate = (t["adj_return"] > 0).mean() avg_ret = t["adj_return"].mean() final_eq = t["equity"].iloc[-1] max_dd = t["drawdown"].min() marker = " ◄" if abs(stop - 0.0025) < 0.00001 else "" print(f" {stop*100:>6.2f}% {len(t):>8} {win_rate:>9.4%} " f"{avg_ret:>10.6f} {final_eq:>11.6f} {max_dd:>9.4%}" f" {max_streak:>4}{marker}") rows.append({ "stop" : stop, "win_rate" : win_rate, "avg_return": avg_ret, "final_eq" : final_eq, "max_dd" : max_dd, "max_streak": max_streak, "n_stopped" : n_stopped, "candidate" : name, }) return pd.DataFrame(rows) # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 60) print("MAE CROSSOVER ANALYSIS V1") print("Finding the empirically optimal stop loss level") print("=" * 60) all_crossover = {} all_ladder = {} for name in CANDIDATES: csv_path = TRADES_DIR / f"trades_{name}.csv" if not csv_path.exists(): print(f"\n[ERROR] {csv_path} not found. Run export_trade_logs_v1.py first.") continue df = pd.read_csv(csv_path, parse_dates=["entry_time", "exit_time"]) df["mae_abs"] = df["mae"].abs() # Determine hold_hours from candidate name hold_hours = 52 if "52h" in name else 25 # MAE crossover analysis result = analyse_mae_crossover(df, name) all_crossover[name] = result # Stop ladder backtest ladder_df = stop_ladder_backtest(df, name, hold_hours) all_ladder[name] = ladder_df # Save ladder results out_path = OUTPUT_DIR / f"stop_ladder_{name}.csv" ladder_df.to_csv(out_path, index=False) print(f"\n Saved: {out_path}") # ── Cross-candidate stop recommendation ────────────────────────────────── print(f"\n{'='*60}") print("STOP RECOMMENDATION SUMMARY") print(f"{'='*60}") for name, result in all_crossover.items(): best = result["best_stop"] cd = result["crossover_df"] # Also find the stop where >50% of winners are preserved (haven't dipped past) winners = result["winners"] for stop in STOP_LADDER: pct_winners_stopped = (winners > stop).mean() if pct_winners_stopped < 0.50: conservative_stop = stop break else: conservative_stop = STOP_LADDER[-1] print(f"\n {name}:") print(f" Current stop : 0.25%") print(f" Optimal (edge score) : {best*100:.2f}%") print(f" Conservative (save 50%+ of winners): {conservative_stop*100:.2f}%") # What % of winners are currently stopped out at 0.25%? pct_winners_killed = (winners > 0.0025).mean() * 100 print(f" Winners killed by current 0.25% stop: {pct_winners_killed:.1f}%") # Compare current vs optimal in ladder if name in all_ladder: ldf = all_ladder[name] curr = ldf[ldf["stop"] == 0.0025].iloc[0] opt_row = ldf[ldf["stop"] == best].iloc[0] print(f"\n Current stop 0.25% → equity {curr['final_eq']:.4f}, " f"DD {curr['max_dd']:.4%}, WR {curr['win_rate']:.4%}") print(f" Optimal stop {best*100:.2f}% → equity {opt_row['final_eq']:.4f}, " f"DD {opt_row['max_dd']:.4%}, WR {opt_row['win_rate']:.4%}") print(f"\n{'='*60}") print("INTERPRETATION GUIDE") print(f"{'='*60}") print(""" Reading the crossover table: ───────────────────────────── "Winners stopped" = % of winning trades that would have been cut at this stop level (their MAE exceeded the stop). These are trades you WANT to stay in — you are destroying edge by stopping them. "Losers stopped" = % of losing trades cut at this stop level. These you WANT to stop — you are saving capital. "Edge score" = (losers stopped) - (winners stopped) Higher = better stop separation. The optimal stop maximises this. Example: If at 0.25% stop: 66% winners stopped, 70% losers stopped Edge score = 70 - 66 = 4 (very poor separation) If at 0.60% stop: 20% winners stopped, 65% losers stopped Edge score = 65 - 20 = 45 (much better separation) The goal is to find the stop that cuts true losers efficiently while letting true winners breathe past their normal adverse dip. Next step: Take the optimal stop into the FTMO simulator and compare pass rates vs current 0.25% stop. """) if __name__ == "__main__": main()