""" Backtest SMC-Only + Pullback Filter ONLY ========================================= Base: backtest_smc_only.py (100% synced with main_live.py v4) IMPROVEMENT #3 — Pullback Filter (re-enable from main_live.py): - Block entry when short-term momentum opposes signal direction - Uses 3 confirmations: 3-candle momentum, MACD histogram, price vs EMA9 - ATR-based dynamic thresholds (not hardcoded) - NO sell filter — pure pullback filter test Usage: python backtests/backtest_pullback_only.py """ import polars as pl import pandas as pd import numpy as np from datetime import datetime, timedelta, date from typing import Dict, List, Tuple, Optional from dataclasses import dataclass, field from enum import Enum import sys import os from zoneinfo import ZoneInfo from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill, Border, Side from openpyxl.chart import LineChart, Reference from openpyxl.utils import get_column_letter sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.mt5_connector import MT5Connector from src.smc_polars import SMCAnalyzer, SMCSignal from src.feature_eng import FeatureEngineer from src.regime_detector import MarketRegimeDetector, MarketRegime from src.ml_model import TradingModel from src.config import get_config from src.dynamic_confidence import DynamicConfidenceManager, create_dynamic_confidence, MarketQuality from loguru import logger logger.remove() logger.add(sys.stderr, level="WARNING") WIB = ZoneInfo("Asia/Jakarta") # ─── Enums & Dataclasses ────────────────────────────────────── class TradeResult(Enum): WIN = "WIN" LOSS = "LOSS" BREAKEVEN = "BREAKEVEN" class ExitReason(Enum): TAKE_PROFIT = "take_profit" SMART_TP = "smart_tp" PEAK_PROTECT = "peak_protect" EARLY_EXIT = "early_exit" EARLY_CUT = "early_cut" MAX_LOSS = "max_loss" STALL = "stall" TREND_REVERSAL = "trend_reversal" TIMEOUT = "timeout" WEEKEND_CLOSE = "weekend_close" TRAILING_SL = "trailing_sl" BREAKEVEN_EXIT = "breakeven_exit" DAILY_LIMIT = "daily_limit" REGIME_DANGER = "regime_danger" MARKET_SIGNAL = "market_signal" class TradingMode(Enum): NORMAL = "normal" RECOVERY = "recovery" PROTECTED = "protected" STOPPED = "stopped" @dataclass class SimulatedTrade: ticket: int entry_time: datetime exit_time: datetime direction: str entry_price: float exit_price: float stop_loss: float take_profit: float lot_size: float profit_usd: float profit_pips: float result: TradeResult exit_reason: ExitReason smc_confidence: float regime: str session: str signal_reason: str has_bos: bool = False has_choch: bool = False has_fvg: bool = False has_ob: bool = False atr_at_entry: float = 0.0 rr_ratio: float = 0.0 trading_mode: str = "normal" @dataclass class BacktestStats: total_trades: int = 0 wins: int = 0 losses: int = 0 total_profit: float = 0.0 total_loss: float = 0.0 max_drawdown: float = 0.0 max_drawdown_usd: float = 0.0 win_rate: float = 0.0 profit_factor: float = 0.0 avg_win: float = 0.0 avg_loss: float = 0.0 avg_trade: float = 0.0 expectancy: float = 0.0 sharpe_ratio: float = 0.0 trades: List[SimulatedTrade] = field(default_factory=list) equity_curve: List[float] = field(default_factory=list) avoided_signals: int = 0 daily_limit_stops: int = 0 recovery_mode_trades: int = 0 pullback_filtered: int = 0 pullback_filtered_buy: int = 0 pullback_filtered_sell: int = 0 # ─── SMC-Only + Pullback Filter Only ──────────────────────── class SMCOnlyPullbackOnly: """ Base: 100% synced with main_live.py Signal Logic v4 + all exit systems. ADDED: Pullback Filter only (no sell filter). """ def __init__( self, capital: float = 5000.0, max_daily_loss_percent: float = 5.0, max_loss_per_trade_percent: float = 1.0, base_lot_size: float = 0.01, max_lot_size: float = 0.02, recovery_lot_size: float = 0.01, trend_reversal_threshold: float = 0.75, max_concurrent_positions: int = 2, breakeven_pips: float = 30.0, trail_start_pips: float = 50.0, trail_step_pips: float = 30.0, min_profit_to_protect: float = 5.0, max_drawdown_from_peak: float = 50.0, trade_cooldown_bars: int = 10, trend_reversal_mult: float = 0.6, ): self.capital = capital self.max_daily_loss_usd = capital * (max_daily_loss_percent / 100) self.max_loss_per_trade = capital * (max_loss_per_trade_percent / 100) self.base_lot_size = base_lot_size self.max_lot_size = max_lot_size self.recovery_lot_size = recovery_lot_size self.trend_reversal_threshold = trend_reversal_threshold self.max_concurrent_positions = max_concurrent_positions self.breakeven_pips = breakeven_pips self.trail_start_pips = trail_start_pips self.trail_step_pips = trail_step_pips self.min_profit_to_protect = min_profit_to_protect self.max_drawdown_from_peak = max_drawdown_from_peak self.trade_cooldown_bars = trade_cooldown_bars self.trend_reversal_mult = trend_reversal_mult config = get_config() self.smc = SMCAnalyzer( swing_length=config.smc.swing_length, ob_lookback=config.smc.ob_lookback, ) self.features = FeatureEngineer() self.dynamic_confidence = create_dynamic_confidence() self.ml_model = TradingModel(model_path="models/xgboost_model.pkl") try: self.ml_model.load() print(" ML model loaded (for exit evaluation)") except Exception: print(" [WARN] ML model not loaded — exit ML checks disabled") self.regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl") try: self.regime_detector.load() except Exception: print(" [WARN] HMM model not loaded") self._ticket_counter = 5000000 # ── Session filter (synced) ── def _get_session_from_time(self, dt: datetime) -> Tuple[str, bool, float]: if dt.tzinfo is None: dt = dt.replace(tzinfo=ZoneInfo("UTC")) wib_time = dt.astimezone(WIB) hour = wib_time.hour if 6 <= hour < 15: return "Sydney-Tokyo", True, 0.5 elif 15 <= hour < 16: return "Tokyo-London Overlap", True, 0.75 elif 16 <= hour < 19: return "London Early", True, 0.8 elif 19 <= hour < 24: return "London-NY Overlap (Golden)", True, 1.0 elif 0 <= hour < 4: return "NY Session", True, 0.9 else: return "Off Hours", False, 0.0 def _hours_to_golden(self, dt: datetime) -> float: if dt.tzinfo is None: dt = dt.replace(tzinfo=ZoneInfo("UTC")) wib = dt.astimezone(WIB) if 19 <= wib.hour < 24: return 0 target = wib.replace(hour=19, minute=0, second=0, microsecond=0) if wib.hour >= 19: target += timedelta(days=1) return max(0, (target - wib).total_seconds() / 3600) def _is_near_weekend_close(self, dt: datetime) -> bool: if dt.tzinfo is None: dt = dt.replace(tzinfo=ZoneInfo("UTC")) wib = dt.astimezone(WIB) if wib.weekday() == 5 and wib.hour >= 4 and wib.minute >= 30: return True return False # ══════════════════════════════════════════════════════════════ # Pullback Filter (100% synced from main_live.py) # ══════════════════════════════════════════════════════════════ def _check_pullback_filter( self, df_slice: pl.DataFrame, signal_direction: str, current_price: float, ) -> Tuple[bool, str]: """100% synced with main_live.py._check_pullback_filter()""" recent = df_slice.tail(10) if len(recent) < 5: return True, "Not enough data" atr = 12.0 if "atr" in df_slice.columns: atr_val = recent["atr"].to_list()[-1] if atr_val is not None and atr_val > 0: atr = atr_val bounce_threshold = atr * 0.15 consolidation_threshold = atr * 0.10 closes = recent["close"].to_list() last_3_closes = closes[-3:] short_momentum = last_3_closes[-1] - last_3_closes[0] momentum_direction = "UP" if short_momentum > 0 else "DOWN" macd_hist_direction = "NEUTRAL" if "macd_histogram" in df_slice.columns: macd_hist = recent["macd_histogram"].to_list() last_hist = macd_hist[-1] if macd_hist[-1] is not None else 0 prev_hist = macd_hist[-2] if macd_hist[-2] is not None else 0 if last_hist > prev_hist: macd_hist_direction = "RISING" else: macd_hist_direction = "FALLING" price_vs_ema = "NEUTRAL" if "ema_9" in df_slice.columns: ema_9 = recent["ema_9"].to_list()[-1] if ema_9 is not None: if current_price > ema_9 * 1.001: price_vs_ema = "ABOVE" elif current_price < ema_9 * 0.999: price_vs_ema = "BELOW" if signal_direction == "SELL": if momentum_direction == "UP" and short_momentum > bounce_threshold: return False, f"SELL blocked: Price bouncing UP (+${short_momentum:.2f})" if macd_hist_direction == "RISING" and momentum_direction == "UP": return False, "SELL blocked: MACD bullish + price rising" if price_vs_ema == "ABOVE" and momentum_direction == "UP": return False, "SELL blocked: Price above EMA9 and rising" if momentum_direction == "DOWN": return True, "SELL OK: Momentum aligned" if abs(short_momentum) < consolidation_threshold: return True, "SELL OK: Consolidation" elif signal_direction == "BUY": if momentum_direction == "DOWN" and short_momentum < -bounce_threshold: return False, f"BUY blocked: Price falling DOWN (${short_momentum:.2f})" if macd_hist_direction == "FALLING" and momentum_direction == "DOWN": return False, "BUY blocked: MACD bearish + price falling" if price_vs_ema == "BELOW" and momentum_direction == "DOWN": return False, "BUY blocked: Price below EMA9 and falling" if momentum_direction == "UP": return True, "BUY OK: Momentum aligned" if abs(short_momentum) < consolidation_threshold: return True, "BUY OK: Consolidation" return True, "Pullback check passed" # ── Lot sizing (synced) ── def _calculate_lot_size(self, confidence, regime, trading_mode, session_mult): if trading_mode == TradingMode.STOPPED: return 0 lot = self.base_lot_size if trading_mode in (TradingMode.RECOVERY, TradingMode.PROTECTED): lot = self.recovery_lot_size else: if confidence >= 0.65: lot = self.max_lot_size elif confidence >= 0.55: lot = self.base_lot_size else: lot = self.recovery_lot_size if regime.lower() in ["high_volatility", "crisis"]: lot = self.recovery_lot_size lot = max(0.01, lot * session_mult) return round(lot, 2) # ── Full exit simulation (UNCHANGED from baseline) ── def _simulate_trade_exit(self, df, entry_idx, direction, entry_price, take_profit, stop_loss, lot_size, daily_loss_so_far, feature_cols, max_bars=100): pip_value = 10 highs = df["high"].to_list() lows = df["low"].to_list() closes = df["close"].to_list() times = df["time"].to_list() atr = 12.0 if "atr" in df.columns: atr_list = df["atr"].to_list() if entry_idx < len(atr_list) and atr_list[entry_idx] is not None: atr = atr_list[entry_idx] reversal_momentum_threshold = atr * self.trend_reversal_mult min_loss_for_reversal_exit = atr * 0.8 profit_history = [] peak_profit = 0.0 stall_count = 0 reversal_warnings = 0 current_sl = stop_loss breakeven_moved = False if direction == "BUY": target_tp_profit = (take_profit - entry_price) / 0.1 * pip_value * lot_size else: target_tp_profit = (entry_price - take_profit) / 0.1 * pip_value * lot_size cached_ml_signal = "" cached_ml_confidence = 0.5 for i in range(entry_idx + 1, min(entry_idx + max_bars, len(df))): high = highs[i] low = lows[i] close = closes[i] current_time = times[i] if direction == "BUY": current_pips = (close - entry_price) / 0.1 pip_profit_from_entry = current_pips else: current_pips = (entry_price - close) / 0.1 pip_profit_from_entry = current_pips current_profit = current_pips * pip_value * lot_size profit_history.append(current_profit) if current_profit > peak_profit: peak_profit = current_profit bars_since_entry = i - entry_idx if bars_since_entry % 4 == 0 and self.ml_model.fitted: try: df_slice = df.head(i + 1) ml_pred = self.ml_model.predict(df_slice, feature_cols) cached_ml_signal = ml_pred.signal cached_ml_confidence = ml_pred.confidence except Exception: pass momentum = 0.0 if len(profit_history) >= 3: recent = profit_history[-5:] if len(profit_history) >= 5 else profit_history profit_change = recent[-1] - recent[0] momentum = max(-100, min(100, (profit_change / 10) * 50)) profit_growing = momentum > 0 # A) SmartPositionManager if direction == "BUY" and high >= take_profit: pips = (take_profit - entry_price) / 0.1 return pips * pip_value * lot_size, pips, ExitReason.TAKE_PROFIT, i, take_profit elif direction == "SELL" and low <= take_profit: pips = (entry_price - take_profit) / 0.1 return pips * pip_value * lot_size, pips, ExitReason.TAKE_PROFIT, i, take_profit if breakeven_moved and current_sl > 0: if direction == "BUY" and low <= current_sl: pips = (current_sl - entry_price) / 0.1 reason = ExitReason.TRAILING_SL if pip_profit_from_entry >= self.trail_start_pips else ExitReason.BREAKEVEN_EXIT return pips * pip_value * lot_size, pips, reason, i, current_sl elif direction == "SELL" and high >= current_sl: pips = (entry_price - current_sl) / 0.1 reason = ExitReason.TRAILING_SL if pip_profit_from_entry >= self.trail_start_pips else ExitReason.BREAKEVEN_EXIT return pips * pip_value * lot_size, pips, reason, i, current_sl if pip_profit_from_entry >= self.breakeven_pips and not breakeven_moved: if direction == "BUY": current_sl = entry_price + 2 else: current_sl = entry_price - 2 breakeven_moved = True if pip_profit_from_entry >= self.trail_start_pips: trail_distance = self.trail_step_pips * 0.1 if direction == "BUY": new_trail_sl = close - trail_distance if new_trail_sl > current_sl: current_sl = new_trail_sl else: new_trail_sl = close + trail_distance if current_sl == 0 or new_trail_sl < current_sl: current_sl = new_trail_sl if peak_profit > self.min_profit_to_protect: drawdown_pct = ((peak_profit - current_profit) / peak_profit) * 100 if peak_profit > 0 else 0 if drawdown_pct > self.max_drawdown_from_peak: return current_profit, current_pips, ExitReason.PEAK_PROTECT, i, close if bars_since_entry % 5 == 0 and bars_since_entry >= 5: if i >= 20: ma_fast = np.mean(closes[i-4:i+1]) ma_slow = np.mean(closes[i-19:i+1]) trend = "NEUTRAL" if ma_fast > ma_slow * 1.001: trend = "BULLISH" elif ma_fast < ma_slow * 0.999: trend = "BEARISH" roc = (closes[i] / closes[max(0,i-4)] - 1) * 100 mom_dir = "BULLISH" if roc > 0.3 else ("BEARISH" if roc < -0.3 else "NEUTRAL") rsi_val = None if "rsi" in df.columns: rsi_list = df["rsi"].to_list() if i < len(rsi_list): rsi_val = rsi_list[i] urgency = 0 should_exit = False if cached_ml_confidence > 0.75: if direction == "BUY" and cached_ml_signal == "SELL": should_exit = True; urgency += 2 elif direction == "SELL" and cached_ml_signal == "BUY": should_exit = True; urgency += 2 if rsi_val: if rsi_val > 75 and direction == "BUY": should_exit = True; urgency += 2 elif rsi_val < 25 and direction == "SELL": should_exit = True; urgency += 2 if direction == "BUY" and trend == "BEARISH" and mom_dir == "BEARISH": should_exit = True; urgency += 3 elif direction == "SELL" and trend == "BULLISH" and mom_dir == "BULLISH": should_exit = True; urgency += 3 if should_exit and current_profit > self.min_profit_to_protect / 2: return current_profit, current_pips, ExitReason.MARKET_SIGNAL, i, close if urgency >= 7 and current_profit > 0: return current_profit, current_pips, ExitReason.MARKET_SIGNAL, i, close if self._is_near_weekend_close(current_time): if current_profit > 0: return current_profit, current_pips, ExitReason.WEEKEND_CLOSE, i, close elif current_profit > -10: return current_profit, current_pips, ExitReason.WEEKEND_CLOSE, i, close # B) SmartRiskManager if current_profit >= 15: if current_profit >= 40: return current_profit, current_pips, ExitReason.SMART_TP, i, close if current_profit >= 25 and momentum < -30: return current_profit, current_pips, ExitReason.SMART_TP, i, close if peak_profit > 30 and current_profit < peak_profit * 0.6: return current_profit, current_pips, ExitReason.PEAK_PROTECT, i, close if current_profit >= 20: progress = (current_profit / target_tp_profit) * 100 if target_tp_profit > 0 else 0 progress_score = min(40, max(0, progress * 0.4)) momentum_score = ((momentum + 100) / 200) * 30 time_penalty = min(10, bars_since_entry / 4 * 2) tp_probability = progress_score + momentum_score + 10 - time_penalty if tp_probability < 25: return current_profit, current_pips, ExitReason.SMART_TP, i, close if 5 <= current_profit < 15: if momentum < -50 and cached_ml_confidence >= 0.65: is_reversal = ( (direction == "BUY" and cached_ml_signal == "SELL") or (direction == "SELL" and cached_ml_signal == "BUY") ) if is_reversal: return current_profit, current_pips, ExitReason.EARLY_EXIT, i, close if current_profit < 0: loss_percent_of_max = abs(current_profit) / self.max_loss_per_trade * 100 if momentum < -30 and loss_percent_of_max >= 30: return current_profit, current_pips, ExitReason.EARLY_CUT, i, close is_ml_reversal = False if direction == "BUY" and cached_ml_signal == "SELL" and cached_ml_confidence >= self.trend_reversal_threshold: is_ml_reversal = True; reversal_warnings += 1 elif direction == "SELL" and cached_ml_signal == "BUY" and cached_ml_confidence >= self.trend_reversal_threshold: is_ml_reversal = True; reversal_warnings += 1 loss_moderate = abs(current_profit) > (self.max_loss_per_trade * 0.4) if is_ml_reversal and current_profit < -8 and loss_moderate: return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close if reversal_warnings >= 3 and current_profit < -10: return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close if current_profit <= -(self.max_loss_per_trade * 0.50): htg = self._hours_to_golden(current_time) if htg <= 1 and htg > 0 and momentum > -40: pass else: return current_profit, current_pips, ExitReason.MAX_LOSS, i, close if len(profit_history) >= 10: recent_range = max(profit_history[-10:]) - min(profit_history[-10:]) if recent_range < 3 and current_profit < -15: stall_count += 1 if stall_count >= 5: return current_profit, current_pips, ExitReason.STALL, i, close potential_daily_loss = daily_loss_so_far + abs(min(0, current_profit)) if potential_daily_loss >= self.max_daily_loss_usd: return current_profit, current_pips, ExitReason.DAILY_LIMIT, i, close # C) Time-based exit if bars_since_entry >= 16: if current_profit < 5 and not profit_growing: if current_profit >= 0: return current_profit, current_pips, ExitReason.TIMEOUT, i, close elif current_profit > -15: return current_profit, current_pips, ExitReason.TIMEOUT, i, close if bars_since_entry >= 24: if current_profit < 10 or not profit_growing: return current_profit, current_pips, ExitReason.TIMEOUT, i, close if bars_since_entry >= 32: return current_profit, current_pips, ExitReason.TIMEOUT, i, close if bars_since_entry > 10: recent_closes = closes[i-5:i+1] mom = recent_closes[-1] - recent_closes[0] if direction == "BUY" and mom < -reversal_momentum_threshold: if current_profit < -min_loss_for_reversal_exit: return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close elif direction == "SELL" and mom > reversal_momentum_threshold: if current_profit < -min_loss_for_reversal_exit: return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close final_idx = min(entry_idx + max_bars - 1, len(df) - 1) final_price = closes[final_idx] pips = (final_price - entry_price) / 0.1 if direction == "BUY" else (entry_price - final_price) / 0.1 return pips * pip_value * lot_size, pips, ExitReason.TIMEOUT, final_idx, final_price # ── Main backtest run ── def run(self, df, start_date=None, end_date=None, initial_capital=5000.0): stats = BacktestStats() capital = initial_capital peak_capital = initial_capital stats.equity_curve.append(capital) daily_loss = 0.0 daily_profit = 0.0 daily_trades = 0 consecutive_losses = 0 trading_mode = TradingMode.NORMAL current_date = None feature_cols = [] if self.ml_model.fitted and self.ml_model.feature_names: feature_cols = [f for f in self.ml_model.feature_names if f in df.columns] times = df["time"].to_list() start_idx = next((i for i, t in enumerate(times) if t >= start_date), 100) if start_date else 100 end_idx = next((i for i, t in enumerate(times) if t > end_date), len(df) - 100) if end_date else len(df) - 100 last_trade_idx = -self.trade_cooldown_bars * 2 print(f"\n Running SMC-Only + Pullback Filter ONLY backtest...") print(f" Date range: {times[start_idx]} to {times[end_idx - 1]}") print(f" Total bars: {end_idx - start_idx}") for i in range(start_idx, end_idx): if i - last_trade_idx < self.trade_cooldown_bars: continue current_time = times[i] trade_date = current_time.date() if hasattr(current_time, 'date') else current_time if current_date is None or trade_date != current_date: daily_loss = 0.0 daily_profit = 0.0 daily_trades = 0 current_date = trade_date if consecutive_losses < 2: trading_mode = TradingMode.NORMAL if trading_mode == TradingMode.STOPPED: continue session_name, can_trade, lot_mult = self._get_session_from_time(current_time) if not can_trade: continue if hasattr(current_time, 'weekday') and current_time.weekday() >= 5: continue df_slice = df.head(i + 1) regime = "normal" regime_state = None try: if self.regime_detector.fitted: regime_state = self.regime_detector.get_current_state(df_slice) if regime_state: regime = regime_state.regime.value if regime_state.regime == MarketRegime.CRISIS: continue if regime_state.recommendation == "SLEEP": continue except Exception: pass ml_signal = "" ml_confidence = 0.5 try: if self.ml_model.fitted and feature_cols: ml_pred = self.ml_model.predict(df_slice, feature_cols) ml_signal = ml_pred.signal ml_confidence = ml_pred.confidence market_analysis = self.dynamic_confidence.analyze_market( session=session_name, regime=regime, volatility="medium", trend_direction=regime, has_smc_signal=True, ml_signal=ml_signal, ml_confidence=ml_confidence, ) if market_analysis.quality == MarketQuality.AVOID: stats.avoided_signals += 1 continue except Exception: pass try: smc_signal = self.smc.generate_signal(df_slice) except Exception: continue if smc_signal is None: continue # ══════════════════════════════════════════════════════════ # PULLBACK FILTER (the only improvement — no sell filter) # ══════════════════════════════════════════════════════════ current_price = df_slice.tail(1)["close"].item() can_trade_pb, pb_reason = self._check_pullback_filter( df_slice, smc_signal.signal_type, current_price ) if not can_trade_pb: stats.pullback_filtered += 1 if smc_signal.signal_type == "BUY": stats.pullback_filtered_buy += 1 else: stats.pullback_filtered_sell += 1 continue # SMC details recent_df = df_slice.tail(10) recent_bos = recent_df["bos"].to_list() if "bos" in df_slice.columns else [] recent_choch = recent_df["choch"].to_list() if "choch" in df_slice.columns else [] recent_fvg_bull = recent_df["is_fvg_bull"].to_list() if "is_fvg_bull" in df_slice.columns else [] recent_fvg_bear = recent_df["is_fvg_bear"].to_list() if "is_fvg_bear" in df_slice.columns else [] recent_obs = recent_df["ob"].to_list() if "ob" in df_slice.columns else [] has_bos = 1 in recent_bos or -1 in recent_bos has_choch = 1 in recent_choch or -1 in recent_choch has_fvg = any(recent_fvg_bull) or any(recent_fvg_bear) has_ob = 1 in recent_obs or -1 in recent_obs atr_at_entry = 12.0 if "atr" in df_slice.columns: atr_val = df_slice.tail(1)["atr"].item() if atr_val is not None and atr_val > 0: atr_at_entry = atr_val confidence = smc_signal.confidence ml_agrees = ( (smc_signal.signal_type == "BUY" and ml_signal == "BUY") or (smc_signal.signal_type == "SELL" and ml_signal == "SELL") ) if ml_agrees: confidence = (smc_signal.confidence + ml_confidence) / 2 if regime == "high_volatility": confidence *= 0.9 lot_size = self._calculate_lot_size(confidence, regime, trading_mode, lot_mult) if lot_size <= 0: continue if trading_mode == TradingMode.RECOVERY: stats.recovery_mode_trades += 1 entry_price = smc_signal.entry_price take_profit_price = smc_signal.take_profit stop_loss_price = smc_signal.stop_loss risk = abs(entry_price - stop_loss_price) rr = abs(take_profit_price - entry_price) / risk if risk > 0 else 0 profit, pips, exit_reason, exit_idx, exit_price = self._simulate_trade_exit( df=df, entry_idx=i, direction=smc_signal.signal_type, entry_price=entry_price, take_profit=take_profit_price, stop_loss=stop_loss_price, lot_size=lot_size, daily_loss_so_far=daily_loss, feature_cols=feature_cols, ) self._ticket_counter += 1 result = TradeResult.WIN if profit > 0 else (TradeResult.LOSS if profit < 0 else TradeResult.BREAKEVEN) trade = SimulatedTrade( ticket=self._ticket_counter, entry_time=current_time, exit_time=times[exit_idx] if exit_idx < len(times) else times[-1], direction=smc_signal.signal_type, entry_price=entry_price, exit_price=exit_price, stop_loss=stop_loss_price, take_profit=take_profit_price, lot_size=lot_size, profit_usd=profit, profit_pips=pips, result=result, exit_reason=exit_reason, smc_confidence=confidence, regime=regime, session=session_name, signal_reason=smc_signal.reason, has_bos=has_bos, has_choch=has_choch, has_fvg=has_fvg, has_ob=has_ob, atr_at_entry=atr_at_entry, rr_ratio=rr, trading_mode=trading_mode.value, ) stats.trades.append(trade) stats.total_trades += 1 daily_trades += 1 capital += profit if profit > 0: stats.wins += 1 stats.total_profit += profit daily_profit += profit consecutive_losses = 0 if trading_mode == TradingMode.RECOVERY: trading_mode = TradingMode.NORMAL else: stats.losses += 1 stats.total_loss += abs(profit) daily_loss += abs(profit) consecutive_losses += 1 if daily_loss >= self.max_daily_loss_usd: trading_mode = TradingMode.STOPPED stats.daily_limit_stops += 1 elif consecutive_losses >= 3 or daily_loss >= self.max_daily_loss_usd * 0.6: trading_mode = TradingMode.PROTECTED elif consecutive_losses >= 2: trading_mode = TradingMode.RECOVERY if capital > peak_capital: peak_capital = capital drawdown_pct = (peak_capital - capital) / peak_capital * 100 drawdown_usd = peak_capital - capital if drawdown_pct > stats.max_drawdown: stats.max_drawdown = drawdown_pct stats.max_drawdown_usd = drawdown_usd stats.equity_curve.append(capital) last_trade_idx = exit_idx if stats.total_trades % 100 == 0: print(f" {stats.total_trades} trades processed...") if stats.total_trades > 0: stats.win_rate = stats.wins / stats.total_trades * 100 stats.avg_win = stats.total_profit / stats.wins if stats.wins > 0 else 0 stats.avg_loss = stats.total_loss / stats.losses if stats.losses > 0 else 0 stats.avg_trade = (stats.total_profit - stats.total_loss) / stats.total_trades stats.profit_factor = stats.total_profit / stats.total_loss if stats.total_loss > 0 else float("inf") win_prob = stats.wins / stats.total_trades loss_prob = stats.losses / stats.total_trades stats.expectancy = (win_prob * stats.avg_win) - (loss_prob * stats.avg_loss) returns = [t.profit_usd for t in stats.trades] if len(returns) > 1: avg_return = np.mean(returns) std_return = np.std(returns) stats.sharpe_ratio = (avg_return / std_return) * np.sqrt(252) if std_return > 0 else 0 return stats # ─── XLSX Report ─────────────────────────────────────────────── def generate_xlsx_report(stats, filepath, start_date, end_date): wb = Workbook() hf = Font(name="Calibri", bold=True, size=12, color="FFFFFF") hfill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid") sf = Font(name="Calibri", bold=True, size=10) sfill = PatternFill(start_color="D6E4F0", end_color="D6E4F0", fill_type="solid") wfill = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid") lfill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid") bdr = Border(left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin")) net_pnl = stats.total_profit - stats.total_loss ws = wb.active ws.title = "Summary" ws.sheet_properties.tabColor = "1F4E79" ws.merge_cells("A1:F1") ws["A1"] = "XAUBot AI — Pullback Filter ONLY Backtest" ws["A1"].font = Font(name="Calibri", bold=True, size=16, color="1F4E79") ws["A2"] = f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}" ws["A2"].font = Font(name="Calibri", size=10, italic=True) ws["A3"] = f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ws["A3"].font = Font(name="Calibri", size=10, italic=True) ws["A4"] = "Change: Pullback Filter (momentum/MACD/EMA9) — NO sell filter" ws["A4"].font = Font(name="Calibri", size=10, italic=True, color="CC6600") data = [ ("Performance Metrics", "", True), ("Total Trades", stats.total_trades, False), ("Wins", stats.wins, False), ("Losses", stats.losses, False), ("Win Rate", f"{stats.win_rate:.1f}%", False), ("Pullback Filtered", stats.pullback_filtered, False), (" - BUY blocked", stats.pullback_filtered_buy, False), (" - SELL blocked", stats.pullback_filtered_sell, False), ("Avoided (AVOID)", stats.avoided_signals, False), ("Recovery Trades", stats.recovery_mode_trades, False), ("", "", False), ("Profit - Loss", "", True), ("Total Profit", f"${stats.total_profit:,.2f}", False), ("Total Loss", f"${stats.total_loss:,.2f}", False), ("Net PnL", f"${net_pnl:,.2f}", False), ("Profit Factor", f"{stats.profit_factor:.2f}", False), ("", "", False), ("Risk Metrics", "", True), ("Max Drawdown", f"{stats.max_drawdown:.1f}%", False), ("Max Drawdown ($)", f"${stats.max_drawdown_usd:,.2f}", False), ("Avg Win", f"${stats.avg_win:,.2f}", False), ("Avg Loss", f"${stats.avg_loss:,.2f}", False), ("Expectancy", f"${stats.expectancy:,.2f}", False), ("Sharpe Ratio", f"{stats.sharpe_ratio:.2f}", False), ] row = 6 for label, value, is_header in data: ws.cell(row=row, column=1, value=label) ws.cell(row=row, column=2, value=value) if is_header: ws.cell(row=row, column=1).font = sf ws.cell(row=row, column=1).fill = sfill ws.cell(row=row, column=2).fill = sfill if label == "Net PnL": ws.cell(row=row, column=2).font = Font(bold=True, color="006100" if net_pnl > 0 else "9C0006") row += 1 ws.column_dimensions["A"].width = 24 ws.column_dimensions["B"].width = 18 # Exit reasons exit_counts = {} for t in stats.trades: exit_counts[t.exit_reason.value] = exit_counts.get(t.exit_reason.value, 0) + 1 ws.cell(row=6, column=4, value="Exit Reasons").font = sf ws.cell(row=6, column=4).fill = sfill ws.cell(row=6, column=5).fill = sfill ws.cell(row=6, column=6).fill = sfill row = 7 for reason, count in sorted(exit_counts.items(), key=lambda x: -x[1]): pct = count / stats.total_trades * 100 if stats.total_trades > 0 else 0 ws.cell(row=row, column=4, value=reason) ws.cell(row=row, column=5, value=count) ws.cell(row=row, column=6, value=f"{pct:.1f}%") row += 1 # Session row += 1 ws.cell(row=row, column=4, value="Session Performance").font = sf ws.cell(row=row, column=4).fill = sfill for c in range(5, 8): ws.cell(row=row, column=c).fill = sfill row += 1 for lbl, col in [("Session", 4), ("Trades", 5), ("WR", 6), ("PnL", 7)]: ws.cell(row=row, column=col, value=lbl).font = Font(bold=True) row += 1 ss = {} for t in stats.trades: if t.session not in ss: ss[t.session] = {"w": 0, "l": 0, "p": 0.0} if t.result == TradeResult.WIN: ss[t.session]["w"] += 1 else: ss[t.session]["l"] += 1 ss[t.session]["p"] += t.profit_usd for s, d in sorted(ss.items(), key=lambda x: -x[1]["p"]): total = d["w"] + d["l"] wr = d["w"] / total * 100 if total > 0 else 0 ws.cell(row=row, column=4, value=s) ws.cell(row=row, column=5, value=total) ws.cell(row=row, column=6, value=f"{wr:.1f}%") ws.cell(row=row, column=7, value=f"${d['p']:,.2f}") ws.cell(row=row, column=7).font = Font(color="006100" if d["p"] >= 0 else "9C0006") row += 1 for c, w in {4: 28, 5: 10, 6: 12, 7: 14}.items(): ws.column_dimensions[get_column_letter(c)].width = w # Trade Log ws2 = wb.create_sheet("Trade Log") ws2.sheet_properties.tabColor = "2E75B6" headers = ["Ticket","Entry Time","Exit Time","Dir","Entry","Exit","SL","TP","Lot","Profit ($)","Pips","Result","Exit Reason","SMC Conf","Regime","Session","Signal","BOS","CHoCH","FVG","OB","ATR","RR","Mode"] for col, h in enumerate(headers, 1): cell = ws2.cell(row=1, column=col, value=h) cell.font = hf; cell.fill = hfill; cell.alignment = Alignment(horizontal="center") for ri, t in enumerate(stats.trades, 2): vals = [t.ticket, t.entry_time.strftime("%Y-%m-%d %H:%M"), t.exit_time.strftime("%Y-%m-%d %H:%M"), t.direction, t.entry_price, t.exit_price, t.stop_loss, t.take_profit, t.lot_size, round(t.profit_usd,2), round(t.profit_pips,1), t.result.value, t.exit_reason.value, round(t.smc_confidence,2), t.regime, t.session, t.signal_reason, "Y" if t.has_bos else "", "Y" if t.has_choch else "", "Y" if t.has_fvg else "", "Y" if t.has_ob else "", round(t.atr_at_entry,2), round(t.rr_ratio,2), t.trading_mode] for ci, v in enumerate(vals, 1): cell = ws2.cell(row=ri, column=ci, value=v) cell.border = bdr if ci == 10 and isinstance(v, (int,float)): cell.fill = wfill if v > 0 else (lfill if v < 0 else PatternFill()) if ci == 12: cell.fill = wfill if v == "WIN" else (lfill if v == "LOSS" else PatternFill()) for col in range(1, len(headers)+1): ws2.column_dimensions[get_column_letter(col)].width = max(11, len(headers[col-1])+3) # Equity ws3 = wb.create_sheet("Equity Curve") ws3.sheet_properties.tabColor = "548235" for c, h in enumerate(["Trade #","Equity","Drawdown ($)"], 1): ws3.cell(row=1, column=c, value=h).font = hf ws3.cell(row=1, column=c).fill = hfill pk = stats.equity_curve[0] if stats.equity_curve else 5000 for idx, eq in enumerate(stats.equity_curve): if eq > pk: pk = eq ws3.cell(row=idx+2, column=1, value=idx) ws3.cell(row=idx+2, column=2, value=round(eq,2)) ws3.cell(row=idx+2, column=3, value=round(pk-eq,2)) if len(stats.equity_curve) > 1: chart = LineChart() chart.title = "Equity Curve (Pullback Only)" chart.style = 10; chart.y_axis.title = "Equity ($)"; chart.x_axis.title = "Trade #" chart.width = 30; chart.height = 15 d = Reference(ws3, min_col=2, min_row=1, max_row=len(stats.equity_curve)+1) chart.add_data(d, titles_from_data=True) chart.series[0].graphicalProperties.line.width = 20000 ws3.add_chart(chart, "E2") # Daily PnL ws4 = wb.create_sheet("Daily PnL") ws4.sheet_properties.tabColor = "BF8F00" dpnl = {} for t in stats.trades: day = t.entry_time.strftime("%Y-%m-%d") if day not in dpnl: dpnl[day] = {"trades":0,"wins":0,"profit":0.0} dpnl[day]["trades"] += 1 if t.result == TradeResult.WIN: dpnl[day]["wins"] += 1 dpnl[day]["profit"] += t.profit_usd for c, h in enumerate(["Date","Trades","Wins","WR","Net PnL","Cumulative"], 1): ws4.cell(row=1, column=c, value=h).font = hf ws4.cell(row=1, column=c).fill = hfill cum = 0.0 for ri, (day, d) in enumerate(sorted(dpnl.items()), 2): wr = d["wins"]/d["trades"]*100 if d["trades"]>0 else 0 cum += d["profit"] ws4.cell(row=ri, column=1, value=day) ws4.cell(row=ri, column=2, value=d["trades"]) ws4.cell(row=ri, column=3, value=d["wins"]) ws4.cell(row=ri, column=4, value=f"{wr:.0f}%") ws4.cell(row=ri, column=5, value=round(d["profit"],2)) ws4.cell(row=ri, column=6, value=round(cum,2)) ws4.cell(row=ri, column=5).fill = wfill if d["profit"]>=0 else lfill for c in range(1,7): ws4.column_dimensions[get_column_letter(c)].width = 16 # vs Baseline ws5 = wb.create_sheet("vs Baseline") ws5.sheet_properties.tabColor = "FF6600" ws5["A1"] = "Comparison: Pullback Only vs Baseline" ws5["A1"].font = Font(name="Calibri", bold=True, size=14, color="1F4E79") ws5["A3"] = "Metric"; ws5["B3"] = "Baseline"; ws5["C3"] = "Pullback Only"; ws5["D3"] = "Delta" for c in range(1,5): ws5.cell(row=3, column=c).font = sf ws5.cell(row=3, column=c).fill = sfill baseline = {"Total Trades":686,"Wins":495,"Losses":191,"Win Rate":72.2,"Net PnL":1449.86,"Profit Factor":1.42,"Max Drawdown %":5.4,"Avg Win":9.92,"Avg Loss":18.11,"Expectancy":2.11,"Sharpe Ratio":1.98,"Early Cut":92} improved = {"Total Trades":stats.total_trades,"Wins":stats.wins,"Losses":stats.losses,"Win Rate":stats.win_rate,"Net PnL":net_pnl,"Profit Factor":stats.profit_factor,"Max Drawdown %":stats.max_drawdown,"Avg Win":stats.avg_win,"Avg Loss":stats.avg_loss,"Expectancy":stats.expectancy,"Sharpe Ratio":stats.sharpe_ratio,"Early Cut":sum(1 for t in stats.trades if t.exit_reason==ExitReason.EARLY_CUT)} row = 4 for m in baseline: ws5.cell(row=row, column=1, value=m) ws5.cell(row=row, column=2, value=baseline[m]) ws5.cell(row=row, column=3, value=improved[m]) if isinstance(baseline[m], (int,float)) and isinstance(improved[m], (int,float)): delta = improved[m] - baseline[m] ws5.cell(row=row, column=4, value=round(delta,2)) better = delta > 0 if m in ["Losses","Max Drawdown %","Avg Loss","Early Cut"]: better = delta < 0 ws5.cell(row=row, column=4).font = Font(bold=True, color="006100" if better else "9C0006") row += 1 for c in range(1,5): ws5.column_dimensions[get_column_letter(c)].width = 22 wb.save(filepath) print(f"\n Report saved: {filepath}") # ─── Log Generator ───────────────────────────────────────────── def generate_log(stats, filepath, start_date, end_date): net_pnl = stats.total_profit - stats.total_loss L = [] L.append("=" * 80) L.append("XAUBOT AI — Pullback Filter ONLY Backtest Log") L.append("=" * 80) L.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") L.append(f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}") L.append(f"Strategy: SMC-Only v4 + Pullback Filter (no sell filter)") L.append("") L.append("--- FILTER STATS ---") L.append(f" Pullback signals blocked: {stats.pullback_filtered}") L.append(f" BUY blocked: {stats.pullback_filtered_buy}") L.append(f" SELL blocked: {stats.pullback_filtered_sell}") L.append("") L.append("--- PERFORMANCE SUMMARY ---") L.append(f" Total Trades: {stats.total_trades}") L.append(f" Wins: {stats.wins}") L.append(f" Losses: {stats.losses}") L.append(f" Win Rate: {stats.win_rate:.1f}%") L.append(f" Total Profit: ${stats.total_profit:,.2f}") L.append(f" Total Loss: ${stats.total_loss:,.2f}") L.append(f" Net PnL: ${net_pnl:,.2f}") L.append(f" Profit Factor: {stats.profit_factor:.2f}") L.append(f" Max Drawdown: {stats.max_drawdown:.1f}% (${stats.max_drawdown_usd:,.2f})") L.append(f" Avg Win: ${stats.avg_win:,.2f}") L.append(f" Avg Loss: ${stats.avg_loss:,.2f}") L.append(f" Expectancy: ${stats.expectancy:,.2f}") L.append(f" Sharpe Ratio: {stats.sharpe_ratio:.2f}") L.append(f" Avoided (AVOID): {stats.avoided_signals}") L.append(f" Recovery Trades: {stats.recovery_mode_trades}") L.append(f" Daily Stops: {stats.daily_limit_stops}") L.append("") L.append("--- COMPARISON vs BASELINE ---") L.append(f" Baseline Net PnL: $1,449.86 | Improved: ${net_pnl:,.2f} | Delta: ${net_pnl-1449.86:+,.2f}") L.append(f" Baseline WR: 72.2% | Improved: {stats.win_rate:.1f}%") L.append(f" Baseline Trades: 686 | Improved: {stats.total_trades}") L.append("") L.append("--- EXIT REASON BREAKDOWN ---") ec = {} for t in stats.trades: ec[t.exit_reason.value] = ec.get(t.exit_reason.value, 0) + 1 for r, c in sorted(ec.items(), key=lambda x: -x[1]): L.append(f" {r:20s}: {c:4d} ({c/stats.total_trades*100:5.1f}%)") L.append("") L.append("--- DIRECTION BREAKDOWN ---") for d in ["BUY","SELL"]: dt = [t for t in stats.trades if t.direction == d] dw = sum(1 for t in dt if t.result == TradeResult.WIN) dp = sum(t.profit_usd for t in dt) dwr = dw/len(dt)*100 if dt else 0 L.append(f" {d}: {len(dt)} trades, {dwr:.1f}% WR, ${dp:,.2f}") L.append("") L.append("--- SESSION BREAKDOWN ---") ss = {} for t in stats.trades: if t.session not in ss: ss[t.session] = {"w":0,"l":0,"p":0.0} if t.result == TradeResult.WIN: ss[t.session]["w"] += 1 else: ss[t.session]["l"] += 1 ss[t.session]["p"] += t.profit_usd for s, d in sorted(ss.items(), key=lambda x: -x[1]["p"]): total = d["w"]+d["l"] wr = d["w"]/total*100 if total > 0 else 0 L.append(f" {s:30s}: {total:3d} trades, {wr:5.1f}% WR, ${d['p']:>8,.2f}") L.append("") L.append("--- SMC COMPONENT ANALYSIS ---") for cn, attr in [("BOS","has_bos"),("CHoCH","has_choch"),("FVG","has_fvg"),("OB","has_ob")]: ct = [t for t in stats.trades if getattr(t, attr)] cw = sum(1 for t in ct if t.result == TradeResult.WIN) cp = sum(t.profit_usd for t in ct) cwr = cw/len(ct)*100 if ct else 0 L.append(f" {cn:6s}: {len(ct):3d} trades, {cwr:5.1f}% WR, ${cp:>8,.2f}") L.append("") L.append("--- TRADE LOG ---") L.append(f"{'#':>4} {'Entry Time':>16} {'Dir':>4} {'Entry':>10} {'Exit':>10} {'P/L($)':>8} {'Result':>6} {'Exit Reason':>18} {'Conf':>5} {'Mode':>10} {'Session':>20}") L.append("-" * 140) for idx, t in enumerate(stats.trades, 1): L.append(f"{idx:4d} {t.entry_time.strftime('%Y-%m-%d %H:%M'):>16} {t.direction:>4} {t.entry_price:>10.2f} {t.exit_price:>10.2f} {t.profit_usd:>8.2f} {t.result.value:>6} {t.exit_reason.value:>18} {t.smc_confidence:>5.0%} {t.trading_mode:>10} {t.session:>20}") L.append("\n" + "=" * 80) L.append("END OF REPORT") with open(filepath, "w", encoding="utf-8") as f: f.write("\n".join(L)) print(f" Log saved: {filepath}") # ─── Main ────────────────────────────────────────────────────── def main(): print("=" * 70) print("XAUBOT AI — Pullback Filter ONLY Backtest") print("Base: SMC-Only v4 (100% synced) + Pullback Filter") print("NO sell filter applied") print("=" * 70) config = get_config() mt5 = MT5Connector(login=config.mt5_login, password=config.mt5_password, server=config.mt5_server, path=config.mt5_path) mt5.connect() print(f"\nConnected to MT5") print("Fetching XAUUSD M15 historical data...") df = mt5.get_market_data(symbol="XAUUSD", timeframe="M15", count=50000) if len(df) == 0: print("ERROR: No data received"); mt5.disconnect(); return print(f" Received {len(df)} bars") times = df["time"].to_list() print(f" Data range: {times[0]} to {times[-1]}") end_date = datetime.now() start_date = datetime(2025, 8, 1) data_start = times[0] if hasattr(data_start, 'replace') and data_start.tzinfo: start_date = start_date.replace(tzinfo=data_start.tzinfo) end_date = end_date.replace(tzinfo=data_start.tzinfo) if data_start > start_date: start_date = data_start + timedelta(days=5) print(f"\n Backtest period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}") print("\nCalculating indicators...") features = FeatureEngineer() smc = SMCAnalyzer(swing_length=config.smc.swing_length, ob_lookback=config.smc.ob_lookback) df = features.calculate_all(df, include_ml_features=True) df = smc.calculate_all(df) regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl") try: regime_detector.load() df = regime_detector.predict(df) print(" HMM regime loaded") except Exception: print(" [WARN] HMM not available") print(" Indicators calculated") bt = SMCOnlyPullbackOnly( capital=5000.0, max_daily_loss_percent=5.0, max_loss_per_trade_percent=1.0, base_lot_size=0.01, max_lot_size=0.02, recovery_lot_size=0.01, breakeven_pips=30.0, trail_start_pips=50.0, trail_step_pips=30.0, min_profit_to_protect=5.0, max_drawdown_from_peak=50.0, trade_cooldown_bars=10, trend_reversal_mult=0.6, ) stats = bt.run(df=df, start_date=start_date, end_date=end_date, initial_capital=5000.0) net_pnl = stats.total_profit - stats.total_loss print("\n" + "=" * 70) print("PULLBACK FILTER ONLY — RESULTS") print("=" * 70) print(f"\n Pullback Filter Stats:") print(f" Total blocked: {stats.pullback_filtered}") print(f" BUY blocked: {stats.pullback_filtered_buy}") print(f" SELL blocked: {stats.pullback_filtered_sell}") print(f"\n Performance:") print(f" Total Trades: {stats.total_trades}") print(f" Wins: {stats.wins}") print(f" Losses: {stats.losses}") print(f" Win Rate: {stats.win_rate:.1f}%") print(f"\n Profit/Loss:") print(f" Total Profit: ${stats.total_profit:,.2f}") print(f" Total Loss: ${stats.total_loss:,.2f}") print(f" Net PnL: ${net_pnl:,.2f}") print(f" Profit Factor: {stats.profit_factor:.2f}") print(f"\n Risk Metrics:") print(f" Max Drawdown: {stats.max_drawdown:.1f}% (${stats.max_drawdown_usd:,.2f})") print(f" Avg Win: ${stats.avg_win:,.2f}") print(f" Avg Loss: ${stats.avg_loss:,.2f}") print(f" Expectancy: ${stats.expectancy:,.2f}") print(f" Sharpe Ratio: {stats.sharpe_ratio:.2f}") print(f"\n vs BASELINE:") print(f" Baseline Net PnL: $1,449.86") print(f" Improved Net PnL: ${net_pnl:,.2f}") print(f" Delta: ${net_pnl - 1449.86:+,.2f}") print(f"\n Exit Reasons:") ec = {} for t in stats.trades: ec[t.exit_reason.value] = ec.get(t.exit_reason.value, 0) + 1 for r, c in sorted(ec.items(), key=lambda x: -x[1]): print(f" {r:20s}: {c} ({c/stats.total_trades*100:.1f}%)") print(f"\n Direction:") for d in ["BUY","SELL"]: dt = [t for t in stats.trades if t.direction == d] dw = sum(1 for t in dt if t.result == TradeResult.WIN) dp = sum(t.profit_usd for t in dt) dwr = dw/len(dt)*100 if dt else 0 print(f" {d}: {len(dt)} trades, {dwr:.1f}% WR, ${dp:,.2f}") ts = datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "04_pullback_only_results") os.makedirs(out_dir, exist_ok=True) log_path = os.path.join(out_dir, f"pullback_only_{ts}.log") xlsx_path = os.path.join(out_dir, f"pullback_only_{ts}.xlsx") generate_log(stats, log_path, start_date, end_date) generate_xlsx_report(stats, xlsx_path, start_date, end_date) mt5.disconnect() print("\n" + "=" * 70) print(f"Output: {out_dir}") print(f" Log: {os.path.basename(log_path)}") print(f" Report: {os.path.basename(xlsx_path)}") print("=" * 70) print("Backtest complete!") if __name__ == "__main__": main()