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