"""Backtesting engine for ICT strategies. Simulates trading on historical data with realistic fees and slippage to evaluate strategy performance. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import List, Optional import pandas as pd from loguru import logger from config import settings from indicators.ict_engine import ICTEngine from indicators.multi_timeframe import MarketBias, TradeDirection from indicators.confluence import ConfluenceChecker from strategy.entry_rules import EntryRules from strategy.exit_rules import ExitRules, ExitReason from risk.risk_manager import RiskManager @dataclass class BacktestTrade: """A single trade in a backtest.""" entry_index: int exit_index: int direction: TradeDirection entry_price: float exit_price: float amount: float pnl: float fee: float entry_reason: str = "" exit_reason: str = "" confluence_score: int = 0 @dataclass class BacktestResult: """Complete backtest outcome.""" trades: List[BacktestTrade] = field(default_factory=list) equity_curve: List[float] = field(default_factory=list) total_pnl: float = 0.0 roi_percent: float = 0.0 win_rate: float = 0.0 profit_factor: float = 0.0 sharpe_ratio: float = 0.0 max_drawdown: float = 0.0 total_trades: int = 0 winning_trades: int = 0 losing_trades: int = 0 avg_trade_duration: float = 0.0 # in candles initial_balance: float = 0.0 def summary(self) -> dict: return { "total_trades": self.total_trades, "winning_trades": self.winning_trades, "losing_trades": self.losing_trades, "win_rate": f"{self.win_rate:.1%}", "total_pnl": f"{self.total_pnl:.2f}", "roi": f"{self.roi_percent:.2%}", "profit_factor": f"{self.profit_factor:.2f}", "sharpe_ratio": f"{self.sharpe_ratio:.2f}", "max_drawdown": f"{self.max_drawdown:.2%}", "avg_duration_candles": f"{self.avg_trade_duration:.1f}", } class Backtester: """ICT strategy backtesting engine. Iterates over historical OHLCV data candle by candle, applying entry/exit rules and tracking performance. """ FEE_RATE = 0.001 # 0.1% taker fee SLIPPAGE_RATE = 0.0005 # 0.05% slippage def __init__( self, ict_engine: ICTEngine | None = None, entry_rules: EntryRules | None = None, exit_rules: ExitRules | None = None, risk_manager: RiskManager | None = None, ): self.engine = ict_engine or ICTEngine() self.entry_rules = entry_rules or EntryRules() self.exit_rules = exit_rules or ExitRules() self.risk = risk_manager or RiskManager() def run( self, ohlc: pd.DataFrame, initial_balance: float = 1000.0, risk_per_trade: float = 0.02, ) -> BacktestResult: """Execute backtest on the provided OHLC data. Args: ohlc: DataFrame with [open, high, low, close, volume]. initial_balance: Starting capital in quote currency. risk_per_trade: Fraction of balance to risk per trade. Returns: BacktestResult with all metrics. """ balance = initial_balance equity_curve = [balance] trades: List[BacktestTrade] = [] min_bars = max(self.engine.swing_length + 10, 60) # Active position state in_position = False pos_direction: Optional[TradeDirection] = None pos_entry_price = 0.0 pos_amount = 0.0 pos_sl = 0.0 pos_tp = 0.0 pos_entry_idx = 0 pos_trailing: Optional[float] = None pos_confluence = 0 pos_candles = 0 logger.info( "Backtest starting: {} candles, balance={}", len(ohlc), initial_balance ) for i in range(min_bars, len(ohlc)): window = ohlc.iloc[:i + 1] current_price = float(window["close"].iloc[-1]) high = float(window["high"].iloc[-1]) low = float(window["low"].iloc[-1]) if in_position: pos_candles += 1 # Check exit using high/low for SL/TP accuracy exit_price = current_price exit_reason: Optional[ExitReason] = None if pos_direction == TradeDirection.LONG: if low <= pos_sl: exit_price = pos_sl exit_reason = ExitReason.STOP_LOSS elif high >= pos_tp: exit_price = pos_tp exit_reason = ExitReason.TAKE_PROFIT else: if high >= pos_sl: exit_price = pos_sl exit_reason = ExitReason.STOP_LOSS elif low <= pos_tp: exit_price = pos_tp exit_reason = ExitReason.TAKE_PROFIT # Update trailing if exit_reason is None: pos_trailing = self.exit_rules.update_trailing_stop( pos_direction, pos_entry_price, current_price, pos_trailing ) if pos_trailing is not None: if pos_direction == TradeDirection.LONG and low <= pos_trailing: exit_price = pos_trailing exit_reason = ExitReason.TRAILING_STOP elif pos_direction == TradeDirection.SHORT and high >= pos_trailing: exit_price = pos_trailing exit_reason = ExitReason.TRAILING_STOP # Time exit if exit_reason is None and pos_candles >= self.exit_rules.time_exit_candles: exit_price = current_price exit_reason = ExitReason.TIME_EXIT # CHOCH check if exit_reason is None: try: signals = self.engine.analyze(window) choch = signals.latest_choch if choch is not None: if pos_direction == TradeDirection.LONG and choch == -1: exit_price = current_price exit_reason = ExitReason.CHOCH elif pos_direction == TradeDirection.SHORT and choch == 1: exit_price = current_price exit_reason = ExitReason.CHOCH except Exception: pass if exit_reason is not None: # Apply slippage if pos_direction == TradeDirection.LONG: exit_price *= (1 - self.SLIPPAGE_RATE) pnl = (exit_price - pos_entry_price) * pos_amount else: exit_price *= (1 + self.SLIPPAGE_RATE) pnl = (pos_entry_price - exit_price) * pos_amount fee = exit_price * pos_amount * self.FEE_RATE pnl -= fee balance += pnl trades.append( BacktestTrade( entry_index=pos_entry_idx, exit_index=i, direction=pos_direction, entry_price=pos_entry_price, exit_price=exit_price, amount=pos_amount, pnl=pnl, fee=fee, exit_reason=exit_reason.value, confluence_score=pos_confluence, ) ) in_position = False pos_trailing = None pos_candles = 0 else: # Look for entry try: signals = self.engine.analyze(window) except Exception: equity_curve.append(balance) continue # Simple bias from BOS bos = signals.latest_bos if bos is None: equity_curve.append(balance) continue if bos == 1: entry_result = self.entry_rules.check_bullish_entry(signals, current_price) direction = TradeDirection.LONG elif bos == -1: entry_result = self.entry_rules.check_bearish_entry(signals, current_price) direction = TradeDirection.SHORT else: equity_curve.append(balance) continue if entry_result.is_valid: sl = self.entry_rules.calculate_stop_loss(direction, signals, current_price) tp = self.entry_rules.calculate_take_profit(direction, signals, current_price, sl) price_risk = abs(current_price - sl) if price_risk == 0: equity_curve.append(balance) continue pos_amount = (balance * risk_per_trade) / price_risk entry_fee = current_price * pos_amount * self.FEE_RATE entry_price = current_price * ( (1 + self.SLIPPAGE_RATE) if direction == TradeDirection.LONG else (1 - self.SLIPPAGE_RATE) ) balance -= entry_fee in_position = True pos_direction = direction pos_entry_price = entry_price pos_sl = sl pos_tp = tp pos_entry_idx = i pos_confluence = len(entry_result.conditions_met) equity_curve.append(balance) result = self._calculate_metrics(trades, equity_curve, initial_balance) logger.info("Backtest complete: {}", result.summary()) return result def _calculate_metrics( self, trades: List[BacktestTrade], equity_curve: List[float], initial_balance: float, ) -> BacktestResult: """Compute performance metrics from trade list and equity curve.""" total = len(trades) wins = [t for t in trades if t.pnl > 0] losses = [t for t in trades if t.pnl <= 0] total_pnl = sum(t.pnl for t in trades) gross_profit = sum(t.pnl for t in wins) if wins else 0 gross_loss = abs(sum(t.pnl for t in losses)) if losses else 0 # Max drawdown peak = initial_balance max_dd = 0.0 for eq in equity_curve: if eq > peak: peak = eq dd = (peak - eq) / peak if peak > 0 else 0 if dd > max_dd: max_dd = dd # Sharpe ratio (simplified, daily returns) if len(equity_curve) > 1: returns = pd.Series(equity_curve).pct_change().dropna() if returns.std() > 0: sharpe = (returns.mean() / returns.std()) * (252 ** 0.5) else: sharpe = 0.0 else: sharpe = 0.0 # Average duration avg_dur = ( sum(t.exit_index - t.entry_index for t in trades) / total if total > 0 else 0 ) return BacktestResult( trades=trades, equity_curve=equity_curve, total_pnl=total_pnl, roi_percent=total_pnl / initial_balance if initial_balance > 0 else 0, win_rate=len(wins) / total if total > 0 else 0, profit_factor=gross_profit / gross_loss if gross_loss > 0 else float("inf"), sharpe_ratio=sharpe, max_drawdown=max_dd, total_trades=total, winning_trades=len(wins), losing_trades=len(losses), avg_trade_duration=avg_dur, initial_balance=initial_balance, ) @staticmethod def generate_report(result: BacktestResult) -> str: """Generate a human-readable backtest report.""" lines = [ "=" * 50, " ICT Strategy Backtest Report", "=" * 50, f" Initial Balance : ${result.initial_balance:,.2f}", f" Final Balance : ${result.initial_balance + result.total_pnl:,.2f}", f" Total PnL : ${result.total_pnl:,.2f}", f" ROI : {result.roi_percent:.2%}", "-" * 50, f" Total Trades : {result.total_trades}", f" Winning : {result.winning_trades}", f" Losing : {result.losing_trades}", f" Win Rate : {result.win_rate:.1%}", f" Profit Factor : {result.profit_factor:.2f}", "-" * 50, f" Sharpe Ratio : {result.sharpe_ratio:.2f}", f" Max Drawdown : {result.max_drawdown:.2%}", f" Avg Duration : {result.avg_trade_duration:.1f} candles", "=" * 50, ] return "\n".join(lines)