diegobeyl/backtesting
2
1"""2Motor de backtesting multi-activo3Capital compartido, hasta 9 posiciones simultáneas4"""5 6from dataclasses import dataclass, field7from typing import List, Optional, Dict, Any, Tuple8from datetime import datetime9import pandas as pd10import numpy as np11import sys12from pathlib import Path13 14# Add parent directory for imports15sys.path.insert(0, str(Path(__file__).parent.parent.parent))16 17from backtesting_app.algorithms.base import BaseAlgorithm, AlgorithmResult, TradeSignal, TrendState, AlgorithmState18 19 20@dataclass21class MultiTrade:22 """Representa una operación en el backtest multi-activo"""23 symbol: str24 entry_date: datetime25 entry_price: float26 direction: str # 'LONG' or 'SHORT'27 initial_sl: float28 position_size: float29 capital_at_entry: float30 exit_date: Optional[datetime] = None31 exit_price: Optional[float] = None32 pnl: float = 0.033 pnl_percent: float = 0.034 bars_held: int = 035 exit_reason: str = ""36 current_sl: float = 0.037 38 def __post_init__(self):39 self.current_sl = self.initial_sl40 41 42@dataclass43class MultiBacktestResult:44 """Resultado del backtest multi-activo"""45 trades: List[MultiTrade]46 total_return: float47 total_return_percent: float48 win_rate: float49 profit_factor: float50 max_drawdown: float51 max_drawdown_percent: float52 sharpe_ratio: float53 total_trades: int54 winning_trades: int55 losing_trades: int56 avg_win: float57 avg_loss: float58 best_trade: float59 worst_trade: float60 avg_bars_held: float61 max_concurrent_positions: int62 equity_curve: pd.Series63 trades_by_symbol: Dict[str, List[MultiTrade]]64 symbol_stats: Dict[str, Dict[str, Any]]65 66 67class MultiAssetBacktester:68 """69 Backtester multi-activo con capital compartido70 71 - Hasta 9 posiciones simultáneas72 - 1% de riesgo sobre capital TOTAL73 - Interés compuesto74 """75 76 MAX_CONCURRENT_POSITIONS = 977 78 def __init__(79 self,80 algorithm: BaseAlgorithm,81 initial_capital: float = 10000,82 risk_percent: float = 1.0,83 commission_pct: float = 0.1,84 max_positions: int = 985 ):86 self.algorithm = algorithm87 self.initial_capital = initial_capital88 self.risk_percent = risk_percent89 self.commission_pct = commission_pct90 self.max_positions = min(max_positions, self.MAX_CONCURRENT_POSITIONS)91 92 # Estado93 self.capital = initial_capital94 self.trades: List[MultiTrade] = []95 self.open_positions: Dict[str, MultiTrade] = {} # symbol -> trade96 self.equity_history: List[Tuple[datetime, float]] = []97 98 def run(99 self,100 data_dict: Dict[str, pd.DataFrame],101 algo_params: Dict[str, Any]102 ) -> MultiBacktestResult:103 """104 Ejecuta backtest multi-activo105 106 Args:107 data_dict: {symbol: DataFrame} con datos OHLCV108 algo_params: Parámetros del algoritmo109 110 Returns:111 MultiBacktestResult112 """113 # Reset estado114 self.capital = self.initial_capital115 self.trades = []116 self.open_positions = {}117 self.equity_history = [(None, self.initial_capital)]118 119 # Ejecutar algoritmo en cada activo120 algo_results: Dict[str, AlgorithmResult] = {}121 for symbol, df in data_dict.items():122 if df is not None and not df.empty:123 algo_results[symbol] = self.algorithm.run(df, algo_params)124 125 # Crear timeline unificado de eventos126 events = self._create_unified_timeline(data_dict, algo_results)127 128 # Procesar eventos en orden cronológico129 for event in events:130 self._process_event(event, data_dict, algo_results)131 132 # Cerrar posiciones abiertas al final133 self._close_all_positions(data_dict)134 135 # Calcular métricas136 return self._calculate_results()137 138 def _create_unified_timeline(139 self,140 data_dict: Dict[str, pd.DataFrame],141 algo_results: Dict[str, AlgorithmResult]142 ) -> List[Dict]:143 """144 Crea timeline unificado de eventos ordenados cronológicamente145 """146 events = []147 148 for symbol, df in data_dict.items():149 if symbol not in algo_results:150 continue151 152 algo_result = algo_results[symbol]153 154 for i, state in enumerate(algo_result.states):155 if i >= len(df):156 break157 158 timestamp = df.index[i]159 bar_data = df.iloc[i]160 161 events.append({162 'timestamp': timestamp,163 'symbol': symbol,164 'bar_index': i,165 'bar_data': bar_data,166 'state': state,167 'signal': state.signal168 })169 170 # Ordenar por timestamp171 events.sort(key=lambda x: x['timestamp'])172 173 return events174 175 def _process_event(176 self,177 event: Dict,178 data_dict: Dict[str, pd.DataFrame],179 algo_results: Dict[str, AlgorithmResult]180 ):181 """Procesa un evento del timeline"""182 symbol = event['symbol']183 bar_data = event['bar_data']184 state = event['state']185 signal = event['signal']186 timestamp = event['timestamp']187 188 # 1. Actualizar posiciones abiertas (trailing stop, check SL hit)189 self._update_open_positions(timestamp, data_dict, algo_results)190 191 # 2. Procesar señal si existe192 if signal:193 self._process_signal(signal, symbol, bar_data, state, timestamp)194 195 # 3. Actualizar equity196 self._update_equity(timestamp, data_dict)197 198 def _update_open_positions(199 self,200 current_time: datetime,201 data_dict: Dict[str, pd.DataFrame],202 algo_results: Dict[str, AlgorithmResult]203 ):204 """Actualiza todas las posiciones abiertas"""205 positions_to_close = []206 207 for symbol, position in self.open_positions.items():208 if symbol not in data_dict:209 continue210 211 df = data_dict[symbol]212 213 # Encontrar la barra actual para este símbolo214 try:215 # Buscar la barra más cercana <= current_time216 mask = df.index <= current_time217 if not mask.any():218 continue219 current_bar = df.loc[mask].iloc[-1]220 bar_idx = df.index.get_loc(df.loc[mask].index[-1])221 except:222 continue223 224 # Verificar stop loss hit225 if position.direction == 'LONG':226 if current_bar['low'] <= position.current_sl:227 positions_to_close.append((symbol, current_bar, bar_idx, "Stop Loss"))228 else:229 # Actualizar trailing stop desde el estado del algoritmo230 if symbol in algo_results and bar_idx < len(algo_results[symbol].states):231 state = algo_results[symbol].states[bar_idx]232 if state.support and state.support > position.current_sl:233 position.current_sl = state.support234 else: # SHORT235 if current_bar['high'] >= position.current_sl:236 positions_to_close.append((symbol, current_bar, bar_idx, "Stop Loss"))237 else:238 # Actualizar trailing stop239 if symbol in algo_results and bar_idx < len(algo_results[symbol].states):240 state = algo_results[symbol].states[bar_idx]241 if state.resistance and state.resistance < position.current_sl:242 position.current_sl = state.resistance243 244 # Cerrar posiciones que tocaron SL245 for symbol, bar, bar_idx, reason in positions_to_close:246 self._close_position(symbol, bar, bar_idx, reason)247 248 def _process_signal(249 self,250 signal: TradeSignal,251 symbol: str,252 bar_data: pd.Series,253 state: AlgorithmState,254 timestamp: datetime255 ):256 """Procesa una señal de trading"""257 258 # Si ya hay posición en este símbolo259 if symbol in self.open_positions:260 current_pos = self.open_positions[symbol]261 262 # Cerrar si la señal es opuesta263 if (signal.signal_type == 'LONG' and current_pos.direction == 'SHORT') or \264 (signal.signal_type == 'SHORT' and current_pos.direction == 'LONG'):265 self._close_position(symbol, bar_data, state.bar_index, "Cambio de tendencia")266 267 # Abrir nueva posición si hay espacio y no hay posición en este símbolo268 if symbol not in self.open_positions and len(self.open_positions) < self.max_positions:269 if signal.signal_type in ['LONG', 'SHORT']:270 self._open_position(signal, symbol, bar_data, state, timestamp)271 272 def _open_position(273 self,274 signal: TradeSignal,275 symbol: str,276 bar_data: pd.Series,277 state: AlgorithmState,278 timestamp: datetime279 ):280 """Abre una nueva posición"""281 entry_price = bar_data['close']282 283 # Stop loss inicial284 if signal.signal_type == 'LONG':285 initial_sl = state.support if state.support else entry_price * 0.95286 else:287 initial_sl = state.resistance if state.resistance else entry_price * 1.05288 289 # Calcular tamaño de posición: 1% del CAPITAL TOTAL290 sl_distance = abs(entry_price - initial_sl)291 if sl_distance == 0:292 sl_distance = entry_price * 0.01293 294 risk_amount = self.capital * (self.risk_percent / 100)295 position_size = risk_amount / sl_distance296 297 # Comisión de entrada298 commission = position_size * entry_price * (self.commission_pct / 100)299 self.capital -= commission300 301 trade = MultiTrade(302 symbol=symbol,303 entry_date=timestamp,304 entry_price=entry_price,305 direction=signal.signal_type,306 initial_sl=initial_sl,307 position_size=position_size,308 capital_at_entry=self.capital,309 current_sl=initial_sl310 )311 312 self.open_positions[symbol] = trade313 314 def _close_position(315 self,316 symbol: str,317 bar_data: pd.Series,318 bar_idx: int,319 exit_reason: str320 ):321 """Cierra una posición"""322 if symbol not in self.open_positions:323 return324 325 position = self.open_positions[symbol]326 327 # Precio de salida328 if exit_reason == "Stop Loss":329 exit_price = position.current_sl330 else:331 exit_price = bar_data['close']332 333 # Calcular P&L334 if position.direction == 'LONG':335 pnl = (exit_price - position.entry_price) * position.position_size336 else:337 pnl = (position.entry_price - exit_price) * position.position_size338 339 # Comisión de salida340 commission = position.position_size * exit_price * (self.commission_pct / 100)341 pnl -= commission342 343 # Actualizar capital (interés compuesto)344 self.capital += pnl345 346 # Completar trade347 position.exit_date = bar_data.name348 position.exit_price = exit_price349 position.pnl = pnl350 position.pnl_percent = (pnl / position.capital_at_entry) * 100351 position.exit_reason = exit_reason352 353 self.trades.append(position)354 del self.open_positions[symbol]355 356 def _close_all_positions(self, data_dict: Dict[str, pd.DataFrame]):357 """Cierra todas las posiciones abiertas al final del backtest"""358 for symbol in list(self.open_positions.keys()):359 if symbol in data_dict:360 df = data_dict[symbol]361 last_bar = df.iloc[-1]362 self._close_position(symbol, last_bar, len(df) - 1, "Fin de datos")363 364 def _update_equity(self, timestamp: datetime, data_dict: Dict[str, pd.DataFrame]):365 """Actualiza la curva de equity"""366 # Capital + P&L no realizado de posiciones abiertas367 unrealized_pnl = 0368 369 for symbol, position in self.open_positions.items():370 if symbol in data_dict:371 df = data_dict[symbol]372 try:373 mask = df.index <= timestamp374 if mask.any():375 current_price = df.loc[mask].iloc[-1]['close']376 if position.direction == 'LONG':377 unrealized_pnl += (current_price - position.entry_price) * position.position_size378 else:379 unrealized_pnl += (position.entry_price - current_price) * position.position_size380 except:381 pass382 383 total_equity = self.capital + unrealized_pnl384 self.equity_history.append((timestamp, total_equity))385 386 def _calculate_results(self) -> MultiBacktestResult:387 """Calcula las métricas finales"""388 389 if not self.trades:390 equity_series = pd.Series([self.initial_capital])391 return MultiBacktestResult(392 trades=[],393 total_return=0,394 total_return_percent=0,395 win_rate=0,396 profit_factor=0,397 max_drawdown=0,398 max_drawdown_percent=0,399 sharpe_ratio=0,400 total_trades=0,401 winning_trades=0,402 losing_trades=0,403 avg_win=0,404 avg_loss=0,405 best_trade=0,406 worst_trade=0,407 avg_bars_held=0,408 max_concurrent_positions=0,409 equity_curve=equity_series,410 trades_by_symbol={},411 symbol_stats={}412 )413 414 # Métricas básicas415 total_trades = len(self.trades)416 winning = [t for t in self.trades if t.pnl > 0]417 losing = [t for t in self.trades if t.pnl <= 0]418 419 win_rate = len(winning) / total_trades * 100 if total_trades > 0 else 0420 421 # P&L422 total_return = self.capital - self.initial_capital423 total_return_percent = (total_return / self.initial_capital) * 100424 425 # Promedios426 avg_win = np.mean([t.pnl for t in winning]) if winning else 0427 avg_loss = abs(np.mean([t.pnl for t in losing])) if losing else 0428 429 # Profit Factor430 gross_profit = sum([t.pnl for t in winning]) if winning else 0431 gross_loss = abs(sum([t.pnl for t in losing])) if losing else 1432 profit_factor = gross_profit / gross_loss if gross_loss > 0 else 0433 434 # Equity curve435 equity_values = [e[1] for e in self.equity_history if e[0] is not None]436 if not equity_values:437 equity_values = [self.initial_capital]438 equity = pd.Series(equity_values)439 440 # Max Drawdown441 rolling_max = equity.expanding().max()442 drawdown = equity - rolling_max443 max_drawdown = drawdown.min()444 max_dd_pct = (max_drawdown / rolling_max[drawdown.idxmin()]) * 100 if len(equity) > 0 and drawdown.idxmin() in rolling_max.index else 0445 446 # Sharpe Ratio447 returns = equity.pct_change().dropna()448 sharpe = (returns.mean() / returns.std()) * np.sqrt(252) if len(returns) > 0 and returns.std() > 0 else 0449 450 # Best/Worst451 pnls = [t.pnl_percent for t in self.trades]452 best = max(pnls) if pnls else 0453 worst = min(pnls) if pnls else 0454 455 # Trades por símbolo456 trades_by_symbol: Dict[str, List[MultiTrade]] = {}457 for trade in self.trades:458 if trade.symbol not in trades_by_symbol:459 trades_by_symbol[trade.symbol] = []460 trades_by_symbol[trade.symbol].append(trade)461 462 # Stats por símbolo463 symbol_stats = {}464 for symbol, symbol_trades in trades_by_symbol.items():465 wins = [t for t in symbol_trades if t.pnl > 0]466 symbol_stats[symbol] = {467 'total_trades': len(symbol_trades),468 'winning': len(wins),469 'losing': len(symbol_trades) - len(wins),470 'win_rate': len(wins) / len(symbol_trades) * 100 if symbol_trades else 0,471 'total_pnl': sum(t.pnl for t in symbol_trades),472 'avg_pnl': np.mean([t.pnl for t in symbol_trades]) if symbol_trades else 0473 }474 475 return MultiBacktestResult(476 trades=self.trades,477 total_return=total_return,478 total_return_percent=total_return_percent,479 win_rate=win_rate,480 profit_factor=profit_factor,481 max_drawdown=max_drawdown,482 max_drawdown_percent=max_dd_pct,483 sharpe_ratio=sharpe,484 total_trades=total_trades,485 winning_trades=len(winning),486 losing_trades=len(losing),487 avg_win=avg_win,488 avg_loss=avg_loss,489 best_trade=best,490 worst_trade=worst,491 avg_bars_held=np.mean([t.bars_held for t in self.trades]) if self.trades else 0,492 max_concurrent_positions=self.max_positions,493 equity_curve=equity,494 trades_by_symbol=trades_by_symbol,495 symbol_stats=symbol_stats496 )497