tika
4df8d53b1a
Full-stack trading bot with: - FastAPI backend with ICT strategy (Order Block + Liquidity Sweep detection) - Backtester engine with rolling window, spread simulation, and performance metrics - Hybrid market data service (yfinance + TwelveData with rate limiting + SQLite cache) - Simulated exchange for paper trading - React/TypeScript frontend with TradingView lightweight-charts v5 - Live dashboard with candlestick chart, OHLC legend, trade markers - Backtest page with configurable parameters, equity curve, and trade table - WebSocket support for real-time updates - Bot runner with asyncio loop for automated trading Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
270 lines
10 KiB
Python
270 lines
10 KiB
Python
"""
|
|
Moteur de backtesting vectorisé.
|
|
|
|
Simule l'exécution de la stratégie sur des données historiques :
|
|
- Rolling window : analyse les N dernières bougies à chaque pas
|
|
- Gestion du spread OANDA
|
|
- Suivi des positions ouvertes (SL / TP hit)
|
|
- Calcul des métriques de performance
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from app.core.strategy.base import AbstractStrategy, TradeSignal
|
|
|
|
|
|
@dataclass
|
|
class BacktestTrade:
|
|
direction: str
|
|
entry_price: float
|
|
stop_loss: float
|
|
take_profit: float
|
|
entry_time: datetime
|
|
exit_price: Optional[float] = None
|
|
exit_time: Optional[datetime] = None
|
|
pnl_pips: Optional[float] = None
|
|
pnl_pct: Optional[float] = None
|
|
status: str = "open" # "win" | "loss" | "open"
|
|
signal_type: str = ""
|
|
|
|
|
|
@dataclass
|
|
class BacktestMetrics:
|
|
initial_balance: float
|
|
final_balance: float
|
|
total_pnl: float
|
|
total_pnl_pct: float
|
|
total_trades: int
|
|
winning_trades: int
|
|
losing_trades: int
|
|
win_rate: float
|
|
avg_win_pips: float
|
|
avg_loss_pips: float
|
|
expectancy: float
|
|
max_drawdown: float
|
|
max_drawdown_pct: float
|
|
sharpe_ratio: float
|
|
profit_factor: float
|
|
equity_curve: list[dict] = field(default_factory=list)
|
|
trades: list[BacktestTrade] = field(default_factory=list)
|
|
|
|
|
|
class Backtester:
|
|
def __init__(
|
|
self,
|
|
strategy: AbstractStrategy,
|
|
initial_balance: float = 10_000.0,
|
|
risk_percent: float = 1.0,
|
|
spread_pips: float = 1.5,
|
|
pip_value: float = 0.0001,
|
|
) -> None:
|
|
self.strategy = strategy
|
|
self.initial_balance = initial_balance
|
|
self.risk_percent = risk_percent # % du capital risqué par trade
|
|
self.spread_pips = spread_pips
|
|
self.pip_value = pip_value
|
|
|
|
def run(
|
|
self,
|
|
df: pd.DataFrame,
|
|
window: int = 100,
|
|
) -> BacktestMetrics:
|
|
"""
|
|
Exécute le backtest sur le DataFrame complet.
|
|
|
|
window : nombre de bougies analysées à chaque pas (rolling)
|
|
"""
|
|
df = df.copy().reset_index(drop=True)
|
|
trades: list[BacktestTrade] = []
|
|
balance = self.initial_balance
|
|
equity_curve: list[dict] = [{"time": str(df.iloc[0]["time"]), "balance": balance}]
|
|
open_trade: Optional[BacktestTrade] = None
|
|
last_signal_time: Optional[datetime] = None
|
|
used_signals: set[tuple[str, int]] = set() # (direction, entry_price_rounded)
|
|
|
|
for i in range(window, len(df)):
|
|
candle = df.iloc[i]
|
|
|
|
# 1. Gérer la position ouverte (SL/TP hit ?)
|
|
if open_trade:
|
|
closed, exit_price = self._check_exit(open_trade, candle)
|
|
if closed:
|
|
pnl_pips = self._calc_pnl_pips(open_trade, exit_price)
|
|
pnl_money = self._pips_to_money(pnl_pips, balance)
|
|
balance += pnl_money
|
|
open_trade.exit_price = exit_price
|
|
open_trade.exit_time = candle["time"]
|
|
open_trade.pnl_pips = pnl_pips
|
|
open_trade.pnl_pct = pnl_money / self.initial_balance * 100
|
|
open_trade.status = "win" if pnl_pips > 0 else "loss"
|
|
trades.append(open_trade)
|
|
open_trade = None
|
|
equity_curve.append({"time": str(candle["time"]), "balance": balance})
|
|
|
|
# 2. Si pas de position ouverte, chercher un signal
|
|
if not open_trade:
|
|
slice_df = df.iloc[i - window:i + 1]
|
|
result = self.strategy.analyze(slice_df)
|
|
|
|
if result.signals:
|
|
# Filtrer les signaux déjà exploités (éviter doublons)
|
|
new_signals = [
|
|
s for s in result.signals
|
|
if last_signal_time is None or s.time > last_signal_time
|
|
]
|
|
if not new_signals:
|
|
continue
|
|
signal = new_signals[-1] # Signal le plus récent
|
|
# Appliquer le spread
|
|
entry, sl, tp = self._apply_spread(signal)
|
|
# Éviter de réouvrir le même trade (même direction + même prix)
|
|
sig_key = (signal.direction, round(entry / self.pip_value))
|
|
if sig_key in used_signals:
|
|
continue
|
|
used_signals.add(sig_key)
|
|
last_signal_time = signal.time
|
|
open_trade = BacktestTrade(
|
|
direction=signal.direction,
|
|
entry_price=entry,
|
|
stop_loss=sl,
|
|
take_profit=tp,
|
|
entry_time=candle["time"],
|
|
signal_type=signal.signal_type,
|
|
)
|
|
|
|
# Fermer la position ouverte en fin de période
|
|
if open_trade:
|
|
last_price = df.iloc[-1]["close"]
|
|
pnl_pips = self._calc_pnl_pips(open_trade, last_price)
|
|
pnl_money = self._pips_to_money(pnl_pips, balance)
|
|
balance += pnl_money
|
|
open_trade.exit_price = last_price
|
|
open_trade.exit_time = df.iloc[-1]["time"]
|
|
open_trade.pnl_pips = pnl_pips
|
|
open_trade.pnl_pct = pnl_money / self.initial_balance * 100
|
|
open_trade.status = "win" if pnl_pips > 0 else "loss"
|
|
trades.append(open_trade)
|
|
|
|
equity_curve.append({"time": str(df.iloc[-1]["time"]), "balance": balance})
|
|
return self._compute_metrics(trades, balance, equity_curve)
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def _check_exit(
|
|
self, trade: BacktestTrade, candle: pd.Series
|
|
) -> tuple[bool, float]:
|
|
"""Vérifie si le SL ou TP est touché sur la bougie. Retourne (closed, exit_price)."""
|
|
if trade.direction == "buy":
|
|
if candle["low"] <= trade.stop_loss:
|
|
return True, trade.stop_loss
|
|
if candle["high"] >= trade.take_profit:
|
|
return True, trade.take_profit
|
|
else:
|
|
if candle["high"] >= trade.stop_loss:
|
|
return True, trade.stop_loss
|
|
if candle["low"] <= trade.take_profit:
|
|
return True, trade.take_profit
|
|
return False, 0.0
|
|
|
|
def _apply_spread(self, signal: TradeSignal) -> tuple[float, float, float]:
|
|
"""Applique le spread à l'entrée."""
|
|
half_spread = (self.spread_pips * self.pip_value) / 2
|
|
if signal.direction == "buy":
|
|
entry = signal.entry_price + half_spread
|
|
sl = signal.stop_loss
|
|
tp = signal.take_profit
|
|
else:
|
|
entry = signal.entry_price - half_spread
|
|
sl = signal.stop_loss
|
|
tp = signal.take_profit
|
|
return entry, sl, tp
|
|
|
|
def _calc_pnl_pips(self, trade: BacktestTrade, exit_price: float) -> float:
|
|
if trade.direction == "buy":
|
|
return (exit_price - trade.entry_price) / self.pip_value
|
|
return (trade.entry_price - exit_price) / self.pip_value
|
|
|
|
def _pips_to_money(self, pips: float, balance: float) -> float:
|
|
"""Convertit les pips en unités monétaires selon le risque défini."""
|
|
risk_amount = balance * (self.risk_percent / 100)
|
|
# Nombre de pips de risque sur ce trade
|
|
return pips * (risk_amount / 10) # Approximation simple
|
|
|
|
def _compute_metrics(
|
|
self,
|
|
trades: list[BacktestTrade],
|
|
final_balance: float,
|
|
equity_curve: list[dict],
|
|
) -> BacktestMetrics:
|
|
if not trades:
|
|
return BacktestMetrics(
|
|
initial_balance=self.initial_balance,
|
|
final_balance=final_balance,
|
|
total_pnl=0, total_pnl_pct=0,
|
|
total_trades=0, winning_trades=0, losing_trades=0,
|
|
win_rate=0, avg_win_pips=0, avg_loss_pips=0,
|
|
expectancy=0, max_drawdown=0, max_drawdown_pct=0,
|
|
sharpe_ratio=0, profit_factor=0,
|
|
equity_curve=equity_curve, trades=trades,
|
|
)
|
|
|
|
winners = [t for t in trades if t.status == "win"]
|
|
losers = [t for t in trades if t.status == "loss"]
|
|
|
|
total_pnl = final_balance - self.initial_balance
|
|
win_rate = len(winners) / len(trades) if trades else 0
|
|
|
|
avg_win = np.mean([t.pnl_pips for t in winners]) if winners else 0
|
|
avg_loss = abs(np.mean([t.pnl_pips for t in losers])) if losers else 0
|
|
expectancy = (win_rate * avg_win) - ((1 - win_rate) * avg_loss)
|
|
|
|
# Max Drawdown
|
|
balances = [eq["balance"] for eq in equity_curve]
|
|
peak = balances[0]
|
|
max_dd = 0.0
|
|
for b in balances:
|
|
if b > peak:
|
|
peak = b
|
|
dd = (peak - b) / peak
|
|
if dd > max_dd:
|
|
max_dd = dd
|
|
|
|
# Profit Factor
|
|
gross_profit = sum(t.pnl_pips for t in winners) if winners else 0
|
|
gross_loss = abs(sum(t.pnl_pips for t in losers)) if losers else 1
|
|
pf = gross_profit / gross_loss if gross_loss > 0 else 0
|
|
|
|
# Sharpe Ratio (simplifié — daily returns)
|
|
pnl_series = [t.pnl_pips or 0 for t in trades]
|
|
if len(pnl_series) > 1:
|
|
mean_ret = np.mean(pnl_series)
|
|
std_ret = np.std(pnl_series)
|
|
sharpe = (mean_ret / std_ret * np.sqrt(252)) if std_ret > 0 else 0
|
|
else:
|
|
sharpe = 0
|
|
|
|
return BacktestMetrics(
|
|
initial_balance=self.initial_balance,
|
|
final_balance=final_balance,
|
|
total_pnl=total_pnl,
|
|
total_pnl_pct=total_pnl / self.initial_balance * 100,
|
|
total_trades=len(trades),
|
|
winning_trades=len(winners),
|
|
losing_trades=len(losers),
|
|
win_rate=win_rate * 100,
|
|
avg_win_pips=float(avg_win),
|
|
avg_loss_pips=float(avg_loss),
|
|
expectancy=float(expectancy),
|
|
max_drawdown=max_dd * self.initial_balance,
|
|
max_drawdown_pct=max_dd * 100,
|
|
sharpe_ratio=float(sharpe),
|
|
profit_factor=float(pf),
|
|
equity_curve=equity_curve,
|
|
trades=trades,
|
|
)
|