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

Engine: add stop loss integration (fixed & trailing) with tests

Browse Source
This commit is contained in:
DaM
2026-01-30 17:05:47 +01:00
parent af7b862f60
commit c569170fcc
24 changed files with 1121 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

293
src/backtest/engine.py
View File

@@ -9,6 +9,8 @@ from datetime import datetime
from ..utils.logger import log
from .strategy import Strategy, Signal
from .trade import Trade, TradeType, TradeStatus, Position
from ..risk.sizing.base import PositionSizer
from ..risk.stops.base import StopLoss
class BacktestEngine:
"""
@@ -21,7 +23,9 @@ class BacktestEngine:
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%)
position_size: float = 1.0, # Fracción del capital por trade (1.0 = 100%)
position_sizer: Optional[PositionSizer] = None,
stop_loss: Optional[StopLoss] = None,
):
"""
Inicializa el motor de backtesting
@@ -32,12 +36,15 @@ class BacktestEngine:
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
position_sizer: Objeto PositionSizer para calcular units dinámicamente
"""
self.strategy = strategy
self.initial_capital = initial_capital
self.commission = commission
self.slippage = slippage
self.position_size_fraction = position_size
self.position_sizer = position_sizer
self.stop_loss = stop_loss
# Estado del backtest
self.cash = initial_capital
@@ -56,6 +63,11 @@ class BacktestEngine:
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%}")
if self.position_sizer is not None:
log.info(f"PositionSizer activo: {self.position_sizer.__class__.__name__}")
else:
log.info(f"PositionSizer: None (fallback position_size={self.position_size_fraction:.2f})")
def run(self, data: pd.DataFrame) -> dict:
"""
@@ -120,26 +132,57 @@ class BacktestEngine:
current_price = current_bar['close']
current_time = current_bar.name # El índice es el timestamp
# Actualizar equity con posición actual
# 🟠 ACTUALIZAR TRAILING STOP (si aplica)
if self.current_position is not None and self.stop_loss is not None:
try:
new_stop = self.stop_loss.update_stop(
data=self.data,
idx=idx,
position=self.current_position,
)
if new_stop is not None:
self.current_position.move_stop(new_stop)
except Exception as e:
log.warning(f"[{current_time}] Error actualizando stop: {e}")
# 🔴 STOP LOSS CHECK
if self.current_position is not None:
if self.current_position.is_stop_hit(current_price):
unrealized = self.current_position.unrealized_pnl(current_price)
log.info(
f"[{current_time}] 🔴 STOP HIT | "
f"{self.current_position.trade_type.value} {self.current_position.symbol} | "
f"Price: {current_price:.2f} | "
f"Stop: {self.current_position.stop_price:.2f} | "
f"Unrealized PnL: {unrealized:.2f}"
)
self._close_position(idx, reason="Stop Loss")
self._update_equity(idx)
return
# 📈 Actualizar equity
self._update_equity(idx)
# Guardar historial
# 📊 Guardar historial
self.equity_curve.append(self.equity)
self.timestamps.append(current_time)
# Generar señal de la estrategia
# 📡 Generar señal
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
Abre una nueva posición, delegando size al PositionSizer si existe
"""
current_bar = self.data.iloc[idx]
current_price = current_bar['close']
@@ -147,26 +190,57 @@ class BacktestEngine:
# 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
# --------------------------------------------------
# ✅ 1) Calcular units (size) vía sizer o fallback legacy
# --------------------------------------------------
if self.position_sizer is not None:
try:
units = float(
self.position_sizer.calculate_size(
capital=self.cash,
entry_price=float(execution_price),
stop_price=None, # stops aún no integrados
volatility=None # vol/ATR aún no integrado aquí
)
)
except Exception as e:
log.warning(f"[{current_time}] PositionSizer rechazó la entrada: {e}")
return
if not np.isfinite(units) or units <= 0:
log.warning(f"[{current_time}] PositionSizer devolvió units inválidos: {units}")
return
position_value = units * execution_price
else:
# Fallback actual: usar fracción del cash
position_value = self.cash * self.position_size_fraction
units = position_value / execution_price
# --------------------------------------------------
# ✅ 2) Comisión basada en el nominal invertido
# --------------------------------------------------
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}")
log.warning(
f"[{current_time}] Cash insuficiente para abrir posición "
f"(cash=${self.cash:.2f}, needed=${position_value + commission_cost:.2f})"
)
return
# Crear trade
# --------------------------------------------------
# ✅ 3) Crear trade + posición
# --------------------------------------------------
trade = Trade(
symbol=current_bar.get('symbol', 'UNKNOWN'),
trade_type=trade_type,
entry_price=execution_price,
entry_time=current_time,
size=size,
size=units,
entry_commission=commission_cost,
entry_reason="Strategy signal"
)
@@ -179,15 +253,25 @@ class BacktestEngine:
symbol=trade.symbol,
trade_type=trade_type,
average_price=execution_price,
total_size=size,
total_size=units,
trades=[trade]
)
# 🔴 FIJAR STOP INICIAL
if self.stop_loss is not None:
stop_price = self.stop_loss.get_stop_price(
data=self.data,
idx=idx,
entry_price=execution_price,
trade_type=trade_type,
)
self.current_position.set_stop(stop_price)
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}")
f"Price: ${execution_price:.2f}, Units: {units:.6f}, "
f"Value: ${position_value:.2f}, Fee: ${commission_cost:.2f}")
def _close_position(self, idx: int, reason: str):
"""
@@ -195,145 +279,150 @@ class BacktestEngine:
"""
if self.current_position is None:
return
current_bar = self.data.iloc[idx]
current_price = current_bar['close']
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}%)")
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']
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')
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
if returns.std() == 0 or np.isnan(returns.std()):
sharpe_ratio = 0.0
else:
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,
"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,
"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,
"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,
"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,
"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,
"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': [],
}
"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": [],
}

51
src/backtest/trade.py
View File

@@ -102,6 +102,8 @@ class Position:
average_price: float
total_size: float
trades: list = field(default_factory=list)
stop_price: Optional[float] = None
initial_stop_price: Optional[float] = None
def add_trade(self, trade: Trade):
"""
@@ -129,4 +131,51 @@ class Position:
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
return (self.average_price - current_price) * self.total_size
def set_stop(self, price: float):
"""
Establece el stop de la posición.
Guarda el stop inicial si aún no existe.
"""
if price is None or price <= 0:
raise ValueError("stop_price inválido")
if self.initial_stop_price is None:
self.initial_stop_price = price
self.stop_price = price
def is_stop_hit(self, current_price: float):
"""
Devuelve True si el stop ha sido alcanzado.
"""
if self.stop_price is None:
return False
if self.trade_type == TradeType.LONG:
return current_price <= self.stop_price
elif self.trade_type == TradeType.SHORT:
return current_price >= self.stop_price
return False
def move_stop(self, new_price: float):
"""
Mueve el stop solo si reduce el riesgo.
"""
if new_price is None or new_price <= 0:
return
if self.stop_price is None:
self.set_stop(new_price)
return
# LONG → solo subir stop
if self.trade_type == TradeType.LONG and new_price > self.stop_price:
self.stop_price = new_price
# Short → solo bajar stop
elif self.trade_type == TradeType.SHORT and new_price < self.stop_price:
self.stop_price = new_price

19
src/data/processor.py
View File

@@ -268,22 +268,3 @@ class DataProcessor:
log.debug(f"Datos normalizados usando {method}")
return df_norm
@staticmethod
def calculate_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""
Calcula indicadores técnicos (ADX, etc.)
"""
adx = ta.adx(
high=df['high'],
low=df['low'],
close=df['close'],
length=14
)
df = df.copy()
df['adx'] = adx['ADX_14']
log.debug("Indicadores técnicos calculados (ADX)")
return df

0
src/risk/__init__.py Normal file
View File

0
src/risk/sizing/__init__.py Normal file
View File

23
src/risk/sizing/base.py Normal file
View File

@@ -0,0 +1,23 @@
# src/risk/sizing/base.py
from abc import ABC, abstractmethod
from typing import Optional
class PositionSizer(ABC):
"""
Clase base para position sizing.
"""
@abstractmethod
def calculate_size(
self,
capital: float,
entry_price: float,
stop_price: Optional[float] = None,
volatility: Optional[float] = None,
) -> float:
"""
Devuelve el número de unidades a comprar.
"""
pass

26
src/risk/sizing/fixed.py Normal file
View File

@@ -0,0 +1,26 @@
# src/risk/sizing/fixed.py
from .base import PositionSizer
class FixedPositionSizer(PositionSizer):
"""
Usa siempre un porcentaje fijo del capital disponible.
"""
def __init__(self, capital_fraction: float = 1.0):
if not 0 < capital_fraction <= 1:
raise ValueError("capital_fraction debe estar entre 0 y 1")
self.capital_fraction = capital_fraction
def calculate_size(
self,
capital: float,
entry_price: float,
stop_price=None,
volatility=None,
) -> float:
notional = capital * self.capital_fraction
units = notional / entry_price
return units

36
src/risk/sizing/percent_risk.py Normal file
View File

@@ -0,0 +1,36 @@
# src/risk/sizing/percent_risk.py
from .base import PositionSizer
class PercentRiskSizer(PositionSizer):
"""
Position sizing basado en % de riesgo por trade.
"""
def __init__(self, risk_fraction: float):
if not 0 < risk_fraction <= 1:
raise ValueError("risk_fraction debe estar entre 0 y 1")
self.risk_fraction = risk_fraction
def calculate_size(
self,
capital: float,
entry_price: float,
stop_price: float | None = None
) -> float:
if stop_price is None:
raise ValueError("PercentRiskSizer requiere stop_price")
risk_amount = capital * self.risk_fraction
distance = abs(entry_price - stop_price)
if distance < 0:
raise ValueError("Distancia entry-stop inválida")
if distance == 0:
return 0.0
position_size = risk_amount / distance
return position_size

34
src/risk/sizing/volatility.py Normal file
View File

@@ -0,0 +1,34 @@
# src/risk/sizing/volatility.py
from .base import PositionSizer
class VolatilitySizer(PositionSizer):
"""
Ajusta el tamaño según la volatilidad (ATR o std).
"""
def __init__(self, volatility_multiplier: float = 2.0):
if volatility_multiplier <= 0:
raise ValueError("volatility_multiplier debe ser > 0")
self.volatility_multiplier = volatility_multiplier
def calculate_size(
self,
capital: float,
entry_price: float,
stop_price=None,
volatility: float = None,
) -> float:
if volatility is None:
raise ValueError("VolatilitySizer requiere volatilidad (ATR o std)")
risk_per_unit = volatility * entry_price * self.volatility_multiplier
if risk_per_unit <= 0:
return 0.0
units = capital / risk_per_unit
return units

0
src/risk/stops/__init__.py Normal file
View File

70
src/risk/stops/atr_stop.py Normal file
View File

@@ -0,0 +1,70 @@
# src/risk/stops/atr_stop.py
import pandas as pd
import numpy as np
from src.risk.stops.base import StopLoss
from src.backtest.trade import TradeType
class ATRStop(StopLoss):
"""
Stop basado en ATR (Average True Range).
"""
def __init__(self, atr_period: int = 14, multiplier: float = 2.0):
if atr_period <= 0:
raise ValueError("atr_period debe ser > 0")
if multiplier <= 0:
raise ValueError("multiplier debe ser > 0")
self.atr_period = atr_period
self.multiplier = multiplier
def _calculate_atr(self, data: pd.DataFrame) -> pd.Series:
"""
Calcula el ATR clásico (Wilder).
"""
high = data["high"]
low = data["low"]
close = data["close"].shift(1)
tr = pd.concat(
[
high - low,
(high - close).abs(),
(low - close).abs(),
],
axis=1,
).max(axis=1)
atr = tr.rolling(self.atr_period).mean()
return atr
def get_stop_price(
self,
*,
data: pd.DataFrame,
idx: int,
entry_price: float,
trade_type: TradeType
) -> float:
if not {"high", "low", "close"}.issubset(data.columns):
raise ValueError("ATRStop requiere columnas high, low, close")
atr_series = self._calculate_atr(data)
atr_value = atr_series.iloc[idx]
if not np.isfinite(atr_value):
raise ValueError("ATR no disponible en este índice")
distance = atr_value * self.multiplier
if trade_type == TradeType.LONG:
return entry_price - distance
elif trade_type == TradeType.SHORT:
return entry_price + distance
else:
raise ValueError(f"TradeType no soportado: {trade_type}")

39
src/risk/stops/base.py Normal file
View File

@@ -0,0 +1,39 @@
# src/risk/stops/base.py
from abc import ABC, abstractmethod
import pandas as pd
from src.backtest.trade import TradeType
class StopLoss(ABC):
"""
Interfaz base para Stop Loss.
Un stop:
- NO ejecuta órdenes
- NO calcula size
- SOLO devuelve un precio de stop
"""
@abstractmethod
def get_stop_price(
self,
*,
data: pd.DataFrame,
idx: int,
entry_price: float,
trade_type: TradeType
) -> float:
"""
Devuelve el precio del stop.
Args:
data: DataFrame OHLCV
idx: índice temporal actual
entry_price: precio de entrada
trade_type: LONG o SHORT
Returns:
stop_price (float)
"""
pass

37
src/risk/stops/fixed_stop.py Normal file
View File

@@ -0,0 +1,37 @@
# src/risk/stops/fixed_stop.py
import pandas as pd
from src.risk.stops.base import StopLoss
from src.backtest.trade import TradeType
class FixedStop(StopLoss):
"""
Stop fijo porcentual desde el precio de entrada.
"""
def __init__(self, stop_fraction: float):
"""
Args:
stop_fraction: ej 0.02 = 2%
"""
if stop_fraction <= 0 or stop_fraction >= 1:
raise ValueError("stop_fraction debe estar entre 0 y 1")
self.stop_fraction = stop_fraction
def get_stop_price(
self,
*,
data: pd.DataFrame,
idx: int,
entry_price: float,
trade_type: TradeType
) -> float:
if trade_type == TradeType.LONG:
return entry_price * (1 - self.stop_fraction)
elif trade_type == TradeType.SHORT:
return entry_price * (1 + self.stop_fraction)
else:
raise ValueError(f"TradeType no soportado: {trade_type}")

70
src/risk/stops/trailing_stop.py Normal file
View File

@@ -0,0 +1,70 @@
# src/risk/stops/trailing_stop.py
import pandas as pd
from src.risk.stops.base import StopLoss
from src.backtest.trade import TradeType, Position
class TrailingStop(StopLoss):
"""
Trailing Stop porcentual basado en el precio de cierre.
- LONG: el stop solo sube
- SHORT: el stop solo baja
"""
def __init__(self, trailing_fraction: float):
"""
Args:
trailing_fraction: ej 0.05 = 5%
"""
if trailing_fraction <= 0 or trailing_fraction >= 1:
raise ValueError("trailing_fraction debe estar entre 0 y 1")
self.trailing_fraction = trailing_fraction
def get_stop_price(
self,
*,
data: pd.DataFrame,
idx: int,
entry_price: float,
trade_type: TradeType
) -> float:
"""
Stop inicial al abrir la posición
"""
if trade_type == TradeType.LONG:
return entry_price * (1 - self.trailing_fraction)
elif trade_type == TradeType.SHORT:
return entry_price * (1 + self.trailing_fraction)
else:
raise ValueError(f"TradeType no soportado: {trade_type}")
def update_stop(
self,
*,
data: pd.DataFrame,
idx: int,
position: Position
) -> float | None:
"""
Calcula un nuevo stop si el precio ha mejorado.
Devuelve None si el stop no debe moverse.
"""
current_price = data.iloc[idx]["close"]
if position.trade_type == TradeType.LONG:
candidate = current_price * (1 - self.trailing_fraction)
if position.stop_price is None or candidate > position.stop_price:
return candidate
elif position.trade_type == TradeType.SHORT:
candidate = current_price * (1 + self.trailing_fraction)
if position.stop_price is None or candidate < position.stop_price:
return candidate
return None