QuantumEdge/backtesting.py

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional
import ccxt
from concurrent.futures import ThreadPoolExecutor
import json
import os

from .trading_strategy import TradingStrategy
from .config import load_config

logger = logging.getLogger(__name__)

class Backtester:
    def __init__(self, config_path: str):
        # Load configuration
        self.config = load_config(config_path)
        
        # Initialize exchange
        self.exchange = self._initialize_exchange()
        
        # Initialize trading strategy
        self.strategy = TradingStrategy(self.config)
        
        # Initialize results storage
        self.results = {
            'trades': [],
            'equity_curve': [],
            'performance_metrics': {}
        }
    
    def _initialize_exchange(self) -> ccxt.Exchange:
        """Initialize the cryptocurrency exchange"""
        try:
            exchange_class = getattr(ccxt, self.config['exchange']['name'])
            exchange = exchange_class({
                'apiKey': self.config['exchange']['api_key'],
                'secret': self.config['exchange']['api_secret'],
                'enableRateLimit': True
            })
            
            # Test connection
            exchange.load_markets()
            logger.info(f"Successfully connected to {self.config['exchange']['name']}")
            
            return exchange
            
        except Exception as e:
            logger.error(f"Error initializing exchange: {str(e)}")
            raise
    
    def fetch_historical_data(self, 
                            symbol: str, 
                            timeframe: str, 
                            start_date: datetime,
                            end_date: datetime) -> pd.DataFrame:
        """Fetch historical market data"""
        try:
            # Convert dates to timestamps
            start_timestamp = int(start_date.timestamp() * 1000)
            end_timestamp = int(end_date.timestamp() * 1000)
            
            # Fetch OHLCV data
            ohlcv = self.exchange.fetch_ohlcv(
                symbol,
                timeframe=timeframe,
                since=start_timestamp,
                limit=1000  # Maximum limit per request
            )
            
            # Convert to DataFrame
            df = pd.DataFrame(
                ohlcv,
                columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
            )
            
            # Convert timestamp to datetime
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df.set_index('timestamp', inplace=True)
            
            return df
            
        except Exception as e:
            logger.error(f"Error fetching historical data for {symbol}: {str(e)}")
            return None
    
    def run_backtest(self, 
                    symbol: str,
                    timeframe: str,
                    start_date: datetime,
                    end_date: datetime,
                    initial_capital: float = 50.0) -> Dict:
        """Run backtest for a given period"""
        try:
            logger.info(f"Starting backtest for {symbol} from {start_date} to {end_date}")
            
            # Fetch historical data
            df = self.fetch_historical_data(symbol, timeframe, start_date, end_date)
            if df is None:
                raise ValueError("Failed to fetch historical data")
            
            # Initialize variables
            current_capital = initial_capital
            position = 0
            entry_price = 0
            trades = []
            equity_curve = []
            
            # Iterate through data
            for i in range(len(df)):
                current_data = df.iloc[:i+1]
                current_price = current_data['close'].iloc[-1]
                
                # Generate trading signal
                signal = self.strategy.analyze_market(symbol, current_data)
                
                if signal:
                    # Execute trade if conditions are met
                    if signal.action == 'buy' and position <= 0:
                        position = 1
                        entry_price = current_price
                        trades.append({
                            'timestamp': current_data.index[-1],
                            'action': 'buy',
                            'price': current_price,
                            'position_size': signal.position_size
                        })
                    
                    elif signal.action == 'sell' and position >= 0:
                        position = -1
                        entry_price = current_price
                        trades.append({
                            'timestamp': current_data.index[-1],
                            'action': 'sell',
                            'price': current_price,
                            'position_size': signal.position_size
                        })
                
                # Update position PnL
                if position != 0:
                    pnl = position * (current_price - entry_price) * signal.position_size
                    current_capital += pnl
                
                # Record equity
                equity_curve.append({
                    'timestamp': current_data.index[-1],
                    'equity': current_capital
                })
            
            # Calculate performance metrics
            self.results['trades'] = trades
            self.results['equity_curve'] = equity_curve
            self.results['performance_metrics'] = self._calculate_performance_metrics(trades)
            
            return self.results
            
        except Exception as e:
            logger.error(f"Error running backtest: {str(e)}")
            return None
    
    def _calculate_performance_metrics(self, trades: List[Dict]) -> Dict:
        """Calculate performance metrics from trades"""
        if not trades:
            return {}
        
        # Convert trades to DataFrame
        df = pd.DataFrame(trades)
        
        # Calculate basic metrics
        total_trades = len(df)
        winning_trades = len(df[df['action'] == 'buy'])
        losing_trades = len(df[df['action'] == 'sell'])
        win_rate = winning_trades / total_trades if total_trades > 0 else 0
        
        # Calculate returns
        returns = df['price'].pct_change()
        total_return = (1 + returns).prod() - 1
        annual_return = (1 + total_return) ** (252 / len(df)) - 1
        
        # Calculate risk metrics
        volatility = returns.std() * np.sqrt(252)
        sharpe_ratio = annual_return / volatility if volatility > 0 else 0
        
        # Calculate drawdown
        cumulative_returns = (1 + returns).cumprod()
        rolling_max = cumulative_returns.expanding().max()
        drawdowns = (cumulative_returns - rolling_max) / rolling_max
        max_drawdown = drawdowns.min()
        
        return {
            'total_trades': total_trades,
            'winning_trades': winning_trades,
            'losing_trades': losing_trades,
            'win_rate': win_rate,
            'total_return': total_return,
            'annual_return': annual_return,
            'volatility': volatility,
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown
        }
    
    def save_results(self, filepath: str):
        """Save backtest results to file"""
        try:
            with open(filepath, 'w') as f:
                json.dump(self.results, f, default=str)
            logger.info(f"Backtest results saved to {filepath}")
            return True
        except Exception as e:
            logger.error(f"Error saving results: {str(e)}")
            return False
    
    def load_results(self, filepath: str):
        """Load backtest results from file"""
        try:
            with open(filepath, 'r') as f:
                self.results = json.load(f)
            logger.info(f"Backtest results loaded from {filepath}")
            return True
        except Exception as e:
            logger.error(f"Error loading results: {str(e)}")
            return False

def main():
    # Load configuration
    config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.json')
    
    # Create backtester
    backtester = Backtester(config_path)
    
    # Define backtest parameters
    symbol = "BTC/USDT"
    timeframe = "1h"
    start_date = datetime.now() - timedelta(days=30)
    end_date = datetime.now()
    initial_capital = 50.0
    
    # Run backtest
    results = backtester.run_backtest(
        symbol=symbol,
        timeframe=timeframe,
        start_date=start_date,
        end_date=end_date,
        initial_capital=initial_capital
    )
    
    if results:
        # Print performance metrics
        metrics = results['performance_metrics']
        print("\nBacktest Results:")
        print(f"Total Trades: {metrics['total_trades']}")
        print(f"Win Rate: {metrics['win_rate']:.2%}")
        print(f"Total Return: {metrics['total_return']:.2%}")
        print(f"Annual Return: {metrics['annual_return']:.2%}")
        print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
        print(f"Max Drawdown: {metrics['max_drawdown']:.2%}")
        
        # Save results
        backtester.save_results('backtest_results.json')
    else:
        print("Backtest failed to complete")

if __name__ == "__main__":
    main()