Add backtest engine with Sharpe, Sortino, max drawdown, IC tracking

backtest_engine.py

@@ -0,0 +1,338 @@
+"""Backtest Engine for AlphaForge with comprehensive metrics."""
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Optional, Callable
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class BacktestEngine:
+    """Portfolio backtest engine with transaction costs and slippage"""
+    
+    def __init__(self, 
+                 initial_capital: float = 1_000_000,
+                 transaction_cost: float = 0.0003,
+                 slippage: float = 0.0001,
+                 benchmark: str = 'SPY'):
+        self.initial_capital = initial_capital
+        self.transaction_cost = transaction_cost
+        self.slippage = slippage
+        self.benchmark = benchmark
+        
+        self.portfolio_values = []
+        self.weights_history = []
+        self.returns_history = []
+        self.dates = []
+        self.trades = []
+        
+    def run_backtest(self,
+                     returns_df: pd.DataFrame,
+                     weights_df: pd.DataFrame,
+                     rebalance_dates: Optional[List[pd.Timestamp]] = None) -> Dict:
+        """
+        Run portfolio backtest
+        
+        Args:
+            returns_df: DataFrame of asset returns (dates x assets)
+            weights_df: DataFrame of portfolio weights (dates x assets)
+            rebalance_dates: List of dates to rebalance (if None, rebalance daily)
+        
+        Returns:
+            Dict with performance metrics
+        """
+        # Align dates
+        common_dates = returns_df.index.intersection(weights_df.index)
+        returns_df = returns_df.loc[common_dates]
+        weights_df = weights_df.loc[common_dates]
+        
+        capital = self.initial_capital
+        current_weights = np.zeros(len(returns_df.columns))
+        portfolio_values = [capital]
+        
+        for i, date in enumerate(common_dates[1:], 1):
+            # Get target weights
+            target_weights = weights_df.iloc[i].values
+            
+            # Check if rebalance needed
+            if rebalance_dates is None or date in rebalance_dates:
+                # Calculate turnover
+                turnover = np.sum(np.abs(target_weights - current_weights))
+                
+                # Transaction costs
+                tc = turnover * self.transaction_cost * capital
+                capital -= tc
+                
+                # Record trade
+                if turnover > 0.001:
+                    self.trades.append({
+                        'date': date,
+                        'turnover': turnover,
+                        'cost': tc,
+                        'old_weights': current_weights.copy(),
+                        'new_weights': target_weights.copy()
+                    })
+                
+                current_weights = target_weights.copy()
+            
+            # Apply slippage to returns
+            daily_returns = returns_df.iloc[i].values
+            slippage_cost = np.sum(np.abs(current_weights)) * self.slippage
+            
+            # Portfolio return
+            port_return = np.dot(current_weights, daily_returns) - slippage_cost
+            capital *= (1 + port_return)
+            
+            portfolio_values.append(capital)
+            self.returns_history.append(port_return)
+            self.weights_history.append(current_weights.copy())
+            self.dates.append(date)
+        
+        self.portfolio_values = np.array(portfolio_values)
+        self.returns_history = np.array(self.returns_history)
+        
+        return self.compute_metrics()
+    
+    def compute_metrics(self, benchmark_returns: Optional[np.ndarray] = None) -> Dict:
+        """Compute comprehensive performance metrics"""
+        returns = self.returns_history
+        
+        if len(returns) == 0:
+            return {}
+        
+        # Basic metrics
+        total_return = (self.portfolio_values[-1] / self.initial_capital) - 1
+        annualized_return = (1 + total_return) ** (252 / len(returns)) - 1
+        
+        # Volatility
+        volatility = np.std(returns) * np.sqrt(252)
+        
+        # Sharpe ratio
+        excess_returns = returns - 0.04 / 252  # Assuming 4% risk-free rate
+        sharpe = np.mean(excess_returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0
+        
+        # Sortino ratio
+        downside_returns = returns[returns < 0]
+        downside_std = np.std(downside_returns) * np.sqrt(252) if len(downside_returns) > 0 else 1e-8
+        sortino = (annualized_return - 0.04) / downside_std
+        
+        # Max drawdown
+        cumulative = np.cumprod(1 + returns)
+        running_max = np.maximum.accumulate(cumulative)
+        drawdown = (cumulative - running_max) / running_max
+        max_drawdown = np.min(drawdown)
+        
+        # Calmar ratio
+        calmar = annualized_return / abs(max_drawdown) if max_drawdown != 0 else 0
+        
+        # Win rate
+        win_rate = np.sum(returns > 0) / len(returns)
+        
+        # Profit factor
+        gross_profit = np.sum(returns[returns > 0])
+        gross_loss = abs(np.sum(returns[returns < 0]))
+        profit_factor = gross_profit / gross_loss if gross_loss > 0 else np.inf
+        
+        # Alpha and Beta (vs benchmark)
+        alpha, beta = 0, 0
+        if benchmark_returns is not None and len(benchmark_returns) == len(returns):
+            cov = np.cov(returns, benchmark_returns)[0, 1]
+            bench_var = np.var(benchmark_returns)
+            beta = cov / bench_var if bench_var > 0 else 0
+            alpha = (np.mean(returns) - beta * np.mean(benchmark_returns)) * 252
+        
+        # Information ratio
+        if benchmark_returns is not None:
+            tracking_error = np.std(returns - benchmark_returns) * np.sqrt(252)
+            info_ratio = (annualized_return - np.mean(benchmark_returns) * 252) / tracking_error if tracking_error > 0 else 0
+        else:
+            info_ratio = 0
+        
+        # Turnover statistics
+        avg_turnover = np.mean([t['turnover'] for t in self.trades]) if self.trades else 0
+        total_cost = sum([t['cost'] for t in self.trades]) if self.trades else 0
+        
+        metrics = {
+            'total_return': total_return,
+            'annualized_return': annualized_return,
+            'volatility': volatility,
+            'sharpe_ratio': sharpe,
+            'sortino_ratio': sortino,
+            'max_drawdown': max_drawdown,
+            'calmar_ratio': calmar,
+            'win_rate': win_rate,
+            'profit_factor': profit_factor,
+            'alpha': alpha,
+            'beta': beta,
+            'information_ratio': info_ratio,
+            'avg_turnover': avg_turnover,
+            'total_transaction_costs': total_cost,
+            'final_capital': self.portfolio_values[-1],
+            'n_trades': len(self.trades),
+            'n_days': len(returns)
+        }
+        
+        return metrics
+    
+    def get_equity_curve(self) -> pd.DataFrame:
+        """Get equity curve"""
+        return pd.DataFrame({
+            'date': [self.dates[0]] + list(self.dates),
+            'portfolio_value': self.portfolio_values,
+            'cumulative_return': (self.portfolio_values / self.initial_capital) - 1
+        })
+    
+    def get_drawdown_series(self) -> pd.Series:
+        """Get drawdown series"""
+        cumulative = np.cumprod(1 + self.returns_history)
+        running_max = np.maximum.accumulate(cumulative)
+        drawdown = (cumulative - running_max) / running_max
+        return pd.Series(drawdown, index=self.dates)
+    
+    def get_monthly_returns(self) -> pd.DataFrame:
+        """Get monthly returns"""
+        returns_series = pd.Series(self.returns_history, index=self.dates)
+        monthly = returns_series.resample('M').apply(lambda x: np.prod(1 + x) - 1)
+        return monthly
+    
+    def get_rolling_metrics(self, window: int = 63) -> pd.DataFrame:
+        """Get rolling performance metrics"""
+        returns_series = pd.Series(self.returns_history, index=self.dates)
+        
+        rolling_sharpe = (
+            returns_series.rolling(window).mean() / 
+            returns_series.rolling(window).std() * np.sqrt(252)
+        )
+        
+        rolling_vol = returns_series.rolling(window).std() * np.sqrt(252)
+        
+        return pd.DataFrame({
+            'rolling_sharpe': rolling_sharpe,
+            'rolling_volatility': rolling_vol
+        })
+
+
+def compute_information_coefficient(predictions: pd.Series, 
+                                     actuals: pd.Series,
+                                     by_date: bool = True) -> Dict:
+    """
+    Compute Information Coefficient (rank correlation)
+    
+    Args:
+        predictions: Series of predicted returns
+        actuals: Series of actual returns
+        by_date: If True, compute IC per date and return mean/std
+    
+    Returns:
+        Dict with IC metrics
+    """
+    from scipy.stats import spearmanr
+    
+    if by_date and hasattr(predictions, 'index') and hasattr(actuals, 'index'):
+        # Group by date
+        ic_by_date = []
+        
+        pred_df = pd.DataFrame({'pred': predictions, 'actual': actuals})
+        pred_df = pred_df.dropna()
+        
+        if hasattr(pred_df.index, 'date'):
+            dates = pred_df.index.date
+        else:
+            dates = pred_df.index
+        
+        for date in np.unique(dates):
+            mask = dates == date
+            if mask.sum() > 3:
+                p = pred_df.loc[mask, 'pred']
+                a = pred_df.loc[mask, 'actual']
+                ic, _ = spearmanr(p, a)
+                if not np.isnan(ic):
+                    ic_by_date.append(ic)
+        
+        if len(ic_by_date) > 0:
+            return {
+                'mean_ic': np.mean(ic_by_date),
+                'ic_std': np.std(ic_by_date),
+                'ic_ir': np.mean(ic_by_date) / np.std(ic_by_date) if np.std(ic_by_date) > 0 else 0,
+                'ic_pct_positive': np.sum(np.array(ic_by_date) > 0) / len(ic_by_date),
+                'n_periods': len(ic_by_date)
+            }
+    
+    # Overall IC
+    mask = ~(np.isnan(predictions) | np.isnan(actuals))
+    ic, pvalue = spearmanr(predictions[mask], actuals[mask])
+    
+    return {
+        'mean_ic': ic if not np.isnan(ic) else 0,
+        'ic_std': 0,
+        'ic_ir': 0,
+        'ic_pct_positive': 1 if ic > 0 else 0,
+        'n_periods': 1,
+        'p_value': pvalue
+    }
+
+
+class RegimeDetector:
+    """Detect market regimes using Hidden Markov Model or simple heuristics"""
+    
+    def __init__(self, method: str = 'simple'):
+        self.method = method
+        self.regimes = []
+        
+    def detect_regimes(self, returns: pd.Series, 
+                       volatility_window: int = 21) -> pd.Series:
+        """
+        Detect market regimes:
+        - Bull: positive trend, low vol
+        - Bear: negative trend, high vol
+        - High Vol: high volatility regardless of trend
+        """
+        # Trend
+        trend = returns.rolling(63).mean()
+        
+        # Volatility
+        vol = returns.rolling(volatility_window).std() * np.sqrt(252)
+        vol_median = vol.median()
+        
+        regimes = pd.Series(index=returns.index, dtype='object')
+        
+        for i, date in enumerate(returns.index):
+            if pd.isna(trend.loc[date]) or pd.isna(vol.loc[date]):
+                regimes.loc[date] = 'unknown'
+                continue
+            
+            t = trend.loc[date]
+            v = vol.loc[date]
+            
+            if v > vol_median * 1.5:
+                regimes.loc[date] = 'high_vol'
+            elif t > 0.001:
+                regimes.loc[date] = 'bull'
+            elif t < -0.001:
+                regimes.loc[date] = 'bear'
+            else:
+                regimes.loc[date] = 'neutral'
+        
+        self.regimes = regimes
+        return regimes
+    
+    def get_regime_stats(self, returns: pd.Series) -> pd.DataFrame:
+        """Get performance statistics by regime"""
+        if len(self.regimes) == 0:
+            self.detect_regimes(returns)
+        
+        stats = []
+        for regime in self.regimes.unique():
+            mask = self.regimes == regime
+            regime_returns = returns[mask]
+            
+            if len(regime_returns) > 0:
+                stats.append({
+                    'regime': regime,
+                    'n_days': len(regime_returns),
+                    'mean_return': regime_returns.mean() * 252,
+                    'volatility': regime_returns.std() * np.sqrt(252),
+                    'sharpe': (regime_returns.mean() / regime_returns.std()) * np.sqrt(252) if regime_returns.std() > 0 else 0,
+                    'max_drawdown': (regime_returns.cumsum() - regime_returns.cumsum().cummax()).min()
+                })
+        
+        return pd.DataFrame(stats)