""" Backtesting engine. Loads candle data, simulates signal generation and trade execution, and reports full statistics including Walk-Forward Analysis. """ import logging import math from dataclasses import dataclass, field from typing import List, Optional, Tuple from tabdeal_bot.exchange.models import Candle, Direction, TradeResult from tabdeal_bot.strategy.signal_engine import SignalEngine from tabdeal_bot.risk.position_sizer import PositionSizer from tabdeal_bot.config.settings import BotConfig logger = logging.getLogger(__name__) @dataclass class BacktestTrade: direction: Direction entry_price: float sl_price: float tp_price: float quantity: float entry_idx: int exit_idx: Optional[int] = None exit_price: Optional[float] = None pnl: float = 0.0 fees: float = 0.0 result: Optional[TradeResult] = None @dataclass class BacktestResult: total_trades: int = 0 wins: int = 0 losses: int = 0 win_rate: float = 0.0 total_net_pnl: float = 0.0 avg_win: float = 0.0 avg_loss: float = 0.0 profit_factor: float = 0.0 expected_value: float = 0.0 max_drawdown: float = 0.0 max_drawdown_pct: float = 0.0 sharpe: float = 0.0 sortino: float = 0.0 max_consec_losses: int = 0 trades: List[BacktestTrade] = field(default_factory=list) buy_hold_pnl: float = 0.0 final_balance: float = 0.0 class BacktestEngine: def __init__(self, cfg: BotConfig): self.cfg = cfg self.signal_engine = SignalEngine(cfg.strategy, cfg.trading) self.sizer = PositionSizer(cfg.trading) def run(self, candles: List[Candle], initial_balance: float = 1000.0) -> BacktestResult: """ Single-pass backtest over candles. Uses a rolling window: at each bar we have all prior candles. """ tc = self.cfg.trading min_bars = self.cfg.strategy.ema_slow + self.cfg.strategy.atr_period + 10 result = BacktestResult() balance = initial_balance peak_balance = initial_balance in_trade: Optional[BacktestTrade] = None last_trade_result: Optional[TradeResult] = None last_direction = Direction.NONE consecutive_losses = 0 net_pnls: List[float] = [] consec_loss_cur = 0 for i in range(min_bars, len(candles)): window = candles[:i + 1] price = candles[i].close # --- Close open trade if SL/TP hit --- if in_trade is not None: high = candles[i].high low = candles[i].low closed = False if in_trade.direction == Direction.LONG: if low <= in_trade.sl_price: exit_p = in_trade.sl_price in_trade.exit_price = exit_p in_trade.pnl = (exit_p - in_trade.entry_price) * in_trade.quantity * tc.leverage closed = True elif high >= in_trade.tp_price: exit_p = in_trade.tp_price in_trade.exit_price = exit_p in_trade.pnl = (exit_p - in_trade.entry_price) * in_trade.quantity * tc.leverage closed = True else: # SHORT if high >= in_trade.sl_price: exit_p = in_trade.sl_price in_trade.exit_price = exit_p in_trade.pnl = (in_trade.entry_price - exit_p) * in_trade.quantity * tc.leverage closed = True elif low <= in_trade.tp_price: exit_p = in_trade.tp_price in_trade.exit_price = exit_p in_trade.pnl = (in_trade.entry_price - exit_p) * in_trade.quantity * tc.leverage closed = True if closed: in_trade.exit_idx = i in_trade.fees = (in_trade.entry_price + in_trade.exit_price) * in_trade.quantity * (tc.taker_fee + tc.slippage) net = in_trade.pnl - in_trade.fees balance += net net_pnls.append(net) result.total_trades += 1 if net > 0: in_trade.result = TradeResult.WIN result.wins += 1 consec_loss_cur = 0 else: in_trade.result = TradeResult.LOSS result.losses += 1 consec_loss_cur += 1 result.max_consec_losses = max(result.max_consec_losses, consec_loss_cur) last_trade_result = in_trade.result last_direction = in_trade.direction result.trades.append(in_trade) in_trade = None # Drawdown if balance > peak_balance: peak_balance = balance dd = peak_balance - balance result.max_drawdown = max(result.max_drawdown, dd) if balance <= 0: logger.warning("Balance depleted at bar %d — stopping backtest", i) break else: continue # --- Seek new entry --- signal = self.signal_engine.analyze(window) if signal is None: continue # Adaptive direction effective_dir = signal.direction if last_trade_result == TradeResult.LOSS and tc.reverse_on_loss: preferred = Direction.SHORT if last_direction == Direction.LONG else Direction.LONG if signal.direction != preferred: continue elif last_trade_result == TradeResult.WIN and tc.continue_on_win: if signal.direction != last_direction and last_direction != Direction.NONE: pass # allow it anyway in backtest qty = self.sizer.calculate(balance, signal.entry_price, signal.sl_price) if qty <= 0: continue in_trade = BacktestTrade( direction=signal.direction, entry_price=signal.entry_price, sl_price=signal.sl_price, tp_price=signal.tp_price, quantity=qty, entry_idx=i, ) # --- Compute statistics --- result.final_balance = balance t = result.total_trades if t == 0: return result result.win_rate = result.wins / t win_pnls = [tr.pnl - tr.fees for tr in result.trades if tr.result == TradeResult.WIN] loss_pnls = [abs(tr.pnl - tr.fees) for tr in result.trades if tr.result == TradeResult.LOSS] result.avg_win = sum(win_pnls) / len(win_pnls) if win_pnls else 0.0 result.avg_loss = sum(loss_pnls) / len(loss_pnls) if loss_pnls else 0.0 result.total_net_pnl = sum(net_pnls) gross_win = sum(win_pnls) gross_loss = sum(loss_pnls) result.profit_factor = gross_win / gross_loss if gross_loss > 0 else float("inf") result.expected_value = (result.win_rate * result.avg_win) - ((1 - result.win_rate) * result.avg_loss) if peak_balance > 0: result.max_drawdown_pct = result.max_drawdown / peak_balance if len(net_pnls) > 1: mean_r = result.total_net_pnl / len(net_pnls) std_r = math.sqrt(sum((r - mean_r) ** 2 for r in net_pnls) / len(net_pnls)) result.sharpe = mean_r / std_r if std_r > 0 else 0.0 neg = [r for r in net_pnls if r < 0] downstd = math.sqrt(sum(r ** 2 for r in neg) / len(neg)) if neg else 0.0 result.sortino = mean_r / downstd if downstd > 0 else 0.0 # Buy & hold comparison if candles: result.buy_hold_pnl = (candles[-1].close - candles[0].close) / candles[0].close * initial_balance return result def walk_forward( self, candles: List[Candle], train_pct: float = 0.7, initial_balance: float = 1000.0, ) -> Tuple[BacktestResult, BacktestResult]: """Split candles into train/test and run backtest on each.""" split = int(len(candles) * train_pct) train = candles[:split] test = candles[split:] logger.info( "Walk-Forward: train=%d bars, test=%d bars", len(train), len(test) ) train_result = self.run(train, initial_balance) test_result = self.run(test, initial_balance) return train_result, test_result @staticmethod def print_result(result: BacktestResult, label: str = "Backtest"): sep = "─" * 48 print(f"\n{'═' * 48}") print(f" {label.upper()} RESULTS") print(f"{'═' * 48}") print(f" Trades: {result.total_trades:>8d}") print(f" Wins / Losses: {result.wins:>4d} / {result.losses:<4d}") print(f" Win Rate: {result.win_rate * 100:>8.1f}%") print(f" Avg Win: {result.avg_win:>+10.4f}") print(f" Avg Loss: {result.avg_loss:>+10.4f}") print(f" Profit Factor: {result.profit_factor:>10.3f}") print(f" Expected Value: {result.expected_value:>+10.4f}") print(f" Total Net PnL: {result.total_net_pnl:>+10.4f}") print(f" Final Balance: {result.final_balance:>10.2f}") print(f" Max Drawdown: {result.max_drawdown:>+10.4f} ({result.max_drawdown_pct * 100:.1f}%)") print(f" Sharpe Ratio: {result.sharpe:>10.3f}") print(f" Sortino Ratio: {result.sortino:>10.3f}") print(f" Max Consec. Losses: {result.max_consec_losses:>8d}") print(sep) print(f" Buy & Hold PnL: {result.buy_hold_pnl:>+10.4f}") print(f"{'═' * 48}\n") if result.expected_value > 0 and result.profit_factor > 1.0: print(" ✓ Positive expectancy — strategy has statistical edge") else: print(" ✗ Negative or zero expectancy — do NOT trade live with these parameters") print()