Backtesting & Simulation

Complete Backtesting Infrastructure

import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Tuple
from datetime import datetime, timedelta
from abc import ABC, abstractmethod
import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats

@dataclass
class Transaction:
    """Represents a single transaction in the simulation"""
    timestamp: datetime
    from_address: str
    to_address: str
    value: float
    gas_used: int
    gas_price: int
    nonce: int
    success: bool = True
    block_number: int = 0
    
    @property
    def gas_cost_eth(self) -> float:
        """Calculate gas cost in ETH"""
        return self.gas_used * self.gas_price / 1e18
    
    @property
    def gas_cost_usd(self) -> float:
        """Calculate gas cost in USD (requires ETH/USD price)"""
        return self.gas_cost_eth * self.eth_price_usd  # Would need real-time price

@dataclass
class Trade:
    """Represents a completed MEV trade"""
    timestamp: datetime
    strategy_name: str
    opportunity_type: str
    token_pair: str
    side: str  # 'buy' or 'sell'
    quantity: float
    price: float
    price_usd: float
    fees: float
    pnl: float = 0.0
    success: bool = True
    execution_time: float = 0.0  # milliseconds
    slippage: float = 0.0
    
    @property
    def notional_value(self) -> float:
        """Calculate notional value of trade"""
        return self.quantity * self.price_usd
    
    @property
    def gross_pnl(self) -> float:
        """PNL before fees"""
        return self.pnl + self.fees
    
    @property
    def return_pct(self) -> float:
        """Return percentage"""
        if self.notional_value > 0:
            return self.gross_pnl / self.notional_value
        return 0.0

@dataclass
class BacktestConfig:
    """Configuration for backtest execution"""
    start_date: datetime
    end_date: datetime
    initial_capital: float = 100000.0
    slippage_model: str = 'linear'  # 'linear', 'quadratic', 'custom'
    gas_model: str = 'realistic'   # 'fixed', 'dynamic', 'realistic'
    fee_tiers: Dict[str, float] = field(default_factory=lambda: {
        'DEX_v2': 0.003,
        'DEX_v3': 0.0005,
        'lending': 0.0005
    })
    max_position_size: float = 0.25  # 25% of portfolio max
    min_profit_threshold: float = 1.0  # $1 minimum profit
    max_trade_duration: int = 300  # seconds
    risk_free_rate: float = 0.02  # 2% annual risk-free rate

@dataclass
class BacktestResults:
    """Results from backtest execution"""
    trades: List[Trade] = field(default_factory=list)
    total_return: float = 0.0
    annual_return: float = 0.0
    volatility: float = 0.0
    sharpe_ratio: float = 0.0
    max_drawdown: float = 0.0
    calmar_ratio: float = 0.0
    win_rate: float = 0.0
    profit_factor: float = 0.0
    avg_trade_duration: float = 0.0
    total_trades: int = 0
    successful_trades: int = 0
    
    def to_dataframe(self) -> pd.DataFrame:
        """Convert trades to DataFrame for analysis"""
        if not self.trades:
            return pd.DataFrame()
        
        data = []
        for trade in self.trades:
            data.append({
                'timestamp': trade.timestamp,
                'strategy': trade.strategy_name,
                'type': trade.opportunity_type,
                'pair': trade.token_pair,
                'side': trade.side,
                'quantity': trade.quantity,
                'price': trade.price,
                'price_usd': trade.price_usd,
                'fees': trade.fees,
                'pnl': trade.pnl,
                'notional': trade.notional_value,
                'return_pct': trade.return_pct,
                'success': trade.success,
                'execution_time': trade.execution_time,
                'slippage': trade.slippage
            })
        
        return pd.DataFrame(data)

class MEVBacktester:
    """Comprehensive MEV backtesting framework"""
    
    def __init__(self, config: BacktestConfig):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.market_data = None
        self.gas_history = None
        self.liquidity_history = None
        
        # Performance tracking
        self.portfolio_value_history = []
        self.daily_returns = []
        self.trades_executed = []
        
    def load_market_data(self, data_source: str):
        """Load historical market data for backtesting"""
        self.logger.info(f"Loading market data from {data_source}")
        
        # This would typically load from databases or files
        # For demonstration, we'll create synthetic data
        self._generate_synthetic_data()
        
    def _generate_synthetic_data(self):
        """Generate synthetic market data for demonstration"""
        date_range = pd.date_range(
            start=self.config.start_date,
            end=self.config.end_date,
            freq='1min'
        )
        
        # Generate price data with realistic MEV patterns
        n_periods = len(date_range)
        
        # Base price movements
        base_prices = {
            'ETH_USD': 2000 + np.random.normal(0, 100, n_periods).cumsum(),
            'BTC_USD': 40000 + np.random.normal(0, 500, n_periods).cumsum(),
            'UNI_ETH': 0.005 + np.random.normal(0, 0.001, n_periods).cumsum()
        }
        
        # Add MEV opportunity patterns
        for pair, prices in base_prices.items():
            # Add periodic volatility spikes (MEV opportunities)
            spike_times = np.random.choice(n_periods, size=n_periods//100, replace=False)
            for spike_time in spike_times:
                if spike_time < len(prices):
                    prices[spike_time:min(spike_time+5, len(prices))] *= (1 + np.random.uniform(-0.05, 0.05))
        
        # Store as DataFrame
        self.market_data = pd.DataFrame({
            'timestamp': date_range,
            'ETH_USD': base_prices['ETH_USD'],
            'BTC_USD': base_prices['BTC_USD'],
            'UNI_ETH': base_prices['UNI_ETH']
        })
        
        # Generate gas price history
        self.gas_history = pd.DataFrame({
            'timestamp': date_range,
            'gas_price': np.random.lognormal(10, 0.5, n_periods) * 1e9,  # Gwei
            'gas_used_avg': np.random.normal(150000, 50000, n_periods)
        })
        
        # Generate liquidity data
        self.liquidity_history = pd.DataFrame({
            'timestamp': date_range,
            'uniswap_liquidity': np.random.normal(10000000, 2000000, n_periods),
            'sushiswap_liquidity': np.random.normal(5000000, 1000000, n_periods)
        })
    
    def run_backtest(self, strategy_func: Callable, parallel: bool = False) -> BacktestResults:
        """Execute backtest with given strategy function"""
        self.logger.info(f"Starting backtest from {self.config.start_date} to {self.config.end_date}")
        
        portfolio_value = self.config.initial_capital
        current_positions = {}
        
        for timestamp in self.market_data['timestamp']:
            # Get market state at this timestamp
            market_state = self._get_market_state(timestamp)
            
            # Execute strategy logic
            strategy_actions = strategy_func(market_state, current_positions, portfolio_value)
            
            # Execute actions and update portfolio
            portfolio_value, current_positions = self._execute_actions(
                strategy_actions, market_state, portfolio_value, current_positions, timestamp
            )
            
            # Track portfolio value
            self.portfolio_value_history.append({
                'timestamp': timestamp,
                'portfolio_value': portfolio_value,
                'positions': current_positions.copy()
            })
        
        # Calculate final results
        results = self._calculate_results()
        
        self.logger.info(f"Backtest completed: {results.total_trades} trades, {results.total_return:.2%} return")
        
        return results
    
    def _get_market_state(self, timestamp: datetime) -> Dict:
        """Get market state at specific timestamp"""
        # Get price data
        price_row = self.market_data[self.market_data['timestamp'] == timestamp]
        if price_row.empty:
            return {}
        
        prices = price_row.iloc[0].to_dict()
        
        # Get gas data
        gas_row = self.gas_history[self.gas_history['timestamp'] == timestamp]
        gas_info = gas_row.iloc[0].to_dict() if not gas_row.empty else {'gas_price': 50e9, 'gas_used_avg': 150000}
        
        # Get liquidity data
        liquidity_row = self.liquidity_history[self.liquidity_history['timestamp'] == timestamp]
        liquidity_info = liquidity_row.iloc[0].to_dict() if not liquidity_row.empty else {'uniswap_liquidity': 1e7, 'sushiswap_liquidity': 5e6}
        
        return {
            'timestamp': timestamp,
            'prices': prices,
            'gas': gas_info,
            'liquidity': liquidity_info,
            'block_number': int(timestamp.timestamp()) // 15  # Approximate block number
        }
    
    def _execute_actions(self, actions: List[Dict], market_state: Dict, 
                        portfolio_value: float, positions: Dict, timestamp: datetime) -> Tuple[float, Dict]:
        """Execute strategy actions in simulation"""
        
        for action in actions:
            if action['type'] == 'arbitrage':
                result = self._execute_arbitrage(action, market_state, portfolio_value, timestamp)
                if result.success:
                    portfolio_value += result.pnl
                    self.trades_executed.append(result)
            
            elif action['type'] == 'liquidation':
                result = self._execute_liquidation(action, market_state, timestamp)
                if result.success:
                    portfolio_value += result.pnl
                    self.trades_executed.append(result)
        
        return portfolio_value, positions
    
    def _execute_arbitrage(self, action: Dict, market_state: Dict, portfolio_value: float, 
                          timestamp: datetime) -> Trade:
        """Execute arbitrage simulation"""
        pair = action['pair']
        buy_dex = action['buy_dex']
        sell_dex = action['sell_dex']
        quantity = action['quantity']
        
        # Get prices from market state
        prices = market_state['prices']
        
        # Find price for pair (simplified)
        if pair == 'ETH_USD':
            buy_price = prices.get('ETH_USD', 2000)
            sell_price = buy_price  # Same DEX for simplicity
        else:
            buy_price = prices.get(pair, 2000)
            sell_price = buy_price
        
        # Simulate price difference (opportunity)
        price_diff_pct = np.random.uniform(0.002, 0.02)  # 0.2% to 2% difference
        if np.random.random() > 0.7:  # 30% chance of actual opportunity
            sell_price *= (1 + price_diff_pct)
        
        # Calculate transaction details
        buy_cost = quantity * buy_price
        sell_proceeds = quantity * sell_price
        
        # Apply slippage
        slippage_pct = self._calculate_slippage(pair, quantity, market_state)
        actual_sell_price = sell_price * (1 - slippage_pct)
        actual_sell_proceeds = quantity * actual_sell_price
        
        # Calculate fees and gas
        gas_price = market_state['gas']['gas_price']
        estimated_gas = 200000  # Arbitrage gas estimate
        gas_cost = estimated_gas * gas_price / 1e18 * prices.get('ETH_USD', 2000)  # Convert to USD
        
        dex_fee_rate = self.config.fee_tiers.get(buy_dex, 0.003)
        fees = buy_cost * dex_fee_rate + actual_sell_proceeds * dex_fee_rate
        
        # Calculate P&L
        gross_pnl = actual_sell_proceeds - buy_cost
        net_pnl = gross_pnl - fees - gas_cost
        
        # Success criteria
        success = net_pnl > self.config.min_profit_threshold
        
        return Trade(
            timestamp=timestamp,
            strategy_name='arbitrage_strategy',
            opportunity_type='arbitrage',
            token_pair=pair,
            side='both',
            quantity=quantity,
            price=buy_price,
            price_usd=buy_cost,
            fees=fees,
            pnl=net_pnl,
            success=success,
            execution_time=np.random.uniform(50, 200),  # milliseconds
            slippage=slippage_pct
        )
    
    def _execute_liquidation(self, action: Dict, market_state: Dict, timestamp: datetime) -> Trade:
        """Execute liquidation simulation"""
        protocol = action['protocol']
        position = action['position']
        
        # Simulate liquidation outcome
        health_factor = np.random.uniform(0.8, 1.2)
        success = health_factor < 1.0
        
        if success:
            # Calculate liquidation reward
            debt_value = position.get('debt_value', 10000)
            liquidation_bonus = 0.05  # 5% bonus
            reward = debt_value * liquidation_bonus
            
            # Gas cost
            gas_cost = 150000 * market_state['gas']['gas_price'] / 1e18 * market_state['prices'].get('ETH_USD', 2000)
            
            net_pnl = reward - gas_cost
        else:
            net_pnl = -1000  # Cost of failed liquidation attempt
        
        return Trade(
            timestamp=timestamp,
            strategy_name='liquidation_strategy',
            opportunity_type='liquidation',
            token_pair=position.get('token', 'ETH'),
            side='liquidation',
            quantity=position.get('debt_value', 10000),
            price=1.0,
            price_usd=position.get('debt_value', 10000),
            fees=0,
            pnl=net_pnl,
            success=success,
            execution_time=np.random.uniform(100, 300)
        )
    
    def _calculate_slippage(self, pair: str, quantity: float, market_state: Dict) -> float:
        """Calculate slippage for given trade"""
        liquidity = market_state['liquidity']['uniswap_liquidity']
        
        if self.config.slippage_model == 'linear':
            # Linear slippage model
            liquidity_utilization = quantity / liquidity
            return min(0.05, liquidity_utilization * 0.1)  # Max 5% slippage
        
        elif self.config.slippage_model == 'quadratic':
            # Quadratic slippage model
            liquidity_utilization = quantity / liquidity
            return min(0.05, (liquidity_utilization ** 2) * 0.2)
        
        else:
            # Custom or realistic model
            base_slippage = 0.001
            volume_impact = (quantity / liquidity) * 0.1
            return min(0.05, base_slippage + volume_impact)
    
    def _calculate_results(self) -> BacktestResults:
        """Calculate comprehensive backtest results"""
        if not self.trades_executed:
            return BacktestResults()
        
        # Convert to DataFrame for analysis
        trades_df = pd.DataFrame([
            {
                'timestamp': trade.timestamp,
                'pnl': trade.pnl,
                'notional': trade.notional_value,
                'fees': trade.fees,
                'success': trade.success,
                'return_pct': trade.return_pct
            }
            for trade in self.trades_executed
        ])
        
        if trades_df.empty:
            return BacktestResults()
        
        # Basic metrics
        total_trades = len(trades_df)
        successful_trades = trades_df['success'].sum()
        win_rate = successful_trades / total_trades if total_trades > 0 else 0
        
        # P&L metrics
        total_pnl = trades_df['pnl'].sum()
        total_fees = trades_df['fees'].sum()
        initial_capital = self.config.initial_capital
        total_return = total_pnl / initial_capital
        
        # Annualized metrics
        days = (self.config.end_date - self.config.start_date).days
        annual_return = ((1 + total_return) ** (365 / days)) - 1 if days > 0 else 0
        
        # Portfolio value series for drawdown calculation
        portfolio_values = [initial_capital]
        current_value = initial_capital
        
        for _, trade in trades_df.iterrows():
            current_value += trade['pnl']
            portfolio_values.append(current_value)
        
        portfolio_series = pd.Series(portfolio_values)
        
        # Drawdown calculation
        peak = portfolio_series.expanding().max()
        drawdown = (portfolio_series - peak) / peak
        max_drawdown = drawdown.min()
        
        # Risk metrics
        trades_df['daily_return'] = trades_df['pnl'] / initial_capital
        volatility = trades_df['daily_return'].std() * np.sqrt(365)
        
        # Sharpe ratio (using risk-free rate)
        excess_return = annual_return - self.config.risk_free_rate
        sharpe_ratio = excess_return / volatility if volatility > 0 else 0
        
        # Calmar ratio
        calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown < 0 else 0
        
        # Profit factor
        winning_trades = trades_df[trades_df['pnl'] > 0]
        losing_trades = trades_df[trades_df['pnl'] < 0]
        
        gross_profit = winning_trades['pnl'].sum() if not winning_trades.empty else 0
        gross_loss = abs(losing_trades['pnl'].sum()) if not losing_trades.empty else 0
        profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
        
        # Average trade duration (would need timing data)
        avg_trade_duration = 0  # Placeholder
        
        return BacktestResults(
            trades=self.trades_executed,
            total_return=total_return,
            annual_return=annual_return,
            volatility=volatility,
            sharpe_ratio=sharpe_ratio,
            max_drawdown=max_drawdown,
            calmar_ratio=calmar_ratio,
            win_rate=win_rate,
            profit_factor=profit_factor,
            avg_trade_duration=avg_trade_duration,
            total_trades=total_trades,
            successful_trades=successful_trades
        )
    
    def monte_carlo_simulation(self, strategy_func: Callable, n_simulations: int = 1000) -> Dict:
        """Run Monte Carlo simulation for strategy validation"""
        self.logger.info(f"Running Monte Carlo simulation with {n_simulations} simulations")
        
        results = []
        
        for sim in range(n_simulations):
            # Add random noise to market data for this simulation
            self._add_noise_to_data(sim)
            
            # Run backtest
            sim_results = self.run_backtest(strategy_func)
            
            results.append({
                'simulation': sim,
                'total_return': sim_results.total_return,
                'sharpe_ratio': sim_results.sharpe_ratio,
                'max_drawdown': sim_results.max_drawdown,
                'win_rate': sim_results.win_rate
            })
        
        # Remove noise and restore original data
        self._restore_original_data()
        
        # Analyze results
        results_df = pd.DataFrame(results)
        
        return {
            'mean_return': results_df['total_return'].mean(),
            'return_std': results_df['total_return'].std(),
            'mean_sharpe': results_df['sharpe_ratio'].mean(),
            'sharpe_std': results_df['sharpe_ratio'].std(),
            'mean_drawdown': results_df['max_drawdown'].mean(),
            'drawdown_std': results_df['max_drawdown'].std(),
            'probability_positive': (results_df['total_return'] > 0).mean(),
            'probability_sharpe_above_1': (results_df['sharpe_ratio'] > 1).mean(),
            'var_95': results_df['total_return'].quantile(0.05),
            'cvar_95': results_df[results_df['total_return'] <= results_df['total_return'].quantile(0.05)]['total_return'].mean()
        }
    
    def stress_test(self, strategy_func: Callable, stress_scenarios: List[Dict]) -> Dict:
        """Run stress tests with various market scenarios"""
        stress_results = {}
        
        for scenario_name, scenario_config in stress_scenarios.items():
            self.logger.info(f"Running stress test: {scenario_name}")
            
            # Apply scenario modifications
            self._apply_scenario(scenario_config)
            
            # Run backtest
            scenario_results = self.run_backtest(strategy_func)
            
            stress_results[scenario_name] = {
                'total_return': scenario_results.total_return,
                'sharpe_ratio': scenario_results.sharpe_ratio,
                'max_drawdown': scenario_results.max_drawdown,
                'total_trades': scenario_results.total_trades
            }
            
            # Restore original data
            self._restore_original_data()
        
        return stress_results
    
    def _add_noise_to_data(self, seed: int):
        """Add random noise to market data for Monte Carlo"""
        np.random.seed(seed)
        
        # Add noise to prices
        for col in ['ETH_USD', 'BTC_USD', 'UNI_ETH']:
            if col in self.market_data.columns:
                noise = np.random.normal(0, 0.001, len(self.market_data))
                self.market_data[col] += noise * self.market_data[col]
    
    def _restore_original_data(self):
        """Restore original market data"""
        # In a real implementation, you'd save and restore the original data
        self._generate_synthetic_data()
    
    def _apply_scenario(self, scenario_config: Dict):
        """Apply stress scenario to market data"""
        # Implementation would modify market data based on scenario
        pass

# Example Strategy Functions
def sample_arbitrage_strategy(market_state: Dict, positions: Dict, portfolio_value: float) -> List[Dict]:
    """Sample arbitrage strategy for testing"""
    actions = []
    
    # Check for arbitrage opportunities
    if np.random.random() > 0.95:  # 5% chance per iteration
        actions.append({
            'type': 'arbitrage',
            'pair': 'ETH_USD',
            'quantity': min(1000, portfolio_value * 0.01),  # 1% of portfolio
            'buy_dex': 'uniswap',
            'sell_dex': 'sushiswap'
        })
    
    return actions

def sample_liquidation_strategy(market_state: Dict, positions: Dict, portfolio_value: float) -> List[Dict]:
    """Sample liquidation strategy for testing"""
    actions = []
    
    # Random liquidation attempts
    if np.random.random() > 0.98:  # 2% chance per iteration
        actions.append({
            'type': 'liquidation',
            'protocol': 'aave',
            'position': {
                'debt_value': 5000,
                'token': 'ETH'
            }
        })
    
    return actions

# Usage Example
if __name__ == "__main__":
    # Configure backtest
    config = BacktestConfig(
        start_date=datetime(2024, 1, 1),
        end_date=datetime(2024, 3, 31),  # 3 months
        initial_capital=100000,
        slippage_model='linear',
        gas_model='realistic'
    )
    
    # Initialize backtester
    backtester = MEVBacktester(config)
    backtester.load_market_data("synthetic_data")
    
    # Run backtest
    results = backtester.run_backtest(sample_arbitrage_strategy)
    
    print(f"Backtest Results:")
    print(f"Total Return: {results.total_return:.2%}")
    print(f"Sharpe Ratio: {results.sharpe_ratio:.2f}")
    print(f"Max Drawdown: {results.max_drawdown:.2%}")
    print(f"Win Rate: {results.win_rate:.2%}")
    print(f"Total Trades: {results.total_trades}")
    print(f"Profit Factor: {results.profit_factor:.2f}")
    
    # Run Monte Carlo simulation
    mc_results = backtester.monte_carlo_simulation(sample_arbitrage_strategy, n_simulations=100)
    
    print(f"\nMonte Carlo Results:")
    print(f"Mean Return: {mc_results['mean_return']:.2%}")
    print(f"Return Std Dev: {mc_results['return_std']:.2%}")
    print(f"Probability Positive: {mc_results['probability_positive']:.2%}")
    print(f"VaR (95%): {mc_results['var_95']:.2%}")
    
    # Run stress tests
    stress_scenarios = {
        'high_gas': {'gas_multiplier': 3.0},
        'low_liquidity': {'liquidity_reduction': 0.5},
        'high_volatility': {'volatility_multiplier': 2.0}
    }
    
    stress_results = backtester.stress_test(sample_arbitrage_strategy, stress_scenarios)
    
    print(f"\nStress Test Results:")
    for scenario, result in stress_results.items():
        print(f"{scenario}: Return = {result['total_return']:.2%}, Sharpe = {result['sharpe_ratio']:.2f}")
    
    # Generate visualizations
    trades_df = results.to_dataframe()
    if not trades_df.empty:
        plt.figure(figsize=(15, 10))
        
        # Portfolio value over time
        plt.subplot(2, 3, 1)
        if backtester.portfolio_value_history:
            timestamps = [p['timestamp'] for p in backtester.portfolio_value_history]
            values = [p['portfolio_value'] for p in backtester.portfolio_value_history]
            plt.plot(timestamps, values)
            plt.title('Portfolio Value Over Time')
            plt.xticks(rotation=45)
        
        # P&L distribution
        plt.subplot(2, 3, 2)
        plt.hist(trades_df['pnl'], bins=50, alpha=0.7)
        plt.title('P&L Distribution')
        plt.xlabel('P&L ($)')
        
        # Cumulative returns
        plt.subplot(2, 3, 3)
        cumulative_pnl = trades_df['pnl'].cumsum()
        plt.plot(cumulative_pnl)
        plt.title('Cumulative P&L')
        plt.xlabel('Trade Number')
        
        # Drawdown
        plt.subplot(2, 3, 4)
        portfolio_series = pd.Series([config.initial_capital] + cumulative_pnl.tolist())
        peak = portfolio_series.expanding().max()
        drawdown = (portfolio_series - peak) / peak
        plt.fill_between(range(len(drawdown)), drawdown, 0, alpha=0.3)
        plt.title('Drawdown')
        plt.xlabel('Time')
        plt.ylabel('Drawdown %')
        
        # Win rate by month
        plt.subplot(2, 3, 5)
        monthly_stats = trades_df.groupby(trades_df['timestamp'].dt.month).agg({
            'pnl': 'sum',
            'success': 'mean'
        })
        plt.bar(monthly_stats.index, monthly_stats['success'])
        plt.title('Monthly Win Rate')
        plt.xlabel('Month')
        plt.ylabel('Win Rate')
        
        # Return vs Risk
        plt.subplot(2, 3, 6)
        monthly_returns = trades_df.groupby(trades_df['timestamp'].dt.month)['return_pct'].apply(list)
        for month, returns in monthly_returns.items():
            plt.scatter(len(returns), np.mean(returns), s=50, label=f'Month {month}')
        plt.title('Monthly Return vs Trade Count')
        plt.xlabel('Number of Trades')
        plt.ylabel('Average Return %')
        
        plt.tight_layout()
        plt.show()