diegobeyl/backtesting
2
1"""2Motor de backtesting genérico3Ejecuta cualquier algoritmo que implemente BaseAlgorithm4"""5 6from dataclasses import dataclass, field7from typing import List, Optional, Dict, Any8from 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, TrendState18from utils.validators import InputValidator19from utils.exceptions import ValidationException, AlgorithmException20 21 22@dataclass23class Trade:24 """Representa una operación completada"""25 entry_date: datetime26 entry_price: float27 direction: str # 'LONG' or 'SHORT'28 initial_sl: float29 position_size: float30 capital_at_entry: float31 exit_date: Optional[datetime] = None32 exit_price: Optional[float] = None33 pnl: float = 0.034 pnl_percent: float = 0.035 bars_held: int = 036 exit_reason: str = ""37 sl_updates: List[float] = field(default_factory=list)38 39 40@dataclass41class BacktestResult:42 """Resultado completo del backtest"""43 trades: List[Trade]44 total_return: float45 total_return_percent: float46 win_rate: float47 profit_factor: float48 max_drawdown: float49 max_drawdown_percent: float50 sharpe_ratio: float51 total_trades: int52 winning_trades: int53 losing_trades: int54 avg_win: float55 avg_loss: float56 best_trade: float57 worst_trade: float58 avg_bars_held: float59 equity_curve: pd.Series60 algorithm_result: AlgorithmResult61 symbol: str = ""62 timeframe: str = ""63 64 65class Backtester:66 """67 Motor de backtesting genérico68 69 Ejecuta cualquier algoritmo y simula las operaciones70 """71 72 def __init__(73 self,74 algorithm: BaseAlgorithm,75 initial_capital: float = 10000,76 risk_percent: float = 1.0,77 commission_pct: float = 0.1,78 position_sizing: str = 'risk', # 'risk' or 'fixed'79 fixed_position_pct: float = 50.0, # % del capital por trade si mode='fixed'80 trade_direction: Optional[str] = None # 'LONG', 'SHORT', or None (ambos)81 ):82 self.algorithm = algorithm83 self.initial_capital = initial_capital84 self.risk_percent = risk_percent85 self.commission_pct = commission_pct86 self.position_sizing = position_sizing87 self.fixed_position_pct = fixed_position_pct88 self.trade_direction = trade_direction # Filtro de dirección89 90 # Estado del backtest91 self.capital = initial_capital92 self.trades: List[Trade] = []93 self.equity_curve: List[float] = []94 self.current_position: Optional[Trade] = None95 self._current_sl: float = 096 self._entry_bar_index: int = 097 98 def run(99 self,100 df: pd.DataFrame,101 algo_params: Dict[str, Any],102 symbol: str = "",103 timeframe: str = ""104 ) -> BacktestResult:105 """106 Ejecuta el backtest completo107 108 Args:109 df: DataFrame con datos OHLCV110 algo_params: Parámetros para el algoritmo111 symbol: Símbolo del activo (para referencia)112 timeframe: Timeframe usado (para referencia)113 114 Returns:115 BacktestResult con métricas y trades116 117 Raises:118 ValidationException: If DataFrame validation fails119 AlgorithmException: If algorithm execution fails120 """121 try:122 # STEP 1: Validate input DataFrame123 InputValidator.validate_dataframe(df, symbol)124 125 # Reset estado126 self.capital = self.initial_capital127 self.trades = []128 self.equity_curve = [self.initial_capital]129 self.current_position = None130 131 # STEP 2: Execute algorithm132 try:133 algo_result = self.algorithm.run(df, algo_params)134 except Exception as e:135 raise AlgorithmException(136 f"Algorithm execution failed: {str(e)}",137 algorithm=self.algorithm.__class__.__name__,138 details={"error": str(e)}139 )140 141 # STEP 3: Simulate trades based on signals142 for i, state in enumerate(algo_result.states):143 try:144 current = df.iloc[i]145 146 # Procesar señal si existe147 if state.signal:148 self._process_signal(state.signal, current, i, algo_result)149 150 # Actualizar trailing stop si hay posición151 if self.current_position:152 self._update_position(current, i, algo_result, state)153 154 # Actualizar equity155 self._update_equity(current)156 157 except Exception as e:158 raise AlgorithmException(159 f"Error processing bar {i}/{len(df)}: {str(e)}",160 algorithm=self.algorithm.__class__.__name__,161 bar_index=i,162 details={"error": str(e)}163 )164 165 # Cerrar posición abierta al final166 if self.current_position:167 self._close_position(df.iloc[-1], len(df) - 1, "Fin de datos")168 169 # Calcular métricas170 return self._calculate_results(df, algo_result, symbol, timeframe)171 172 except ValidationException:173 # Re-raise validation errors as-is174 raise175 except AlgorithmException:176 # Re-raise algorithm errors as-is177 raise178 except Exception as e:179 # Catch any unexpected errors and wrap them180 raise AlgorithmException(181 f"Unexpected error during backtest execution: {str(e)}",182 algorithm=self.algorithm.__class__.__name__,183 details={"error": str(e), "symbol": symbol}184 )185 186 def _process_signal(187 self,188 signal: TradeSignal,189 current: pd.Series,190 bar_index: int,191 algo_result: AlgorithmResult192 ):193 """Procesa una señal de trading"""194 195 # Filtrar por dirección si está configurado196 if self.trade_direction is not None:197 if signal.signal_type != self.trade_direction:198 return # Ignorar señal si no coincide con la dirección configurada199 200 # Cerrar posición opuesta si existe201 if self.current_position:202 if (signal.signal_type == 'LONG' and self.current_position.direction == 'SHORT') or \203 (signal.signal_type == 'SHORT' and self.current_position.direction == 'LONG'):204 self._close_position(current, bar_index, "Cambio de tendencia")205 206 # Abrir nueva posición207 if signal.signal_type in ['LONG', 'SHORT'] and not self.current_position:208 self._open_position(signal, current, bar_index)209 210 def _open_position(self, signal: TradeSignal, current: pd.Series, bar_index: int):211 """Abre una nueva posición"""212 entry_price = current['close']213 initial_sl = signal.stop_loss if signal.stop_loss else (214 current['low'] * 0.99 if signal.signal_type == 'LONG' else current['high'] * 1.01215 )216 217 # Calcular tamaño de posición según el modo218 if self.position_sizing == 'fixed':219 # MODO FIJO: Usar porcentaje fijo del capital220 position_value = self.capital * (self.fixed_position_pct / 100)221 position_size = position_value / entry_price222 else:223 # MODO RIESGO: Calcular basado en distancia al stop224 sl_distance = abs(entry_price - initial_sl)225 226 # PROTECCIÓN 1: Distancia mínima del 0.5% para evitar stops muy cercanos227 min_sl_distance = entry_price * 0.005 # 0.5%228 if sl_distance < min_sl_distance:229 sl_distance = min_sl_distance230 # Ajustar el stop loss al mínimo permitido231 if signal.signal_type == 'LONG':232 initial_sl = entry_price - min_sl_distance233 else:234 initial_sl = entry_price + min_sl_distance235 236 risk_amount = self.capital * (self.risk_percent / 100)237 position_size = risk_amount / sl_distance238 239 # PROTECCIÓN 2: Limitar el valor de la posición al 95% del capital disponible240 position_value = position_size * entry_price241 max_position_value = self.capital * 0.95242 243 if position_value > max_position_value:244 position_size = max_position_value / entry_price245 246 # Comisión de entrada247 commission = position_size * entry_price * (self.commission_pct / 100)248 self.capital -= commission249 250 self.current_position = Trade(251 entry_date=signal.timestamp,252 entry_price=entry_price,253 direction=signal.signal_type,254 initial_sl=initial_sl,255 position_size=position_size,256 capital_at_entry=self.capital257 )258 259 self._current_sl = initial_sl260 self._entry_bar_index = bar_index261 262 def _update_position(263 self,264 current: pd.Series,265 bar_index: int,266 algo_result: AlgorithmResult,267 state268 ):269 """Actualiza el trailing stop de la posición abierta"""270 pos = self.current_position271 272 # NO VERIFICAR STOP EN LA BARRA DE ENTRADA273 # En velas grandes, el low/high de la barra de entrada puede estar fuera del stop274 # Esperamos al menos 1 barra para verificar el stop275 if bar_index == self._entry_bar_index:276 return277 278 if pos.direction == 'LONG':279 # Verificar stop loss hit280 if current['low'] <= self._current_sl:281 self._close_position(current, bar_index, "Stop Loss")282 return283 284 # Actualizar SL si el soporte subió285 if state.support is not None:286 new_sl = state.support287 # Debug: contar cuántas veces se intenta actualizar288 if not hasattr(pos, '_debug_checks'):289 pos._debug_checks = 0290 pos._debug_checks += 1291 292 if new_sl > self._current_sl:293 pos.sl_updates.append(new_sl)294 self._current_sl = new_sl295 else:296 # Debug: contar None297 if not hasattr(pos, '_debug_none_count'):298 pos._debug_none_count = 0299 pos._debug_none_count += 1300 301 else: # SHORT302 # Verificar stop loss hit303 if current['high'] >= self._current_sl:304 self._close_position(current, bar_index, "Stop Loss")305 return306 307 # Actualizar SL si la resistencia bajó308 if state.resistance is not None:309 new_sl = state.resistance310 # Debug: contar cuántas veces se intenta actualizar311 if not hasattr(pos, '_debug_checks'):312 pos._debug_checks = 0313 pos._debug_checks += 1314 315 if new_sl < self._current_sl:316 pos.sl_updates.append(new_sl)317 self._current_sl = new_sl318 else:319 # Debug: contar None320 if not hasattr(pos, '_debug_none_count'):321 pos._debug_none_count = 0322 pos._debug_none_count += 1323 324 def _close_position(self, bar: pd.Series, bar_index: int, exit_reason: str):325 """Cierra la posición actual"""326 pos = self.current_position327 exit_price = self._current_sl if exit_reason == "Stop Loss" else bar['close']328 329 # Calcular P&L330 if pos.direction == 'LONG':331 pnl = (exit_price - pos.entry_price) * pos.position_size332 else:333 pnl = (pos.entry_price - exit_price) * pos.position_size334 335 # Comisión de salida336 commission = pos.position_size * exit_price * (self.commission_pct / 100)337 pnl -= commission338 339 # Actualizar capital340 self.capital += pnl341 342 # Completar trade343 pos.exit_date = bar.name344 pos.exit_price = exit_price345 pos.pnl = pnl346 pos.pnl_percent = (pnl / pos.capital_at_entry) * 100347 pos.bars_held = bar_index - self._entry_bar_index348 pos.exit_reason = exit_reason349 350 self.trades.append(pos)351 self.current_position = None352 353 def _update_equity(self, bar: pd.Series):354 """Actualiza la curva de equity"""355 if self.current_position:356 pos = self.current_position357 if pos.direction == 'LONG':358 unrealized = (bar['close'] - pos.entry_price) * pos.position_size359 else:360 unrealized = (pos.entry_price - bar['close']) * pos.position_size361 self.equity_curve.append(self.capital + unrealized)362 else:363 self.equity_curve.append(self.capital)364 365 def _calculate_results(366 self,367 df: pd.DataFrame,368 algo_result: AlgorithmResult,369 symbol: str,370 timeframe: str371 ) -> BacktestResult:372 """Calcula las métricas finales del backtest"""373 374 if not self.trades:375 return BacktestResult(376 trades=[],377 total_return=0,378 total_return_percent=0,379 win_rate=0,380 profit_factor=0,381 max_drawdown=0,382 max_drawdown_percent=0,383 sharpe_ratio=0,384 total_trades=0,385 winning_trades=0,386 losing_trades=0,387 avg_win=0,388 avg_loss=0,389 best_trade=0,390 worst_trade=0,391 avg_bars_held=0,392 equity_curve=pd.Series(self.equity_curve),393 algorithm_result=algo_result,394 symbol=symbol,395 timeframe=timeframe396 )397 398 # Métricas básicas399 total_trades = len(self.trades)400 winning = [t for t in self.trades if t.pnl > 0]401 losing = [t for t in self.trades if t.pnl <= 0]402 403 win_rate = len(winning) / total_trades * 100 if total_trades > 0 else 0404 405 # P&L406 total_return = self.capital - self.initial_capital407 total_return_percent = (total_return / self.initial_capital) * 100408 409 # Promedios410 avg_win = np.mean([t.pnl for t in winning]) if winning else 0411 avg_loss = abs(np.mean([t.pnl for t in losing])) if losing else 0412 413 # Profit Factor414 gross_profit = sum([t.pnl for t in winning]) if winning else 0415 gross_loss = abs(sum([t.pnl for t in losing])) if losing else 1416 profit_factor = gross_profit / gross_loss if gross_loss > 0 else 0417 418 # Max Drawdown419 equity = pd.Series(self.equity_curve)420 rolling_max = equity.expanding().max()421 drawdown = equity - rolling_max422 max_drawdown = drawdown.min()423 max_dd_pct = (max_drawdown / rolling_max[drawdown.idxmin()]) * 100 if len(equity) > 0 else 0424 425 # Sharpe Ratio426 returns = equity.pct_change().dropna()427 sharpe = (returns.mean() / returns.std()) * np.sqrt(252) if returns.std() > 0 else 0428 429 # Best/Worst430 pnls = [t.pnl_percent for t in self.trades]431 best = max(pnls) if pnls else 0432 worst = min(pnls) if pnls else 0433 434 # Avg bars435 avg_bars = np.mean([t.bars_held for t in self.trades])436 437 return BacktestResult(438 trades=self.trades,439 total_return=total_return,440 total_return_percent=total_return_percent,441 win_rate=win_rate,442 profit_factor=profit_factor,443 max_drawdown=max_drawdown,444 max_drawdown_percent=max_dd_pct,445 sharpe_ratio=sharpe,446 total_trades=total_trades,447 winning_trades=len(winning),448 losing_trades=len(losing),449 avg_win=avg_win,450 avg_loss=avg_loss,451 best_trade=best,452 worst_trade=worst,453 avg_bars_held=avg_bars,454 equity_curve=equity,455 algorithm_result=algo_result,456 symbol=symbol,457 timeframe=timeframe458 )459