""" Simple backtesting engine for strategy evaluation """ import pandas as pd import numpy as np from typing import Dict, List, Optional from datetime import datetime import logging from config import Config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class BacktestEngine: """ Simple backtesting engine for evaluating trading strategies """ def __init__( self, initial_capital: float = None, commission: float = 0.001, # 0.1% per trade slippage: float = 0.0005 # 0.05% slippage ): self.initial_capital = initial_capital or Config.INITIAL_CAPITAL self.commission = commission self.slippage = slippage self.reset() def reset(self): """Reset backtest state""" self.capital = self.initial_capital self.position = 0 # Current position size self.entry_price = 0 self.trades = [] self.equity_curve = [] self.current_trade = None def execute_trade(self, signal: str, price: float, timestamp, confidence: float = 1.0): """ Execute a trade based on signal Args: signal: 'BUY', 'SELL', or 'HOLD' price: Current price timestamp: Trade timestamp confidence: Signal confidence (affects position size) """ # Apply slippage if signal == 'BUY': actual_price = price * (1 + self.slippage) elif signal == 'SELL': actual_price = price * (1 - self.slippage) else: actual_price = price # BUY signal if signal == 'BUY' and self.position == 0: # Calculate position size based on confidence and max position size position_size = min(confidence, Config.MAX_POSITION_SIZE) position_value = self.capital * position_size # Account for commission commission_cost = position_value * self.commission position_value -= commission_cost # Calculate number of units units = position_value / actual_price self.position = units self.entry_price = actual_price self.capital -= (position_value + commission_cost) self.current_trade = { 'entry_time': timestamp, 'entry_price': actual_price, 'type': 'LONG', 'units': units, 'commission': commission_cost } # SELL signal (close position) elif signal == 'SELL' and self.position > 0: # Calculate exit value exit_value = self.position * actual_price commission_cost = exit_value * self.commission exit_value -= commission_cost self.capital += exit_value # Calculate P&L pnl = (actual_price - self.entry_price) * self.position - self.current_trade['commission'] - commission_cost pnl_pct = (actual_price - self.entry_price) / self.entry_price # Record trade trade_record = { **self.current_trade, 'exit_time': timestamp, 'exit_price': actual_price, 'pnl': pnl, 'pnl_pct': pnl_pct, 'exit_commission': commission_cost } self.trades.append(trade_record) # Reset position self.position = 0 self.entry_price = 0 self.current_trade = None # Record equity current_equity = self.capital if self.position > 0: current_equity += self.position * price self.equity_curve.append({ 'timestamp': timestamp, 'equity': current_equity, 'cash': self.capital, 'position_value': self.position * price if self.position > 0 else 0 }) def run_backtest(self, df: pd.DataFrame, strategy) -> Dict: """ Run backtest on historical data Args: df: OHLCV dataframe strategy: Trading strategy instance Returns: Dict with backtest results """ self.reset() # Calculate indicators df = strategy.calculate_indicators(df) # Iterate through data for idx in range(len(df)): if idx < 50: # Skip initial period for indicators to stabilize continue current_df = df.iloc[:idx+1] signal, confidence, metadata = strategy.generate_signal(current_df) timestamp = df.iloc[idx]['timestamp'] price = df.iloc[idx]['close'] self.execute_trade(signal, price, timestamp, confidence) # Close any open position at the end if self.position > 0: last_price = df.iloc[-1]['close'] last_timestamp = df.iloc[-1]['timestamp'] self.execute_trade('SELL', last_price, last_timestamp, 1.0) # Calculate metrics metrics = self.calculate_metrics() return { 'strategy': strategy.name, 'initial_capital': self.initial_capital, 'final_capital': self.capital + (self.position * df.iloc[-1]['close']), 'metrics': metrics, 'trades': self.trades, 'equity_curve': self.equity_curve } def calculate_metrics(self) -> Dict: """Calculate performance metrics""" if not self.trades: return { 'total_trades': 0, 'win_rate': 0, 'avg_profit': 0, 'avg_loss': 0, 'profit_factor': 0, 'max_drawdown': 0, 'sharpe_ratio': 0 } trades_df = pd.DataFrame(self.trades) # Basic metrics total_trades = len(trades_df) winning_trades = trades_df[trades_df['pnl'] > 0] losing_trades = trades_df[trades_df['pnl'] < 0] win_rate = len(winning_trades) / total_trades if total_trades > 0 else 0 avg_profit = winning_trades['pnl'].mean() if len(winning_trades) > 0 else 0 avg_loss = abs(losing_trades['pnl'].mean()) if len(losing_trades) > 0 else 0 total_profit = winning_trades['pnl'].sum() if len(winning_trades) > 0 else 0 total_loss = abs(losing_trades['pnl'].sum()) if len(losing_trades) > 0 else 0 profit_factor = total_profit / total_loss if total_loss > 0 else float('inf') # Drawdown equity_df = pd.DataFrame(self.equity_curve) if len(equity_df) > 0: equity_df['cummax'] = equity_df['equity'].cummax() equity_df['drawdown'] = (equity_df['equity'] - equity_df['cummax']) / equity_df['cummax'] max_drawdown = abs(equity_df['drawdown'].min()) # Sharpe Ratio (simplified) equity_df['returns'] = equity_df['equity'].pct_change() sharpe_ratio = (equity_df['returns'].mean() / equity_df['returns'].std() * np.sqrt(252) if equity_df['returns'].std() > 0 else 0) else: max_drawdown = 0 sharpe_ratio = 0 # Total return final_equity = equity_df.iloc[-1]['equity'] if len(equity_df) > 0 else self.initial_capital total_return = (final_equity - self.initial_capital) / self.initial_capital return { 'total_trades': total_trades, 'winning_trades': len(winning_trades), 'losing_trades': len(losing_trades), 'win_rate': win_rate, 'avg_profit': avg_profit, 'avg_loss': avg_loss, 'profit_factor': profit_factor, 'max_drawdown': max_drawdown, 'sharpe_ratio': sharpe_ratio, 'total_return': total_return, 'total_return_pct': total_return * 100 } def get_equity_curve_df(self) -> pd.DataFrame: """Get equity curve as dataframe""" return pd.DataFrame(self.equity_curve) def get_trades_df(self) -> pd.DataFrame: """Get trades as dataframe""" if not self.trades: return pd.DataFrame() return pd.DataFrame(self.trades) def compare_strategies(self, df: pd.DataFrame, strategies: List) -> pd.DataFrame: """ Compare multiple strategies Args: df: OHLCV dataframe strategies: List of strategy instances Returns: DataFrame with comparison results """ results = [] for strategy in strategies: logger.info(f"Backtesting {strategy.name}...") result = self.run_backtest(df.copy(), strategy) results.append({ 'Strategy': strategy.name, 'Total Return %': result['metrics']['total_return_pct'], 'Win Rate %': result['metrics']['win_rate'] * 100, 'Total Trades': result['metrics']['total_trades'], 'Profit Factor': result['metrics']['profit_factor'], 'Max Drawdown %': result['metrics']['max_drawdown'] * 100, 'Sharpe Ratio': result['metrics']['sharpe_ratio'], 'Final Capital': result['final_capital'] }) return pd.DataFrame(results).sort_values('Total Return %', ascending=False)