backtest/engine.py

"""
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)