dam/Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Backtesting engine completo + documentación (Semanas 3-4)

Browse Source 
✅ Motor de backtesting:
- BacktestEngine con simulación de trades
- Sistema de Trade y Position
- Gestión de capital y comisiones
- Slippage simulado

✅ Estrategias implementadas:
- MovingAverageCrossover (SMA/EMA configurable)
- RSIStrategy (umbrales personalizables)
- BuyAndHold (baseline)

✅ Métricas de performance:
- Sharpe Ratio, Sortino Ratio, Calmar Ratio
- Max Drawdown, Win Rate, Profit Factor
- Expectancy, Risk/Reward Ratio

✅ Scripts:
- backtest.py: Ejecutar backtests individuales
- backtest.py compare: Comparar múltiples estrategias

✅ Documentación:
- README actualizado con sección de backtesting
- Ejemplos de uso programático
- Estructura de proyecto actualizada
This commit is contained in:
DaM
2026-01-27 21:37:39 +01:00
parent 0522ea17ca
commit 9b34de3127
14 changed files with 2020 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/backtest/__init__.py
View File

@@ -0,0 +1,29 @@
# src/backtest/__init__.py
"""
Módulo de backtesting
"""
from .engine import BacktestEngine
from .strategy import Strategy, Signal
from .trade import Trade, TradeType, TradeStatus, Position
from .metrics import (
calculate_sharpe_ratio,
calculate_sortino_ratio,
calculate_max_drawdown,
print_backtest_report,
compare_strategies
)
__all__ = [
'BacktestEngine',
'Strategy',
'Signal',
'Trade',
'TradeType',
'TradeStatus',
'Position',
'calculate_sharpe_ratio',
'calculate_sortino_ratio',
'calculate_max_drawdown',
'print_backtest_report',
'compare_strategies',
]

339
src/backtest/engine.py
View File

@@ -0,0 +1,339 @@
# src/backtest/engine.py
"""
Motor de backtesting para simular estrategias de trading
"""
import pandas as pd
import numpy as np
from typing import Optional, List
from datetime import datetime
from ..utils.logger import log
from .strategy import Strategy, Signal
from .trade import Trade, TradeType, TradeStatus, Position
class BacktestEngine:
"""
Motor de backtesting que simula la ejecución de una estrategia
"""
def __init__(
self,
strategy: Strategy,
initial_capital: float = 10000,
commission: float = 0.001, # 0.1% por trade
slippage: float = 0.0005, # 0.05% de slippage
position_size: float = 1.0 # Fracción del capital por trade (1.0 = 100%)
):
"""
Inicializa el motor de backtesting
Args:
strategy: Estrategia a testear
initial_capital: Capital inicial
commission: Comisión por trade (como fracción, ej: 0.001 = 0.1%)
slippage: Slippage simulado (como fracción)
position_size: Fracción del capital a usar por trade
"""
self.strategy = strategy
self.initial_capital = initial_capital
self.commission = commission
self.slippage = slippage
self.position_size_fraction = position_size
# Estado del backtest
self.cash = initial_capital
self.equity = initial_capital
self.trades: List[Trade] = []
self.current_position: Optional[Position] = None
# Historial para métricas
self.equity_curve: List[float] = []
self.timestamps: List[datetime] = []
self.drawdown_curve: List[float] = []
# Datos
self.data: Optional[pd.DataFrame] = None
log.info(f"Backtesting Engine inicializado: {strategy}")
log.info(f"Capital inicial: ${initial_capital:,.2f}")
log.info(f"Comisión: {commission*100:.2%}, Slippage: {slippage*100:.2%}")
def run(self, data: pd.DataFrame) -> dict:
"""
Ejecuta el backtest sobre los datos históricos
Args:
data: DataFrame con datos OHLCV
Returns:
Diccionario con resultados del backtest
"""
log.info("="*70)
log.info("INICIANDO BACKTEST")
log.info("="*70)
# Reset state
self._reset()
# Establecer datos en la estrategia
self.data = data.copy()
self.strategy.set_data(self.data)
log.info(f"Datos: {len(data)} velas")
log.info(f"Periodo: {data.index[0]} a {data.index[-1]}")
log.info(f"Estrategia: {self.strategy}")
# Iterar por cada timestamp
for idx in range(len(self.data)):
self._process_bar(idx)
# Cerrar posición abierta al final
if self.current_position is not None:
self._close_position(len(self.data) - 1, "End of backtest")
# Calcular resultados
results = self._calculate_results()
log.info("="*70)
log.success("BACKTEST COMPLETADO")
log.info("="*70)
return results
def _reset(self):
"""Reinicia el estado del backtest"""
self.cash = self.initial_capital
self.equity = self.initial_capital
self.trades = []
self.current_position = None
self.equity_curve = []
self.timestamps = []
self.drawdown_curve = []
def _process_bar(self, idx: int):
"""
Procesa una barra de datos (timestamp)
Args:
idx: Índice de la barra actual
"""
current_bar = self.data.iloc[idx]
current_price = current_bar['close']
current_time = current_bar.name # El índice es el timestamp
# Actualizar equity con posición actual
self._update_equity(idx)
# Guardar historial
self.equity_curve.append(self.equity)
self.timestamps.append(current_time)
# Generar señal de la estrategia
signal = self.strategy.generate_signal(idx)
# Ejecutar acciones según señal
if signal == Signal.BUY and self.current_position is None:
self._open_position(idx, TradeType.LONG)
elif signal == Signal.SELL and self.current_position is not None:
self._close_position(idx, "Strategy signal")
def _open_position(self, idx: int, trade_type: TradeType):
"""
Abre una nueva posición
"""
current_bar = self.data.iloc[idx]
current_price = current_bar['close']
current_time = current_bar.name
# Aplicar slippage (en compra, pagamos más)
execution_price = current_price * (1 + self.slippage)
# Calcular tamaño de la posición
position_value = self.cash * self.position_size_fraction
size = position_value / execution_price
# Calcular comisión
commission_cost = position_value * self.commission
# Verificar que tenemos suficiente cash
if self.cash < position_value + commission_cost:
log.warning(f"Cash insuficiente para abrir posición: ${self.cash:.2f}")
return
# Crear trade
trade = Trade(
symbol=current_bar.get('symbol', 'UNKNOWN'),
trade_type=trade_type,
entry_price=execution_price,
entry_time=current_time,
size=size,
entry_commission=commission_cost,
entry_reason="Strategy signal"
)
# Actualizar cash
self.cash -= (position_value + commission_cost)
# Crear posición
self.current_position = Position(
symbol=trade.symbol,
trade_type=trade_type,
average_price=execution_price,
total_size=size,
trades=[trade]
)
self.trades.append(trade)
log.debug(f"[{current_time}] OPEN {trade_type.value}: "
f"Price: ${execution_price:.2f}, Size: {size:.4f}, "
f"Value: ${position_value:.2f}")
def _close_position(self, idx: int, reason: str):
"""
Cierra la posición actual
"""
if self.current_position is None:
return
current_bar = self.data.iloc[idx]
current_price = current_bar['close']
current_time = current_bar.name
# Aplicar slippage (en venta, recibimos menos)
execution_price = current_price * (1 - self.slippage)
# Valor de la posición
position_value = self.current_position.total_size * execution_price
# Calcular comisión
commission_cost = position_value * self.commission
# Cerrar todos los trades de la posición
for trade in self.current_position.trades:
trade.exit_commission = commission_cost / len(self.current_position.trades)
trade.close(execution_price, current_time, reason)
# Actualizar cash
self.cash += (position_value - commission_cost)
# Calcular PnL de la posición
total_pnl = sum(t.pnl for t in self.current_position.trades)
log.debug(f"[{current_time}] CLOSE {self.current_position.trade_type.value}: "
f"Price: ${execution_price:.2f}, PnL: ${total_pnl:.2f} "
f"({total_pnl/self.initial_capital*100:.2f}%)")
# Limpiar posición actual
self.current_position = None
def _update_equity(self, idx: int):
"""
Actualiza el equity total (cash + valor de posiciones abiertas)
"""
self.equity = self.cash
if self.current_position is not None:
current_price = self.data.iloc[idx]['close']
position_value = self.current_position.total_size * current_price
self.equity += position_value
def _calculate_results(self) -> dict:
"""
Calcula métricas y resultados del backtest
"""
closed_trades = [t for t in self.trades if t.status == TradeStatus.CLOSED]
if not closed_trades:
log.warning("No hay trades cerrados para analizar")
return self._empty_results()
# Métricas básicas
total_trades = len(closed_trades)
winning_trades = [t for t in closed_trades if t.pnl > 0]
losing_trades = [t for t in closed_trades if t.pnl < 0]
win_rate = len(winning_trades) / total_trades if total_trades > 0 else 0
total_pnl = sum(t.pnl for t in closed_trades)
total_return = (self.equity - self.initial_capital) / self.initial_capital
# Profit factor
gross_profit = sum(t.pnl for t in winning_trades)
gross_loss = abs(sum(t.pnl for t in losing_trades))
profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
# Drawdown
equity_curve = np.array(self.equity_curve)
running_max = np.maximum.accumulate(equity_curve)
drawdown = (equity_curve - running_max) / running_max
max_drawdown = drawdown.min()
# Sharpe Ratio (aproximado con returns diarios)
returns = pd.Series(equity_curve).pct_change().dropna()
sharpe_ratio = (returns.mean() / returns.std()) * np.sqrt(252) if len(returns) > 1 else 0
results = {
# Generales
'initial_capital': self.initial_capital,
'final_equity': self.equity,
'total_return': total_return,
'total_return_pct': total_return * 100,
# Trades
'total_trades': total_trades,
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate': win_rate,
'win_rate_pct': win_rate * 100,
# PnL
'total_pnl': total_pnl,
'gross_profit': gross_profit,
'gross_loss': gross_loss,
'profit_factor': profit_factor,
# Average trades
'avg_win': np.mean([t.pnl for t in winning_trades]) if winning_trades else 0,
'avg_loss': np.mean([t.pnl for t in losing_trades]) if losing_trades else 0,
'avg_trade': total_pnl / total_trades if total_trades > 0 else 0,
# Risk metrics
'max_drawdown': max_drawdown,
'max_drawdown_pct': max_drawdown * 100,
'sharpe_ratio': sharpe_ratio,
# Datos para gráficos
'equity_curve': self.equity_curve,
'timestamps': self.timestamps,
'trades': closed_trades,
}
return results
def _empty_results(self) -> dict:
"""Resultados vacíos cuando no hay trades"""
return {
'initial_capital': self.initial_capital,
'final_equity': self.equity,
'total_return': 0,
'total_return_pct': 0,
'total_trades': 0,
'winning_trades': 0,
'losing_trades': 0,
'win_rate': 0,
'win_rate_pct': 0,
'total_pnl': 0,
'gross_profit': 0,
'gross_loss': 0,
'profit_factor': 0,
'avg_win': 0,
'avg_loss': 0,
'avg_trade': 0,
'max_drawdown': 0,
'max_drawdown_pct': 0,
'sharpe_ratio': 0,
'equity_curve': self.equity_curve,
'timestamps': self.timestamps,
'trades': [],
}

278
src/backtest/metrics.py
View File

@@ -0,0 +1,278 @@
# src/backtest/metrics.py
"""
Métricas avanzadas de performance para backtesting
"""
import numpy as np
import pandas as pd
from typing import List
from .trade import Trade
def calculate_sharpe_ratio(returns: pd.Series, risk_free_rate: float = 0.0) -> float:
"""
Sharpe Ratio - Retorno ajustado por riesgo
Args:
returns: Serie de retornos
risk_free_rate: Tasa libre de riesgo (anualizada)
Returns:
Sharpe Ratio anualizado
"""
if len(returns) == 0 or returns.std() == 0:
return 0.0
excess_returns = returns - risk_free_rate / 252 # Daily risk-free rate
sharpe = (excess_returns.mean() / excess_returns.std()) * np.sqrt(252)
return sharpe
def calculate_sortino_ratio(returns: pd.Series, risk_free_rate: float = 0.0) -> float:
"""
Sortino Ratio - Como Sharpe pero solo penaliza volatilidad a la baja
Args:
returns: Serie de retornos
risk_free_rate: Tasa libre de riesgo
Returns:
Sortino Ratio anualizado
"""
if len(returns) == 0:
return 0.0
excess_returns = returns - risk_free_rate / 252
downside_returns = excess_returns[excess_returns < 0]
if len(downside_returns) == 0 or downside_returns.std() == 0:
return 0.0
sortino = (excess_returns.mean() / downside_returns.std()) * np.sqrt(252)
return sortino
def calculate_calmar_ratio(returns: pd.Series, max_drawdown: float) -> float:
"""
Calmar Ratio - Retorno anualizado / Max Drawdown
Args:
returns: Serie de retornos
max_drawdown: Max drawdown (como valor positivo, ej: 0.2 para 20%)
Returns:
Calmar Ratio
"""
if max_drawdown == 0:
return 0.0
annual_return = returns.mean() * 252
calmar = annual_return / abs(max_drawdown)
return calmar
def calculate_max_drawdown(equity_curve: List[float]) -> tuple[float, int, int]:
"""
Calcula el máximo drawdown
Args:
equity_curve: Lista con valores de equity
Returns:
Tupla (max_drawdown, start_idx, end_idx)
"""
equity = np.array(equity_curve)
running_max = np.maximum.accumulate(equity)
drawdown = (equity - running_max) / running_max
max_dd = drawdown.min()
end_idx = drawdown.argmin()
# Encontrar el inicio del drawdown
start_idx = 0
for i in range(end_idx, -1, -1):
if equity[i] == running_max[end_idx]:
start_idx = i
break
return max_dd, start_idx, end_idx
def calculate_win_rate(trades: List[Trade]) -> float:
"""
Porcentaje de trades ganadores
"""
if not trades:
return 0.0
winning = sum(1 for t in trades if t.pnl > 0)
return winning / len(trades)
def calculate_profit_factor(trades: List[Trade]) -> float:
"""
Profit Factor = Gross Profit / Gross Loss
"""
if not trades:
return 0.0
gross_profit = sum(t.pnl for t in trades if t.pnl > 0)
gross_loss = abs(sum(t.pnl for t in trades if t.pnl < 0))
if gross_loss == 0:
return float('inf') if gross_profit > 0 else 0.0
return gross_profit / gross_loss
def calculate_expectancy(trades: List[Trade]) -> float:
"""
Expectancy = Average Win * Win Rate - Average Loss * Loss Rate
"""
if not trades:
return 0.0
winning_trades = [t for t in trades if t.pnl > 0]
losing_trades = [t for t in trades if t.pnl < 0]
win_rate = len(winning_trades) / len(trades)
loss_rate = len(losing_trades) / len(trades)
avg_win = np.mean([t.pnl for t in winning_trades]) if winning_trades else 0
avg_loss = abs(np.mean([t.pnl for t in losing_trades])) if losing_trades else 0
expectancy = (avg_win * win_rate) - (avg_loss * loss_rate)
return expectancy
def calculate_recovery_factor(total_return: float, max_drawdown: float) -> float:
"""
Recovery Factor = Net Profit / Max Drawdown
"""
if max_drawdown == 0:
return 0.0
return total_return / abs(max_drawdown)
def calculate_risk_reward_ratio(trades: List[Trade]) -> float:
"""
Average Win / Average Loss
"""
if not trades:
return 0.0
winning_trades = [t for t in trades if t.pnl > 0]
losing_trades = [t for t in trades if t.pnl < 0]
if not winning_trades or not losing_trades:
return 0.0
avg_win = np.mean([t.pnl for t in winning_trades])
avg_loss = abs(np.mean([t.pnl for t in losing_trades]))
if avg_loss == 0:
return float('inf')
return avg_win / avg_loss
def print_backtest_report(results: dict):
"""
Imprime un reporte detallado de los resultados del backtest
"""
print("\n" + "="*70)
print("📊 REPORTE DE BACKTEST")
print("="*70)
# Capital
print(f"\n💰 CAPITAL:")
print(f" Capital Inicial: ${results['initial_capital']:>12,.2f}")
print(f" Capital Final: ${results['final_equity']:>12,.2f}")
print(f" Retorno Total: {results['total_return_pct']:>12.2f}%")
print(f" PnL Total: ${results['total_pnl']:>12,.2f}")
# Trades
print(f"\n📈 TRADES:")
print(f" Total Trades: {results['total_trades']:>12}")
print(f" Trades Ganadores: {results['winning_trades']:>12}")
print(f" Trades Perdedores: {results['losing_trades']:>12}")
print(f" Win Rate: {results['win_rate_pct']:>12.2f}%")
# Performance
print(f"\n💵 PERFORMANCE:")
print(f" Ganancia Bruta: ${results['gross_profit']:>12,.2f}")
print(f" Pérdida Bruta: ${results['gross_loss']:>12,.2f}")
print(f" Profit Factor: {results['profit_factor']:>12.2f}")
print(f" Trade Promedio: ${results['avg_trade']:>12,.2f}")
print(f" Ganancia Promedio: ${results['avg_win']:>12,.2f}")
print(f" Pérdida Promedio: ${results['avg_loss']:>12,.2f}")
# Risk
print(f"\n⚠️ RIESGO:")
print(f" Max Drawdown: {results['max_drawdown_pct']:>12.2f}%")
print(f" Sharpe Ratio: {results['sharpe_ratio']:>12.2f}")
# Calcular métricas adicionales si hay trades
if results['trades']:
sortino = calculate_sortino_ratio(
pd.Series(results['equity_curve']).pct_change().dropna()
)
expectancy = calculate_expectancy(results['trades'])
rr_ratio = calculate_risk_reward_ratio(results['trades'])
print(f" Sortino Ratio: {sortino:>12.2f}")
print(f" Expectancy: ${expectancy:>12,.2f}")
print(f" Risk/Reward Ratio: {rr_ratio:>12.2f}")
print("\n" + "="*70)
def compare_strategies(results_dict: dict):
"""
Compara múltiples estrategias
Args:
results_dict: Diccionario {nombre_estrategia: results}
"""
print("\n" + "="*90)
print("🔍 COMPARACIÓN DE ESTRATEGIAS")
print("="*90)
# Crear tabla comparativa
comparison = pd.DataFrame({
name: {
'Retorno Total (%)': res['total_return_pct'],
'Sharpe Ratio': res['sharpe_ratio'],
'Max Drawdown (%)': res['max_drawdown_pct'],
'Win Rate (%)': res['win_rate_pct'],
'Profit Factor': res['profit_factor'],
'Total Trades': res['total_trades'],
}
for name, res in results_dict.items()
}).T
print(comparison.to_string())
print("="*90 + "\n")
# Mejor estrategia por cada métrica
print("🏆 MEJORES POR MÉTRICA:")
for col in comparison.columns:
best = comparison[col].idxmax()
value = comparison.loc[best, col]
print(f" {col:.<30} {best:>20} ({value:.2f})")
print()
def calculate_all_metrics(results: dict) -> dict:
"""
Calcula todas las métricas disponibles sobre los resultados
Returns:
Diccionario con métricas adicionales
"""
if not results['trades']:
return {}
returns = pd.Series(results['equity_curve']).pct_change().dropna()
additional_metrics = {
'sortino_ratio': calculate_sortino_ratio(returns),
'calmar_ratio': calculate_calmar_ratio(returns, results['max_drawdown']),
'expectancy': calculate_expectancy(results['trades']),
'risk_reward_ratio': calculate_risk_reward_ratio(results['trades']),
'recovery_factor': calculate_recovery_factor(
results['total_return'],
results['max_drawdown']
),
}
return additional_metrics

195
src/backtest/optimizer.py Normal file
View File

@@ -0,0 +1,195 @@
# src/backtest/optimizer.py
"""
Optimizador de parámetros para estrategias
"""
import pandas as pd
from typing import Dict, List, Any, Type
from itertools import product
from ..utils.logger import log
from .engine import BacktestEngine
from .strategy import Strategy
class ParameterOptimizer:
"""
Optimiza parámetros de estrategias mediante grid search
"""
def __init__(self,
strategy_class: Type[Strategy],
data: pd.DataFrame,
initial_capital: float = 10000,
commission: float = 0.001):
"""
Args:
strategy_class: Clase de estrategia (no instancia)
data: Datos para backtest
initial_capital: Capital inicial
commission: Comisión por trade
"""
self.strategy_class = strategy_class
self.data = data
self.initial_capital = initial_capital
self.commission = commission
self.results: List[Dict] = []
def optimize(self, param_grid: Dict[str, List[Any]]) -> pd.DataFrame:
"""
Optimiza parámetros mediante grid search
Args:
param_grid: Diccionario con parámetros a probar
Ejemplo: {
'fast_period': [5, 10, 15, 20],
'slow_period': [30, 50, 100],
'ma_type': ['sma', 'ema']
}
Returns:
DataFrame con resultados ordenados por retorno
"""
# Generar todas las combinaciones posibles
param_names = list(param_grid.keys())
param_values = list(param_grid.values())
combinations = list(product(*param_values))
total_tests = len(combinations)
log.info(f"Iniciando optimización: {total_tests} combinaciones a probar")
# Probar cada combinación
for i, values in enumerate(combinations, 1):
params = dict(zip(param_names, values))
log.debug(f"[{i}/{total_tests}] Probando: {params}")
try:
# Crear estrategia con estos parámetros
strategy = self.strategy_class(**params)
# Ejecutar backtest
engine = BacktestEngine(
strategy=strategy,
initial_capital=self.initial_capital,
commission=self.commission
)
results = engine.run(self.data)
# Guardar resultados
result_entry = {
**params, # Parámetros probados
'total_return_pct': results['total_return_pct'],
'sharpe_ratio': results['sharpe_ratio'],
'max_drawdown_pct': results['max_drawdown_pct'],
'total_trades': results['total_trades'],
'win_rate_pct': results['win_rate_pct'],
'profit_factor': results['profit_factor'],
}
self.results.append(result_entry)
except Exception as e:
log.error(f"Error con parámetros {params}: {e}")
continue
# Convertir a DataFrame
df_results = pd.DataFrame(self.results)
# Ordenar por retorno (mejor primero)
df_results = df_results.sort_values('total_return_pct', ascending=False)
log.success(f"Optimización completa: {len(df_results)} resultados válidos")
return df_results
def get_best_params(self, metric: str = 'total_return_pct') -> Dict:
"""
Obtiene los mejores parámetros según una métrica
Args:
metric: Métrica a optimizar ('total_return_pct', 'sharpe_ratio', etc)
Returns:
Diccionario con los mejores parámetros
"""
if not self.results:
raise ValueError("No hay resultados. Ejecuta optimize() primero.")
df_results = pd.DataFrame(self.results)
# Ordenar por la métrica elegida
if metric in ['max_drawdown_pct']:
# Para drawdown, queremos el MENOR (menos negativo)
best_idx = df_results[metric].idxmax()
else:
# Para otras métricas, queremos el MAYOR
best_idx = df_results[metric].idxmax()
best_result = df_results.loc[best_idx]
# Extraer solo los parámetros (no las métricas)
param_names = [col for col in df_results.columns
if col not in ['total_return_pct', 'sharpe_ratio',
'max_drawdown_pct', 'total_trades',
'win_rate_pct', 'profit_factor']]
best_params = {param: best_result[param] for param in param_names}
log.info(f"Mejores parámetros según {metric}: {best_params}")
log.info(f" {metric}: {best_result[metric]:.2f}")
return best_params
# ============================================================================
# Ejemplo de Uso
# ============================================================================
"""
from src.data.storage import StorageManager
from src.strategies.moving_average import MovingAverageCrossover
from src.backtest.optimizer import ParameterOptimizer
# Cargar datos
storage = StorageManager(...)
data = storage.load_ohlcv('BTC/USDT', '1h')
# Crear optimizador
optimizer = ParameterOptimizer(
strategy_class=MovingAverageCrossover, # Clase, no instancia
data=data,
initial_capital=10000,
commission=0.001
)
# Definir parámetros a probar
param_grid = {
'fast_period': [5, 10, 15, 20, 25],
'slow_period': [30, 50, 70, 100, 150],
'ma_type': ['sma', 'ema']
}
# Ejecutar optimización (probará 5 × 5 × 2 = 50 combinaciones)
results_df = optimizer.optimize(param_grid)
# Ver mejores resultados
print(results_df.head(10))
# Obtener mejores parámetros
best_params = optimizer.get_best_params(metric='sharpe_ratio')
print(f"Mejores parámetros: {best_params}")
# Usar los mejores parámetros
best_strategy = MovingAverageCrossover(**best_params)
```
---
## 📊 Output esperado:
```
fast_period slow_period ma_type total_return_pct sharpe_ratio max_drawdown_pct
0 15 50 ema 45.20 2.10 -12.30
1 10 30 sma 42.80 1.95 -15.20
2 20 70 ema 38.50 1.85 -14.10
3 5 30 sma 35.20 1.75 -18.50
...
"""

198
src/backtest/strategy.py
View File

@@ -0,0 +1,198 @@
# src/backtest/strategy.py
"""
Clase base para estrategias de trading
"""
from abc import ABC, abstractmethod
from enum import Enum
import pandas as pd
from typing import Optional, Dict, Any
class Signal(Enum):
"""Señales de Trading"""
BUY = 1
SELL = -1
HOLD = 0
class Strategy(ABC):
"""
Clase base abstracta para todas las estrategias
Para crear una estrategia:
1. Heredar de Strategy
2. Implementar init_indicators()
3. Implementar generate_signal()
"""
def __init__(self, name: str, params: Optional[Dict[str, Any]] = 0):
"""
Inicializa la estrategia
Args:
name: Nombre de la estrategia
params: Parámetros configurables de la estrategia
"""
self.name = name
self.params = params or {}
self.data: Optional[pd.DataFrame] = None
self.current_position = 0 # 1 = long, -1 = short, 0 = sin posición
@abstractmethod
def init_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
"""
Inicializa y calcula indicadores técnicos sobre los datos
Args:
data: DataFrame con datos OHLCV
Returns:
DataFrame con indicadores añadidos
"""
pass
@abstractmethod
def generate_signal(self, idx: int) -> Signal:
"""
Genera señal de trading para un punto específico en el tiempo
Args:
idx: Índice del DataFrame (posición temporal)
Returns:
Signal (BUY, SELL, HOLD)
"""
pass
def set_data(self, data: pd.DataFrame):
"""
Establece los datos y calcula indicadores
"""
self.data = self.init_indicators(data.copy())
def get_indicator_value(self, idx: int, indicator_name: str) -> Any:
"""
Obtiene el valor de un indicador en un punto temporal específico
"""
if self.data is None:
raise ValueError("Data no ha sido establecida. Llama a set_data() primero.")
if indicator_name not in self.data.columns:
raise ValueError(f"Indicador '{indicator_name}' no existe en los datos.")
return self.data.iloc[idx][indicator_name]
def __repr__(self):
params_str = ", ".join([f"{k}={v}" for k, v in self.params.items()])
return f"{self.name}({params_str})"
# ============================================================================
# Funciones auxiliares para indicadores técnicos comunes
# ============================================================================
def calculate_sma(data: pd.Series, period: int) -> pd.Series:
"""
Simple Moving Average
Args:
data: Serie de precios
period: Periodo de la media
Returns:
Serie con SMA
"""
return data.rolling(window=period).mean()
def calculate_ema(data: pd.Series, period: int) -> pd.Series:
"""
Exponential Moving Average
Args:
data: Serie de precios
period: Periodo de la media
Returns:
Serie con EMA
"""
return data.ewm(span=period, adjust=False).mean()
def calculate_rsi(data: pd.Series, period: int = 14) -> pd.Series:
"""
Relative Strength Index
Args:
data: Serie de precios
period: Periodo del RSI
Returns:
Serie con RSI
"""
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def calculate_macd(data: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> tuple:
"""
MACD (Moving Average Convergence Divergence)
Args:
data: Serie de precios
fast: Periodo EMA rápida
slow: Periodo EMA lenta
signal: Periodo de la línea de señal
Returns:
Tupla (MACD, Signal Line, Histogram)
"""
ema_fast = calculate_ema(data, fast)
ema_slow = calculate_ema(data, slow)
macd_line = ema_fast - ema_slow
signal_line = calculate_ema(macd_line, signal)
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
def calculate_bollinger_bands(data: pd.Series, period: int = 20, std_dev: float = 2) -> tuple:
"""
Bollinger Bands
Args:
data: Serie de precios
period: Periodo de la media móvil
std_dev: Número de desviaciones estándar
Returns:
Tupla (Upper Band, Middle Band, Lower Band)
"""
middle = calculate_sma(data, period)
std = data.rolling(window=period).std()
upper = middle + (std * std_dev)
lower = middle - (std * std_dev)
return upper, middle, lower
def calculate_atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series:
"""
Average True Range (volatilidad)
Args:
high: Serie de precios máximos
low: Serie de precios mínimos
close: Serie de precios de cierre
period: Periodo del ATR
Returns:
Serie con ATR
"""
tr1 = high - low
tr2 = abs(high - close.shift())
tr3 = abs(low - close.shift())
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.rolling(window=period).mean()
return atr

132
src/backtest/trade.py
View File

@@ -0,0 +1,132 @@
# src/backtest/trade.py
"""
Clases para representar trades y posiciones
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
class TradeType(Enum):
"""Tipo de trade"""
LONG = "LONG"
SHORT = "SHORT"
class TradeStatus(Enum):
"""Estado del trade"""
OPEN = "OPEN"
CLOSED = "CLOSED"
@dataclass
class Trade:
"""
Representa un trade individual
"""
symbol: str
trade_type: TradeType
entry_price: float
entry_time: datetime
size: float
exit_price: Optional[float] = None
exit_time: Optional[datetime] = None
status: TradeStatus = field(default=TradeStatus.OPEN)
# Costes
entry_commission: float = 0.0
exit_commission: float = 0.0
#Razones
entry_reason: str = ""
exit_reason: str = ""
@property
def pnl(self) -> float:
"""
Calcula profit and loss del trade
"""
if self.status == TradeStatus.OPEN or self.exit_price is None:
return 0.0
if self.trade_type == TradeType.LONG:
raw_pnl = (self.exit_price - self.entry_price) * self.size
else:
raw_pnl = (self.entry_price - self.exit_price) * self.size
total_pnl = raw_pnl - self.entry_commission - self.exit_commission
return total_pnl
@property
def pnl_percentage(self) -> float:
"""
Retorno porcentual del trade
"""
if self.status == TradeStatus.OPEN or self.exit_price is None:
return 0.0
investment = self.entry_price * self.size
return (self.pnl / investment) * 100
@property
def duration(self) -> Optional[float]:
"""
Duración del trade en horas
"""
if self.exit_time is None:
return None
delta = self.exit_time - self.entry_time
return delta.total_seconds() / 3600
def close(self, exit_price: float, exit_time: datetime, reason: str = ""):
"""
Cierra el trade
"""
self.exit_price = exit_price
self.exit_time = exit_time
self.status = TradeStatus.CLOSED
self.exit_reason = reason
def __repr__(self):
status_str = "OPEN" if self.status == TradeStatus.OPEN else f"CLOSED (PnL: {self.pnl:.2f})"
return (f"Trade({self.trade_type.value} {self.symbol} @ {self.entry_price:.2f}, "
f"Size: {self.size:.4f}, {status_str})")
@dataclass
class Position:
"""
Representa una posición activa (puede contener múltiples trades)
"""
symbol: str
trade_type: TradeType
average_price: float
total_size: float
trades: list = field(default_factory=list)
def add_trade(self, trade: Trade):
"""
Añade un trade a la posición y actualiza el precio promedio
"""
total_cost = self.average_price * self.total_size
new_cost = trade.entry_price * trade.size
self.total_size += trade.size
self.average_price = (total_cost + new_cost) / self.total_size
self.trades.append(trade)
def close_all(self, exit_price: float, exit_time: datetime, reason: str = ""):
"""
Cierra todos los trades de la posición
"""
for trade in self.trades:
if trade.status == TradeStatus.OPEN:
trade.close(exit_price, exit_time, reason)
def unrealized_pnl(self, current_price: float) -> float:
"""
PnL no realizado basado en el precio actual
"""
if self.trade_type == TradeType.LONG:
return (current_price - self.average_price) * self.total_size
else:
return (self.average_price - current_price) * self.total_size