File size: 7,952 Bytes
a147000 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | 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
|