""" take_profit_exit_v1.py ======================= Tests take profit targets combined with the z-score reversal exit. The Problem ----------- Avg MFE of winners = 0.699% but avg winning trade only captures ~0.45%. 35% of winner potential is left on the table due to z-score natural decay triggering the reversal exit before the full move completes. Solution -------- Add a TP target so winners exit cleanly at a defined profit level before z-score decay can cut them. Losers still exit via z-reversal or hard stop. This creates an asymmetric exit structure: Winners → TP hit → clean profitable exit Losers → z-reversal → early loss reduction Disaster → hard stop → tail risk protection Default → time exit → holds uncaptured opportunities Methods Tested -------------- A) Baseline: z=1.5 exit, no TP, 52h hold (best from previous test) B) TP 0.30% + z=1.5 exit + 0.25% stop + 52h max C) TP 0.40% + z=1.5 exit + 0.25% stop + 52h max D) TP 0.50% + z=1.5 exit + 0.25% stop + 52h max E) TP 0.60% + z=1.5 exit + 0.25% stop + 52h max F) TP 0.50% + z=1.5 exit + 0.25% stop + 36h max (shorter hold) G) TP 0.50% + z=1.5 exit + 0.25% stop + 24h max (even shorter) H) Two-stage: 50% close at TP 0.50%, 50% hold to z-exit or time (52h) Each method uses signal scaling (1x/1.5x/2x z-score bands). Full FTMO Monte Carlo on all methods. Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\take_profit_exit_v1.py Run from project root: python src/research/take_profit_exit_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)) from research.combined_candidate_matrix_v1 import ( build_frozen_signals, find_entry_pullback, get_first_m15_idx_at_or_after, ) from ingestion.price_loader_15m import load_eurusd_15m from features.spot_lag_v3 import get_model_ready_spot_lag_v3 TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" TRADES_DIR.mkdir(parents=True, exist_ok=True) # ── Frozen parameters ───────────────────────────────────────────────────────── THRESHOLD = 2.75 FIB = 0.786 STOP = 0.0025 # 0.25% hard stop SPREAD_COST = 0.0001 ALLOWED_HOURS = set(range(7, 17)) ZSCORE_REVERSAL = 1.5 # validated best from zscore_exit_tuning_v1 ACCOUNT_START = 100_000.0 PROFIT_TARGET_MC = ACCOUNT_START * 0.10 MAX_OVERALL = ACCOUNT_START * 0.10 MAX_DAILY = ACCOUNT_START * 0.05 BASE_RISK_PCT = 0.003 BASE_NOTIONAL = (ACCOUNT_START * BASE_RISK_PCT) / STOP # $120,000 ZSCORE_BANDS = [ (2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0), ] N_SIMS = 2000 RANDOM_SEED = 42 YEARS = 22.0 def get_multiplier(z: float) -> float: for lo, hi, mult in ZSCORE_BANDS: if lo <= z < hi: return mult return ZSCORE_BANDS[-1][2] # ── Core trade simulator with TP ────────────────────────────────────────────── def simulate_trade_with_tp( m15 : pd.DataFrame, signal_df : pd.DataFrame, entry_time : pd.Timestamp, entry_price : float, signal : int, hold_hours : int, stop : float = STOP, tp : float = None, # Take profit as decimal e.g. 0.005 = 0.5% z_threshold : float = ZSCORE_REVERSAL, spread_cost : float = SPREAD_COST, two_stage : bool = False, # Two-stage: 50% at TP, 50% continues ) -> dict | None: entry_idx = get_first_m15_idx_at_or_after(m15, entry_time) if entry_idx is None: return None hold_bars = hold_hours * 4 exit_idx = min(entry_idx + hold_bars, len(m15) - 1) path = m15.iloc[entry_idx : exit_idx + 1].copy() if path.empty: return None # Z-score window for exit monitoring signal_window = signal_df[ (signal_df["datetime"] >= entry_time) & (signal_df["datetime"] <= path.iloc[-1]["datetime"]) ].copy() # Full path MAE / MFE (for analysis) if signal == 1: full_mae = (path["low"].min() - entry_price) / entry_price full_mfe = (path["high"].max() - entry_price) / entry_price else: full_mae = -((path["high"].max() - entry_price) / entry_price) full_mfe = (entry_price - path["low"].min()) / entry_price exit_price = float(path.iloc[-1]["close"]) exit_reason = "time" stop_hit = False final_bar_i = len(path) - 1 # Two-stage tracking partial_closed = False partial_price = None partial_reason = None partial_bar_i = None for bar_i in range(len(path)): bar = path.iloc[bar_i] bar_time = bar["datetime"] # ── TP check ───────────────────────────────────────────────────────── if tp is not None: if signal == 1: tp_hit = (float(bar["high"]) - entry_price) / entry_price >= tp else: tp_hit = (entry_price - float(bar["low"])) / entry_price >= tp if tp_hit: tp_price = entry_price * (1 + tp) if signal == 1 \ else entry_price * (1 - tp) if two_stage and not partial_closed: # Close 50% at TP, let remaining 50% continue partial_closed = True partial_price = tp_price partial_reason = "tp_partial" partial_bar_i = bar_i # Don't break — continue monitoring the second half elif not two_stage: exit_price = tp_price exit_reason = "tp" final_bar_i = bar_i break # ── Stop check ─────────────────────────────────────────────────────── if signal == 1: adverse = (float(bar["low"]) - entry_price) / entry_price else: adverse = -((float(bar["high"]) - entry_price) / entry_price) if adverse <= -stop: exit_price = entry_price * (1 - stop) if signal == 1 \ else entry_price * (1 + stop) exit_reason = "stop" stop_hit = True final_bar_i = bar_i break # ── Z-score reversal exit ───────────────────────────────────────────── z_bars = signal_window[signal_window["datetime"] <= bar_time] if not z_bars.empty: current_z = float(z_bars.iloc[-1]["lag_zscore_24h_v3"]) if signal == 1 and current_z <= -z_threshold: exit_price = float(bar["close"]) exit_reason = "zscore_reversal" final_bar_i = bar_i break elif signal == -1 and current_z >= z_threshold: exit_price = float(bar["close"]) exit_reason = "zscore_reversal" final_bar_i = bar_i break # ── Calculate return ────────────────────────────────────────────────────── if stop_hit: raw_ret = -stop elif two_stage and partial_closed: # Blend: 50% closed at partial_price, 50% closed at exit_price if signal == 1: ret_partial = (partial_price - entry_price) / entry_price ret_remainder = (exit_price - entry_price) / entry_price else: ret_partial = -(partial_price - entry_price) / entry_price ret_remainder = -(exit_price - entry_price) / entry_price raw_ret = 0.5 * ret_partial + 0.5 * ret_remainder else: if signal == 1: raw_ret = (exit_price - entry_price) / entry_price else: raw_ret = -(exit_price - entry_price) / entry_price exit_time = path.iloc[final_bar_i]["datetime"] return { "entry_time" : entry_time, "exit_time" : exit_time, "signal" : signal, "entry_price" : entry_price, "exit_price" : exit_price, "stop_hit" : stop_hit, "exit_reason" : exit_reason, "partial_closed": partial_closed, "mae" : full_mae, "mfe" : full_mfe, "return" : raw_ret - spread_cost, } # ── Run one configuration ───────────────────────────────────────────────────── def run_config( name : str, signals : pd.DataFrame, m15 : pd.DataFrame, signal_df : pd.DataFrame, hold_hours : int, tp : float | None, z_threshold : float, two_stage : bool = False, ) -> pd.DataFrame: print(f" {name}...") trades = [] last_exit_time = None for _, sig in signals.iterrows(): if last_exit_time is not None and sig["datetime"] < last_exit_time: continue entry = find_entry_pullback( m15=m15, signal_row=sig, fib=FIB, wait_hours=6 ) if entry is None: continue trade = simulate_trade_with_tp( m15 =m15, signal_df =signal_df, entry_time =entry["entry_time"], entry_price =entry["entry_price"], signal =int(sig["signal"]), hold_hours =hold_hours, tp =tp, z_threshold =z_threshold, two_stage =two_stage, ) if trade is None: continue trade["zscore_abs"] = abs(float(sig["lag_zscore_24h_v3"])) trades.append(trade) last_exit_time = trade["exit_time"] if not trades: return pd.DataFrame() df = pd.DataFrame(trades) df["equity"] = (1 + df["return"]).cumprod() df["peak"] = df["equity"].cummax() df["drawdown"] = df["equity"] / df["peak"] - 1 df["win"] = (df["return"] > 0).astype(int) streak = max_streak = 0 for r in df["return"]: if r <= 0: streak += 1 max_streak = max(max_streak, streak) else: streak = 0 df["max_streak_val"] = max_streak return df # ── FTMO Monte Carlo ────────────────────────────────────────────────────────── def run_mc(df: pd.DataFrame) -> dict: rng = np.random.default_rng(RANDOM_SEED) returns = df["return"].values zscores = df["zscore_abs"].values exits = df["exit_time"].values per_yr = len(df) / YEARS 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 dpnl = {} outcome = "INCOMPLETE" ttrades = 0 for ret, mult, ex_dt in zip(sh_ret, sh_mult, exits): dk = str(pd.Timestamp(ex_dt).date()) if dk not in dpnl: dpnl[dk] = 0.0 notional = min(BASE_NOTIONAL * mult, BASE_NOTIONAL * 3) pnl = ret * notional balance += pnl ttrades += 1 dpnl[dk] += pnl if balance > peak: peak = balance dd = (peak - balance) / ACCOUNT_START if dd > max_dd: max_dd = dd if dpnl[dk] < -MAX_DAILY: outcome = "BREACH_DAILY"; break if balance <= ACCOUNT_START - MAX_OVERALL: outcome = "BREACH_OVERALL"; break if balance >= ACCOUNT_START + PROFIT_TARGET_MC: outcome = "PASS"; break if outcome == "INCOMPLETE": outcome = ("PASS" if balance >= ACCOUNT_START + PROFIT_TARGET_MC else "BREACH_OVERALL" if (peak - balance) / ACCOUNT_START >= 0.10 else "INCOMPLETE") results.append({"outcome": outcome, "max_dd": max_dd*100, "trades": ttrades}) 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), "dd_50" : mc["max_dd"].quantile(0.50), "avg_months" : (passing["trades"].mean() / per_yr * 12 if len(passing) > 0 else np.nan), "med_months" : (passing["trades"].median() / per_yr * 12 if len(passing) > 0 else np.nan), "p25_months" : (passing["trades"].quantile(0.25) / per_yr * 12 if len(passing) > 0 else np.nan), } # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 70) print("TAKE PROFIT EXIT V1 — TP + Z-Exit Optimisation") print(f" Z-exit threshold: {ZSCORE_REVERSAL} | Stop: {STOP:.2%}") print(f" Signal scaling: ON (1x/1.5x/2x)") print("=" * 70) print("\nLoading data...") m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) signal_df = get_model_ready_spot_lag_v3().copy() signal_df["datetime"] = pd.to_datetime(signal_df["datetime"]) signal_df = signal_df.sort_values("datetime").reset_index(drop=True) signals = build_frozen_signals( threshold=THRESHOLD, allowed_hours=ALLOWED_HOURS ) print(f" 15M bars: {len(m15):,} | Signals: {len(signals)}") # ── Define configurations ───────────────────────────────────────────────── configs = [ # (name, hold_hours, tp, two_stage) ("A) z=1.5, no TP, 52h [BASELINE]", 52, None, False), ("B) z=1.5 + TP 0.30%, 52h", 52, 0.0030, False), ("C) z=1.5 + TP 0.40%, 52h", 52, 0.0040, False), ("D) z=1.5 + TP 0.50%, 52h", 52, 0.0050, False), ("E) z=1.5 + TP 0.60%, 52h", 52, 0.0060, False), ("F) z=1.5 + TP 0.50%, 36h", 36, 0.0050, False), ("G) z=1.5 + TP 0.50%, 24h", 24, 0.0050, False), ("H) z=1.5 + TP 0.50%, two-stage", 52, 0.0050, True), ] print(f"\nRunning {len(configs)} configurations...\n") rows = [] all_dfs = {} for name, hold_hours, tp, two_stage in configs: df = run_config( name =name, signals =signals, m15 =m15, signal_df =signal_df, hold_hours =hold_hours, tp =tp, z_threshold=ZSCORE_REVERSAL, two_stage =two_stage, ) if df.empty: print(f" [WARNING] No trades for {name}") continue mc = run_mc(df) n = len(df) wr = df["win"].mean() * 100 ar = df["return"].mean() fe = df["equity"].iloc[-1] mdd = df["drawdown"].min() * 100 streak = int(df["max_streak_val"].iloc[-1]) er = df["exit_reason"].value_counts(normalize=True) * 100 stop_pct = er.get("stop", 0) z_pct = er.get("zscore_reversal", 0) tp_pct = er.get("tp", 0) time_pct = er.get("time", 0) print(f" Trades={n:,} WR={wr:.1f}% AvgRet={ar:.6f} " f"FinalEq={fe:.4f} MaxDD={mdd:.2f}%") print(f" Exits: stop={stop_pct:.1f}% z={z_pct:.1f}% " f"tp={tp_pct:.1f}% time={time_pct:.1f}%") print(f" MC: pass={mc['pass_rate']:.2f}% " f"DD95={mc['dd_95']:.3f}% " f"p25={mc['p25_months']:.1f}m " f"med={mc['med_months']:.1f}m\n") rows.append({ "name" : name, "hold" : hold_hours, "tp" : tp, "two_stage" : two_stage, "trades" : n, "per_yr" : n / YEARS, "win_rate" : wr, "avg_return": ar, "final_eq" : fe, "max_dd" : mdd, "streak" : streak, "stop_pct" : stop_pct, "z_pct" : z_pct, "tp_pct" : tp_pct, "time_pct" : time_pct, "pass_rate" : mc["pass_rate"], "dd_95" : mc["dd_95"], "dd_50" : mc["dd_50"], "p25_months": mc["p25_months"], "med_months": mc["med_months"], "avg_months": mc["avg_months"], }) all_dfs[name] = (df, mc) # ── Results table ───────────────────────────────────────────────────────── print(f"\n{'='*80}") print("FULL RESULTS TABLE") print(f"{'='*80}") print(f"\n {'Config':<40}{'WR%':>6}{'AvgRet':>10}{'FinalEq':>9}" f"{'MaxDD':>7}{'Pass%':>7}{'DD95':>7}{'P25m':>6}{'Medm':>6}") print(f" {'─'*98}") best_score = -999 best_name = "" for r in rows: valid = r["dd_95"] < 10.0 marker = "" # Score: pass_rate weight + avg_return weight + DD safety score = r["pass_rate"] + r["avg_return"] * 5000 - r["dd_95"] * 2 if valid and score > best_score: best_score = score best_name = r["name"] marker = " ◄" short = r["name"][:38] print(f" {short:<40}{r['win_rate']:>6.1f}" f"{r['avg_return']:>10.6f}{r['final_eq']:>9.4f}" f"{r['max_dd']:>7.2f}{r['pass_rate']:>7.2f}" f"{r['dd_95']:>7.3f}{r['p25_months']:>6.1f}" f"{r['med_months']:>6.1f}{'✓' if valid else '✗'}{marker}") # ── Best configuration detail ───────────────────────────────────────────── print(f"\n{'='*80}") print("BEST CONFIGURATION DETAIL") print(f"{'='*80}") if best_name and best_name in all_dfs: best_df, best_mc = all_dfs[best_name] baseline_df, baseline_mc = list(all_dfs.values())[0] wr_gain = best_df["win"].mean()*100 - baseline_df["win"].mean()*100 ret_gain = best_df["return"].mean() - baseline_df["return"].mean() dd_change = best_mc["dd_95"] - baseline_mc["dd_95"] mths_saved = baseline_mc["med_months"] - best_mc["med_months"] print(f""" Best: {best_name} vs z=1.5 baseline (no TP): Win rate change : {wr_gain:+.2f} pp Avg return change : {ret_gain:+.6f} Final equity change : {best_df['equity'].iloc[-1]-baseline_df['equity'].iloc[-1]:+.4f} 95th pct DD change : {dd_change:+.3f}% Median months saved : {mths_saved:+.1f} months P25 months (best 25%): {best_mc['p25_months']:.1f} months Exit breakdown: Stop hits : {best_df['exit_reason'].value_counts(normalize=True).get('stop', 0)*100:.1f}% Z-exit : {best_df['exit_reason'].value_counts(normalize=True).get('zscore_reversal', 0)*100:.1f}% TP hits : {best_df['exit_reason'].value_counts(normalize=True).get('tp', 0)*100:.1f}% Time exit : {best_df['exit_reason'].value_counts(normalize=True).get('time', 0)*100:.1f}% """) # ── Save results ────────────────────────────────────────────────────────── result_df = pd.DataFrame(rows) out_path = TRADES_DIR / "tp_exit_results.csv" result_df.to_csv(out_path, index=False) print(f" Results saved: {out_path}") print(f"\n Next step: if a valid TP config improves avg_months,") print(f" add AUDUSD with the same exit framework and run") print(f" the 2-pair portfolio FTMO simulation.") if __name__ == "__main__": main()