"""Event-Driven Historical Backtesting Engine.""" from __future__ import annotations from dataclasses import dataclass, field from decimal import Decimal from typing import Any, Optional, Sequence from backend.app.backtesting.metrics import BacktestReport, calculate_performance_metrics from backend.app.risk.engine import risk_engine from backend.app.schemas.market import Candle, Ticker from backend.app.schemas.risk import AccountState, InstrumentFilters from backend.app.schemas.signal import SignalAction from backend.app.strategy.signal_engine import signal_engine @dataclass class BacktestTrade: entry_bar_idx: int exit_bar_idx: int entry_time: int exit_time: int symbol: str side: str quantity: float entry_price: float exit_price: float pnl: float fee_paid: float exit_reason: str # STOP_LOSS, TAKE_PROFIT, END_OF_DATA class BacktestEngine: """Simulates realistic chronological execution with full fee and slippage modeling.""" def __init__( self, initial_equity: float = 10000.0, taker_fee_pct: float = 0.00055, slippage_pct: float = 0.00020, ) -> None: self.initial_equity = initial_equity self.taker_fee_pct = taker_fee_pct self.slippage_pct = slippage_pct def run( self, candles: Sequence[Candle], filters: Optional[InstrumentFilters] = None, ) -> tuple[BacktestReport, list[BacktestTrade], list[float]]: if len(candles) < 40: empty_report = calculate_performance_metrics(self.initial_equity, [], [], []) return empty_report, [], [] symbol = candles[0].symbol inst_filters = filters or InstrumentFilters( symbol=symbol, min_order_qty=Decimal("0.001"), qty_step=Decimal("0.001"), min_notional=Decimal("5.0"), tick_size=Decimal("0.1"), max_leverage=Decimal("1.0"), ) equity = Decimal(str(self.initial_equity)) peak_equity = equity starting_day_equity = equity trades: list[BacktestTrade] = [] equity_curve: list[float] = [float(equity)] # Active position state active_pos: Optional[dict[str, Any]] = None # Warm-up 35 candles for indicators for i in range(35, len(candles)): current_candle = candles[i] history = candles[: i + 1] # 1. Manage active position if open if active_pos is not None: side = active_pos["side"] entry_p = active_pos["entry_price"] sl_p = active_pos["stop_loss"] tp_p = active_pos["take_profit"] qty = active_pos["quantity"] c_high = float(current_candle.high) c_low = float(current_candle.low) c_close = float(current_candle.close) exit_triggered = False exit_price = 0.0 exit_reason = "" if side == "LONG": # Conservative check: If low touched SL, stop out first if c_low <= sl_p: exit_triggered = True exit_price = sl_p * (1.0 - self.slippage_pct) exit_reason = "STOP_LOSS" elif c_high >= tp_p: exit_triggered = True exit_price = tp_p * (1.0 - self.slippage_pct) exit_reason = "TAKE_PROFIT" elif side == "SHORT": if c_high >= sl_p: exit_triggered = True exit_price = sl_p * (1.0 + self.slippage_pct) exit_reason = "STOP_LOSS" elif c_low <= tp_p: exit_triggered = True exit_price = tp_p * (1.0 + self.slippage_pct) exit_reason = "TAKE_PROFIT" if exit_triggered: # Calculate PnL and friction mult = 1.0 if side == "LONG" else -1.0 raw_pnl = (exit_price - entry_p) * qty * mult exit_fee = exit_price * qty * self.taker_fee_pct total_fee = active_pos["entry_fee"] + exit_fee net_trade_pnl = raw_pnl - total_fee equity += Decimal(str(net_trade_pnl)) if equity > peak_equity: peak_equity = equity trades.append( BacktestTrade( entry_bar_idx=active_pos["entry_bar_idx"], exit_bar_idx=i, entry_time=active_pos["entry_time"], exit_time=current_candle.timestamp, symbol=symbol, side=side, quantity=qty, entry_price=entry_p, exit_price=exit_price, pnl=net_trade_pnl, fee_paid=total_fee, exit_reason=exit_reason, ) ) active_pos = None # 2. Evaluate new entry if no active position if active_pos is None: mock_ticker = Ticker( symbol=symbol, bid_price=current_candle.close * Decimal("0.9999"), ask_price=current_candle.close * Decimal("1.0001"), last_price=current_candle.close, volume_24h=Decimal("10000.0"), price_24h_pcnt=Decimal("0.0"), high_24h=current_candle.high, low_24h=current_candle.low, timestamp=current_candle.timestamp, ) signal = signal_engine.generate_signal(history, mock_ticker) if signal.action in [SignalAction.LONG, SignalAction.SHORT]: account_state = AccountState( equity=equity, available_balance=equity, starting_day_equity=starting_day_equity, peak_equity=peak_equity, open_positions_count=0, current_leverage=Decimal("1.0"), realized_daily_pnl=Decimal("0.0"), unrealized_daily_pnl=Decimal("0.0"), ) decision = risk_engine.evaluate_signal(signal, account_state, inst_filters) if decision.approved and decision.approved_quantity > Decimal("0"): qty_f = float(decision.approved_quantity) entry_p_f = float(signal.entry_price) * ( 1.0 + self.slippage_pct if signal.action == SignalAction.LONG else 1.0 - self.slippage_pct ) entry_fee = entry_p_f * qty_f * self.taker_fee_pct active_pos = { "side": signal.action.value, "entry_price": entry_p_f, "stop_loss": float(decision.stop_loss_price), "take_profit": float(decision.take_profit_price), "quantity": qty_f, "entry_fee": entry_fee, "entry_bar_idx": i, "entry_time": current_candle.timestamp, } equity_curve.append(float(equity)) # Close any lingering position at final candle if active_pos is not None: last_c = candles[-1] c_close = float(last_c.close) side = active_pos["side"] mult = 1.0 if side == "LONG" else -1.0 raw_pnl = (c_close - active_pos["entry_price"]) * active_pos["quantity"] * mult exit_fee = c_close * active_pos["quantity"] * self.taker_fee_pct total_fee = active_pos["entry_fee"] + exit_fee net_pnl = raw_pnl - total_fee equity += Decimal(str(net_pnl)) trades.append( BacktestTrade( entry_bar_idx=active_pos["entry_bar_idx"], exit_bar_idx=len(candles) - 1, entry_time=active_pos["entry_time"], exit_time=last_c.timestamp, symbol=symbol, side=side, quantity=active_pos["quantity"], entry_price=active_pos["entry_price"], exit_price=c_close, pnl=net_pnl, fee_paid=total_fee, exit_reason="END_OF_DATA", ) ) equity_curve[-1] = float(equity) # Benchmark Buy & Hold return first_close = float(candles[35].close) last_close = float(candles[-1].close) bnh_return = ((last_close - first_close) / first_close) * 100.0 pnls = [t.pnl for t in trades] fees = [t.fee_paid for t in trades] report = calculate_performance_metrics( self.initial_equity, equity_curve, pnls, fees, benchmark_buy_and_hold_return=bnh_return ) return report, trades, equity_curve backtest_engine = BacktestEngine()