363 lines
13 KiB
Python
363 lines
13 KiB
Python
"""Backtesting engine for ICT strategies.
|
|
|
|
Simulates trading on historical data with realistic fees
|
|
and slippage to evaluate strategy performance.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Optional
|
|
|
|
import pandas as pd
|
|
from loguru import logger
|
|
|
|
from config import settings
|
|
from indicators.ict_engine import ICTEngine
|
|
from indicators.multi_timeframe import MarketBias, TradeDirection
|
|
from indicators.confluence import ConfluenceChecker
|
|
from strategy.entry_rules import EntryRules
|
|
from strategy.exit_rules import ExitRules, ExitReason
|
|
from risk.risk_manager import RiskManager
|
|
|
|
|
|
@dataclass
|
|
class BacktestTrade:
|
|
"""A single trade in a backtest."""
|
|
|
|
entry_index: int
|
|
exit_index: int
|
|
direction: TradeDirection
|
|
entry_price: float
|
|
exit_price: float
|
|
amount: float
|
|
pnl: float
|
|
fee: float
|
|
entry_reason: str = ""
|
|
exit_reason: str = ""
|
|
confluence_score: int = 0
|
|
|
|
|
|
@dataclass
|
|
class BacktestResult:
|
|
"""Complete backtest outcome."""
|
|
|
|
trades: List[BacktestTrade] = field(default_factory=list)
|
|
equity_curve: List[float] = field(default_factory=list)
|
|
total_pnl: float = 0.0
|
|
roi_percent: float = 0.0
|
|
win_rate: float = 0.0
|
|
profit_factor: float = 0.0
|
|
sharpe_ratio: float = 0.0
|
|
max_drawdown: float = 0.0
|
|
total_trades: int = 0
|
|
winning_trades: int = 0
|
|
losing_trades: int = 0
|
|
avg_trade_duration: float = 0.0 # in candles
|
|
initial_balance: float = 0.0
|
|
|
|
def summary(self) -> dict:
|
|
return {
|
|
"total_trades": self.total_trades,
|
|
"winning_trades": self.winning_trades,
|
|
"losing_trades": self.losing_trades,
|
|
"win_rate": f"{self.win_rate:.1%}",
|
|
"total_pnl": f"{self.total_pnl:.2f}",
|
|
"roi": f"{self.roi_percent:.2%}",
|
|
"profit_factor": f"{self.profit_factor:.2f}",
|
|
"sharpe_ratio": f"{self.sharpe_ratio:.2f}",
|
|
"max_drawdown": f"{self.max_drawdown:.2%}",
|
|
"avg_duration_candles": f"{self.avg_trade_duration:.1f}",
|
|
}
|
|
|
|
|
|
class Backtester:
|
|
"""ICT strategy backtesting engine.
|
|
|
|
Iterates over historical OHLCV data candle by candle,
|
|
applying entry/exit rules and tracking performance.
|
|
"""
|
|
|
|
FEE_RATE = 0.001 # 0.1% taker fee
|
|
SLIPPAGE_RATE = 0.0005 # 0.05% slippage
|
|
|
|
def __init__(
|
|
self,
|
|
ict_engine: ICTEngine | None = None,
|
|
entry_rules: EntryRules | None = None,
|
|
exit_rules: ExitRules | None = None,
|
|
risk_manager: RiskManager | None = None,
|
|
):
|
|
self.engine = ict_engine or ICTEngine()
|
|
self.entry_rules = entry_rules or EntryRules()
|
|
self.exit_rules = exit_rules or ExitRules()
|
|
self.risk = risk_manager or RiskManager()
|
|
|
|
def run(
|
|
self,
|
|
ohlc: pd.DataFrame,
|
|
initial_balance: float = 1000.0,
|
|
risk_per_trade: float = 0.02,
|
|
) -> BacktestResult:
|
|
"""Execute backtest on the provided OHLC data.
|
|
|
|
Args:
|
|
ohlc: DataFrame with [open, high, low, close, volume].
|
|
initial_balance: Starting capital in quote currency.
|
|
risk_per_trade: Fraction of balance to risk per trade.
|
|
|
|
Returns:
|
|
BacktestResult with all metrics.
|
|
"""
|
|
balance = initial_balance
|
|
equity_curve = [balance]
|
|
trades: List[BacktestTrade] = []
|
|
min_bars = max(self.engine.swing_length + 10, 60)
|
|
|
|
# Active position state
|
|
in_position = False
|
|
pos_direction: Optional[TradeDirection] = None
|
|
pos_entry_price = 0.0
|
|
pos_amount = 0.0
|
|
pos_sl = 0.0
|
|
pos_tp = 0.0
|
|
pos_entry_idx = 0
|
|
pos_trailing: Optional[float] = None
|
|
pos_confluence = 0
|
|
pos_candles = 0
|
|
|
|
logger.info(
|
|
"Backtest starting: {} candles, balance={}", len(ohlc), initial_balance
|
|
)
|
|
|
|
for i in range(min_bars, len(ohlc)):
|
|
window = ohlc.iloc[:i + 1]
|
|
current_price = float(window["close"].iloc[-1])
|
|
high = float(window["high"].iloc[-1])
|
|
low = float(window["low"].iloc[-1])
|
|
|
|
if in_position:
|
|
pos_candles += 1
|
|
|
|
# Check exit using high/low for SL/TP accuracy
|
|
exit_price = current_price
|
|
exit_reason: Optional[ExitReason] = None
|
|
|
|
if pos_direction == TradeDirection.LONG:
|
|
if low <= pos_sl:
|
|
exit_price = pos_sl
|
|
exit_reason = ExitReason.STOP_LOSS
|
|
elif high >= pos_tp:
|
|
exit_price = pos_tp
|
|
exit_reason = ExitReason.TAKE_PROFIT
|
|
else:
|
|
if high >= pos_sl:
|
|
exit_price = pos_sl
|
|
exit_reason = ExitReason.STOP_LOSS
|
|
elif low <= pos_tp:
|
|
exit_price = pos_tp
|
|
exit_reason = ExitReason.TAKE_PROFIT
|
|
|
|
# Update trailing
|
|
if exit_reason is None:
|
|
pos_trailing = self.exit_rules.update_trailing_stop(
|
|
pos_direction, pos_entry_price, current_price, pos_trailing
|
|
)
|
|
if pos_trailing is not None:
|
|
if pos_direction == TradeDirection.LONG and low <= pos_trailing:
|
|
exit_price = pos_trailing
|
|
exit_reason = ExitReason.TRAILING_STOP
|
|
elif pos_direction == TradeDirection.SHORT and high >= pos_trailing:
|
|
exit_price = pos_trailing
|
|
exit_reason = ExitReason.TRAILING_STOP
|
|
|
|
# Time exit
|
|
if exit_reason is None and pos_candles >= self.exit_rules.time_exit_candles:
|
|
exit_price = current_price
|
|
exit_reason = ExitReason.TIME_EXIT
|
|
|
|
# CHOCH check
|
|
if exit_reason is None:
|
|
try:
|
|
signals = self.engine.analyze(window)
|
|
choch = signals.latest_choch
|
|
if choch is not None:
|
|
if pos_direction == TradeDirection.LONG and choch == -1:
|
|
exit_price = current_price
|
|
exit_reason = ExitReason.CHOCH
|
|
elif pos_direction == TradeDirection.SHORT and choch == 1:
|
|
exit_price = current_price
|
|
exit_reason = ExitReason.CHOCH
|
|
except Exception:
|
|
pass
|
|
|
|
if exit_reason is not None:
|
|
# Apply slippage
|
|
if pos_direction == TradeDirection.LONG:
|
|
exit_price *= (1 - self.SLIPPAGE_RATE)
|
|
pnl = (exit_price - pos_entry_price) * pos_amount
|
|
else:
|
|
exit_price *= (1 + self.SLIPPAGE_RATE)
|
|
pnl = (pos_entry_price - exit_price) * pos_amount
|
|
|
|
fee = exit_price * pos_amount * self.FEE_RATE
|
|
pnl -= fee
|
|
|
|
balance += pnl
|
|
trades.append(
|
|
BacktestTrade(
|
|
entry_index=pos_entry_idx,
|
|
exit_index=i,
|
|
direction=pos_direction,
|
|
entry_price=pos_entry_price,
|
|
exit_price=exit_price,
|
|
amount=pos_amount,
|
|
pnl=pnl,
|
|
fee=fee,
|
|
exit_reason=exit_reason.value,
|
|
confluence_score=pos_confluence,
|
|
)
|
|
)
|
|
in_position = False
|
|
pos_trailing = None
|
|
pos_candles = 0
|
|
|
|
else:
|
|
# Look for entry
|
|
try:
|
|
signals = self.engine.analyze(window)
|
|
except Exception:
|
|
equity_curve.append(balance)
|
|
continue
|
|
|
|
# Simple bias from BOS
|
|
bos = signals.latest_bos
|
|
if bos is None:
|
|
equity_curve.append(balance)
|
|
continue
|
|
|
|
if bos == 1:
|
|
entry_result = self.entry_rules.check_bullish_entry(signals, current_price)
|
|
direction = TradeDirection.LONG
|
|
elif bos == -1:
|
|
entry_result = self.entry_rules.check_bearish_entry(signals, current_price)
|
|
direction = TradeDirection.SHORT
|
|
else:
|
|
equity_curve.append(balance)
|
|
continue
|
|
|
|
if entry_result.is_valid:
|
|
sl = self.entry_rules.calculate_stop_loss(direction, signals, current_price)
|
|
tp = self.entry_rules.calculate_take_profit(direction, signals, current_price, sl)
|
|
|
|
price_risk = abs(current_price - sl)
|
|
if price_risk == 0:
|
|
equity_curve.append(balance)
|
|
continue
|
|
|
|
pos_amount = (balance * risk_per_trade) / price_risk
|
|
entry_fee = current_price * pos_amount * self.FEE_RATE
|
|
entry_price = current_price * (
|
|
(1 + self.SLIPPAGE_RATE) if direction == TradeDirection.LONG
|
|
else (1 - self.SLIPPAGE_RATE)
|
|
)
|
|
|
|
balance -= entry_fee
|
|
|
|
in_position = True
|
|
pos_direction = direction
|
|
pos_entry_price = entry_price
|
|
pos_sl = sl
|
|
pos_tp = tp
|
|
pos_entry_idx = i
|
|
pos_confluence = len(entry_result.conditions_met)
|
|
|
|
equity_curve.append(balance)
|
|
|
|
result = self._calculate_metrics(trades, equity_curve, initial_balance)
|
|
logger.info("Backtest complete: {}", result.summary())
|
|
return result
|
|
|
|
def _calculate_metrics(
|
|
self,
|
|
trades: List[BacktestTrade],
|
|
equity_curve: List[float],
|
|
initial_balance: float,
|
|
) -> BacktestResult:
|
|
"""Compute performance metrics from trade list and equity curve."""
|
|
total = len(trades)
|
|
wins = [t for t in trades if t.pnl > 0]
|
|
losses = [t for t in trades if t.pnl <= 0]
|
|
|
|
total_pnl = sum(t.pnl for t in trades)
|
|
gross_profit = sum(t.pnl for t in wins) if wins else 0
|
|
gross_loss = abs(sum(t.pnl for t in losses)) if losses else 0
|
|
|
|
# Max drawdown
|
|
peak = initial_balance
|
|
max_dd = 0.0
|
|
for eq in equity_curve:
|
|
if eq > peak:
|
|
peak = eq
|
|
dd = (peak - eq) / peak if peak > 0 else 0
|
|
if dd > max_dd:
|
|
max_dd = dd
|
|
|
|
# Sharpe ratio (simplified, daily returns)
|
|
if len(equity_curve) > 1:
|
|
returns = pd.Series(equity_curve).pct_change().dropna()
|
|
if returns.std() > 0:
|
|
sharpe = (returns.mean() / returns.std()) * (252 ** 0.5)
|
|
else:
|
|
sharpe = 0.0
|
|
else:
|
|
sharpe = 0.0
|
|
|
|
# Average duration
|
|
avg_dur = (
|
|
sum(t.exit_index - t.entry_index for t in trades) / total
|
|
if total > 0 else 0
|
|
)
|
|
|
|
return BacktestResult(
|
|
trades=trades,
|
|
equity_curve=equity_curve,
|
|
total_pnl=total_pnl,
|
|
roi_percent=total_pnl / initial_balance if initial_balance > 0 else 0,
|
|
win_rate=len(wins) / total if total > 0 else 0,
|
|
profit_factor=gross_profit / gross_loss if gross_loss > 0 else float("inf"),
|
|
sharpe_ratio=sharpe,
|
|
max_drawdown=max_dd,
|
|
total_trades=total,
|
|
winning_trades=len(wins),
|
|
losing_trades=len(losses),
|
|
avg_trade_duration=avg_dur,
|
|
initial_balance=initial_balance,
|
|
)
|
|
|
|
@staticmethod
|
|
def generate_report(result: BacktestResult) -> str:
|
|
"""Generate a human-readable backtest report."""
|
|
lines = [
|
|
"=" * 50,
|
|
" ICT Strategy Backtest Report",
|
|
"=" * 50,
|
|
f" Initial Balance : ${result.initial_balance:,.2f}",
|
|
f" Final Balance : ${result.initial_balance + result.total_pnl:,.2f}",
|
|
f" Total PnL : ${result.total_pnl:,.2f}",
|
|
f" ROI : {result.roi_percent:.2%}",
|
|
"-" * 50,
|
|
f" Total Trades : {result.total_trades}",
|
|
f" Winning : {result.winning_trades}",
|
|
f" Losing : {result.losing_trades}",
|
|
f" Win Rate : {result.win_rate:.1%}",
|
|
f" Profit Factor : {result.profit_factor:.2f}",
|
|
"-" * 50,
|
|
f" Sharpe Ratio : {result.sharpe_ratio:.2f}",
|
|
f" Max Drawdown : {result.max_drawdown:.2%}",
|
|
f" Avg Duration : {result.avg_trade_duration:.1f} candles",
|
|
"=" * 50,
|
|
]
|
|
return "\n".join(lines)
|