backtest_engine.py

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