tika/trader-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: trading bot MVP — ICT Order Block + Liquidity Sweep strategy

Browse Source 
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>
This commit is contained in:
tika
2026-02-24 23:25:51 +01:00
commit 4df8d53b1a
58 changed files with 7484 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
backend/app/core/backtester.py Normal file
View File

@@ -0,0 +1,269 @@
"""
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,
)