engine/simulator.py

import logging
import pandas as pd
import vectorbt as vbt
from typing import Dict, Any, Optional

logger = logging.getLogger("bqe.engine.simulator")

class VectorSimulator:
    """
    Parallelized, vectorized portfolio simulation engine using vectorbt.
    Runs backtests over asset panels and computes standard performance metrics.
    """
    
    def __init__(self, init_cash: float = 100000.0):
        """
        Initializes the simulator with default configuration.
        
        Args:
            init_cash: The starting capital for each asset/portfolio.
        """
        self.init_cash = init_cash

    def calculate_turnover(self, pf: vbt.Portfolio, total_days: int) -> pd.Series:
        """
        Computes the estimated mean daily portfolio turnover rate for each asset.
        
        Turnover = (Total value of all trades / 2) / (Average portfolio value * total days)
        
        Args:
            pf: The executed vectorbt Portfolio object.
            total_days: Number of days in backtest period.
            
        Returns:
            pd.Series: Daily turnover rate per symbol.
        """
        turnover_series = pd.Series(0.0, index=pf.wrapper.columns)
        
        try:
            orders_df = pf.orders.records_readable.copy()
            if orders_df.empty:
                return turnover_series
                
            # Explicitly compute Value if not present in the columns
            if 'Value' not in orders_df.columns:
                orders_df['Value'] = orders_df['Size'] * orders_df['Price']
                
            # Find the column that represents the symbol or asset grouping
            col_key = None
            for k in ['Column', 'Symbol', 'column', 'symbol']:
                if k in orders_df.columns:
                    col_key = k
                    break
                    
            if col_key:
                # Group total traded value (Size * Price) by asset/column
                grouped_value = orders_df.groupby(col_key)['Value'].sum()
                avg_portfolio_value = pf.value().mean()
                
                for col_name in pf.wrapper.columns:
                    val = grouped_value.get(col_name, 0.0)
                    if val == 0.0:
                        # Try indexing by integer position if column name is represented as an index
                        try:
                            col_idx = list(pf.wrapper.columns).index(col_name)
                            val = grouped_value.get(col_idx, 0.0)
                        except ValueError:
                            pass
                            
                    avg_val = avg_portfolio_value.get(col_name, self.init_cash)
                    if avg_val <= 0:
                        avg_val = self.init_cash
                        
                    # Estimated daily turnover: (total trade value / 2) / (average value * total days)
                    turnover_series[col_name] = (val / 2.0) / (avg_val * total_days)
                    
        except Exception as e:
            logger.warning(f"Failed to calculate turnover from orders: {e}. Using trades fallback.")
            try:
                trades_df = pf.trades.records_readable.copy()
                if not trades_df.empty:
                    col_key = next((k for k in ['Column', 'Symbol', 'column', 'symbol'] if k in trades_df.columns), None)
                    if col_key:
                        # Find entry price and exit price keys dynamically
                        entry_price_key = next((k for k in ['Avg Entry Price', 'Entry Price'] if k in trades_df.columns), 'Price')
                        exit_price_key = next((k for k in ['Avg Exit Price', 'Exit Price'] if k in trades_df.columns), entry_price_key)
                        
                        # Approximate traded value: Entry value + Exit value
                        trades_df['TradeValue'] = trades_df['Size'] * (
                            trades_df[entry_price_key] + trades_df[exit_price_key].fillna(trades_df[entry_price_key])
                        )
                        grouped_val = trades_df.groupby(col_key)['TradeValue'].sum()
                        avg_portfolio_value = pf.value().mean()
                        
                        for col_name in pf.wrapper.columns:
                            val = grouped_val.get(col_name, 0.0)
                            avg_val = avg_portfolio_value.get(col_name, self.init_cash)
                            turnover_series[col_name] = (val / 2.0) / (avg_val * total_days)
            except Exception as ex:
                logger.error(f"Fallback turnover calculation failed: {ex}")
                
        return turnover_series

    def run_backtest(
        self, 
        close_matrix: pd.DataFrame, 
        entry_matrix: pd.DataFrame, 
        exit_matrix: pd.DataFrame, 
        execution_delay: int = 1,
        freq: str = 'd'
    ) -> Dict[str, Any]:
        """
        Executes a vectorized signal-based portfolio backtest.
        
        To prevent look-ahead bias, the entry and exit signal matrices are shifted
        by the `execution_delay` parameter.
        
        Args:
            close_matrix: Wide DataFrame of asset close prices.
            entry_matrix: Boolean DataFrame of entry signals.
            exit_matrix: Boolean DataFrame of exit signals.
            execution_delay: Step delay before executing signals (default = 1).
            freq: Index frequency string for vectorbt annualization (default = 'd').
            
        Returns:
            dict: Cleaned and structured performance metrics dictionary.
        """
        logger.info(f"Preparing backtest simulation with execution_delay={execution_delay} and freq={freq}...")
        
        # Enforce look-ahead bias prevention by shifting signals
        shifted_entries = entry_matrix.shift(execution_delay).fillna(False).astype(bool)
        shifted_exits = exit_matrix.shift(execution_delay).fillna(False).astype(bool)
        
        total_days = len(close_matrix)
        
        logger.info("Executing backtest via vectorbt...")
        # Run backtest simultaneously across all symbols
        pf = vbt.Portfolio.from_signals(
            close=close_matrix,
            entries=shifted_entries,
            exits=shifted_exits,
            init_cash=self.init_cash,
            freq=freq
        )
        
        logger.info("Backtest execution completed. Computing performance metrics...")
        
        # Extract individual metrics per asset
        total_returns = pf.total_return() * 100        # convert to %
        sharpe_ratios = pf.sharpe_ratio()              # annualized Sharpe
        max_drawdowns = pf.max_drawdown() * 100        # convert to %
        win_rates = pf.trades.win_rate() * 100         # convert to %
        
        # Compute daily turnovers per asset
        turnovers = self.calculate_turnover(pf, total_days)

        # Standardize return dictionary
        metrics_payload = {
            "total_return": {
                "portfolio_average": float(total_returns.mean()),
                "per_symbol": total_returns.to_dict()
            },
            "sharpe_ratio": {
                "portfolio_average": float(sharpe_ratios.mean()) if not pd.isna(sharpe_ratios.mean()) else 0.0,
                "per_symbol": sharpe_ratios.to_dict()
            },
            "max_drawdown": {
                "portfolio_average": float(max_drawdowns.mean()),
                "per_symbol": max_drawdowns.to_dict()
            },
            "turnover": {
                "portfolio_average": float(turnovers.mean()),
                "per_symbol": turnovers.to_dict()
            },
            "win_rate": {
                "portfolio_average": float(win_rates.fillna(0.0).mean()),
                "per_symbol": win_rates.fillna(0.0).to_dict()
            }
        }
        
        return metrics_payload