"""Backtest engine for neural network trading system. Event-driven backtester that processes signals from NeuralPredictor. Each signal contains entry_price, sl_price, tp_price, confidence. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, List, Optional, Dict from zoneinfo import ZoneInfo import pandas as pd from loguru import logger from domain import ( Trade, TradeDirection, DEFAULT_RISK_PER_TRADE, DEFAULT_COMMISSION, DEFAULT_SLIPPAGE, ) from core.risk_manager import calculate_position_size _tz = ZoneInfo("Europe/Moscow") @dataclass class BacktestPosition: entry_time: int entry: float sl: float tp: float size: int direction: TradeDirection class NeuralBacktester: """Event-driven backtester for neural network signals. Iterates through all price bars chronologically. On each bar: - Opens new positions for matching signal timestamps - Checks SL/TP for all open positions - Records equity snapshot Attributes: capital: Current available capital initial_capital: Starting capital risk_pct: Risk percentage per trade commission: Trading commission rate slippage: Slippage rate max_positions: Maximum concurrent open positions """ def __init__( self, capital: float = 1_000_000, risk_pct: float = DEFAULT_RISK_PER_TRADE, commission: float = DEFAULT_COMMISSION, slippage: float = DEFAULT_SLIPPAGE, max_positions: int = 5, ): self.capital = capital self.initial_capital = capital self.risk_pct = risk_pct self.commission = commission self.slippage = slippage self.max_positions = max_positions self.positions: List[BacktestPosition] = [] self.trades: List[Trade] = [] self.equity: List[Dict[str, Any]] = [] self._current_ticker: str = "" def run( self, signals: pd.DataFrame, prices: pd.DataFrame, ticker: str = "", ) -> List[Trade]: """Run backtest on neural network signals and price data. Args: signals: DataFrame with columns [timestamp, signal_type, entry_price, sl_price, tp_price, ...] prices: DataFrame with OHLCV price data ticker: Ticker symbol being backtested Returns: List of completed trades """ if signals.empty or prices.empty: logger.warning("Empty signals or prices") return [] self.positions.clear() self.trades.clear() self.equity.clear() self._current_ticker = ticker logger.debug( f"NeuralBacktester.run: capital={self.capital:.2f}, " f"max_positions={self.max_positions}, ticker={ticker}" ) prices_idx = prices.set_index("timestamp").sort_index() bar_timestamps = list(prices_idx.index) signals_by_ts = {} for _, row in signals.iterrows(): ts = row.get("timestamp") if ts is not None: signals_by_ts[ts] = row self.equity.append({"timestamp": bar_timestamps[0], "equity": self.capital}) for ts in bar_timestamps: price_row = prices_idx.loc[ts] if ts in signals_by_ts: signal = signals_by_ts[ts] if len(self.positions) < self.max_positions: self._open_position(signal, price_row, ticker) closed_any = False for pos in list(self.positions): self._check_exits(pos, price_row, ts) if pos.size == 0: self.positions.remove(pos) closed_any = True if closed_any or ts == bar_timestamps[-1]: self.equity.append({"timestamp": ts, "equity": self.capital}) self._close_open_positions(prices_idx) self._dedup_equity() if self.equity: self.equity[-1]["equity"] = self.capital logger.info( f"Backtest completed: {len(self.trades)} trades, " f"final capital {self.capital:,.2f}" ) return self.trades def _open_position( self, signal: pd.Series, price_row: pd.Series, ticker: str ) -> None: entry = signal.get("entry_price", price_row.get("Close")) sl = signal.get("sl_price", entry * 0.99) tp = signal.get("tp_price", entry * 1.02) signal_side = signal.get("signal_type", "LONG").upper() if signal_side not in ("LONG", "SHORT"): logger.warning(f"Unknown signal_type: {signal_side}, skipping") return size = calculate_position_size( self.capital, self.risk_pct, entry, sl, side=signal_side ) if size <= 0: return direction = ( TradeDirection.BUY if signal_side == "LONG" else TradeDirection.SELL ) pos = BacktestPosition( entry_time=signal.get("timestamp"), entry=entry, sl=sl, tp=tp, size=size, direction=direction, ) self.positions.append(pos) logger.info( f"Opened {direction.value}: entry={entry:.2f}, sl={sl:.4f}, " f"tp={tp:.2f}, size={size}, positions={len(self.positions)}, " f"confidence={signal.get('ai_confidence', 0):.2f}" ) def _check_exits( self, pos: BacktestPosition, price_row: pd.Series, current_ts: int ) -> None: entry = pos.entry sl = pos.sl tp = pos.tp direction = pos.direction size = pos.size if size <= 0: return current_price = price_row.get("Close", 0) bar_low = price_row.get("Low", current_price) bar_high = price_row.get("High", current_price) status = "OPEN" exit_price = None pnl = 0.0 if direction == TradeDirection.BUY: if bar_low <= sl: exit_price = sl pnl = (sl - entry) * size status = "SL" elif bar_high >= tp: exit_price = tp pnl = (tp - entry) * size status = "TP" else: if bar_high >= sl: exit_price = sl pnl = (entry - sl) * size status = "SL" elif bar_low <= tp: exit_price = tp pnl = (entry - tp) * size status = "TP" if status != "OPEN": self._close_position(current_ts, exit_price, pnl, status, pos, ticker=self._current_ticker) def _close_position( self, exit_ts: int, exit_price: float, pnl: float, status: str, pos: BacktestPosition, ticker: str = "", ) -> None: entry = pos.entry size = pos.size direction = pos.direction trade_value = entry * size commission = trade_value * self.commission slippage_cost = trade_value * self.slippage net_pnl = pnl - commission - slippage_cost trade = Trade( ticker=ticker, direction=direction, entry_price=entry, exit_price=exit_price, quantity=size, entry_time=datetime.fromtimestamp(int(pos.entry_time), _tz), exit_time=datetime.fromtimestamp(int(exit_ts), _tz), pnl=net_pnl, pnl_percent=net_pnl / (entry * size) if entry > 0 else 0, commission=commission, slippage=slippage_cost, sl_price=pos.sl, tp_price=pos.tp, status=status, ) self.trades.append(trade) self.capital += net_pnl pos.size = 0 logger.info( f"Closed {direction.value}: {status}, PnL={net_pnl:.2f}, " f"positions={len(self.positions)}" ) def _close_open_positions(self, prices: pd.DataFrame) -> None: if not self.positions: return last_ts = prices.index[-1] last_price = prices.iloc[-1]["Close"] for pos in list(self.positions): if pos.size > 0: pnl = self._calculate_pnl(last_price, pos) self._close_position(last_ts, last_price, pnl, "END", pos, ticker=self._current_ticker) self.positions.remove(pos) def _calculate_pnl(self, current_price: float, pos: BacktestPosition) -> float: if pos.direction == TradeDirection.BUY: return (current_price - pos.entry) * pos.size else: return (pos.entry - current_price) * pos.size def _dedup_equity(self) -> None: """Keep only the last equity entry per timestamp.""" self.equity.sort(key=lambda x: x["timestamp"]) seen = {} for entry in self.equity: seen[entry["timestamp"]] = entry self.equity = sorted(seen.values(), key=lambda x: x["timestamp"]) def save_trades(self, ticker: str, tf: str, output_dir: Path = Path("logs")): if not self.trades: logger.warning("No trades to save") return df = pd.DataFrame(self.trades) path = output_dir / f"trades_{ticker}_{tf}.csv" df.to_csv(path, index=False) logger.info(f"Saved trades to {path}")