En tant qu'ingénieur quantitatif ayant passé 3 ans à reconstruire des stratégies de market-making sur des plateformes décentralisées, je me souviens vividly de cette nuit de novembre 2025 où j'ai compris que mes backtests étaient fondamentalement biaisés. Mes modèles affichaient un Sharpe de 3.2 sur Backpack, mais en live trading, le slippage mangeait 40% de mes profits. Le problème ? Je backtestais avec des données tickernized стандартные, sans la profondeur de book accurate que seuls les carnets d'ordres historiques haute fréquence peuvent fournir.
Dans cet article, je vais vous montrer comment j'ai résolu ce problème en combinant HolySheep AI pour le calcul intensif des modèles de prédiction avec l'API Tardis pour les données historiques d'orderbook. Nous allons implémenter un pipeline complet de backtesting pour FTX-Restart, Backpack et Aevo — trois exchanges où l'accès à des données granulares est crucial pour la précision des stratégies derivatives.
为什么需要 HolySheep + Tardis 的组合？
Avant d'entrer dans le code, laissez-moi vous expliquer pourquoi j'ai choisi cette architecture après avoir testé une dizaine d'alternatives. Tardis propose l'API REST et WebSocket la plus complète pour les données historiques de carnets d'ordres, avec une granularité jusqu'au niveau du tick pour FTX-Restart (réplication exacte de l'historique) et Backpack. HolySheep, de son côté, offre des modèles de deep learning avec une latence moyenne de 32ms pour les appels synchrones — essentielle quand votre pipeline de backtesting traite des millions de ticks.
Le coût est également un facteur déterminant. Avec les prix HolySheep 2026 (DeepSeek V3.2 à $0.42/MTok, soit 85% moins cher que GPT-4.1 à $8/MTok), je peux maintenant entraîner mes modèles de prédiction de liquidité sur des datasets de 50 Go sans exploser mon budget recherche. Aevo, étant un exchange perppetual sur couche 2, nécessite des données avec latence minimale pour capturer les squeezes de liquidité — une combinaison que seul ce stack peut offrir de manière fiable.
先决条件与架构概述
Architecture du Pipeline de Backtesting
┌─────────────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE BACKTESTING HYBRID │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ TARDIS │ │ HOLYSHEEP │ │ BACKTEST │ │
│ │ API │─────▶│ AI │─────▶│ ENGINE │ │
│ │ │ │ (inférence) │ │ (Python/C++)│ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ Orderbook Historique Modèles ML Métriques PnL │
│ - FTX-Restart - Prédiction slippage - Sharpe ratio │
│ - Backpack - Estimation liquidité - Max drawdown │
│ - Aevo - Détection anomalies - Win rate │
│ │
│ LATENCE TYPIQUE: │
│ - HolySheep inference: 32ms (DeepSeek V3.2) │
│ - Tardis REST response: 120ms (p95) │
│ - Pipeline complet: ~200ms/tick │
│ │
└─────────────────────────────────────────────────────────────────────────┘
先决条件安装
# Installation des dépendances
pip install requests websockets asyncio pandas numpy
pip install tardis-client # API Tardis officielle
pip install holy-sheep-sdk # SDK HolySheep (si disponible)
Vérification de la configuration
python -c "import tardis; import holy_sheep; print('✅ Config OK')"
Variables d'environnement requises
export TARDIS_API_KEY="your_tardis_api_key_here"
export HOLYSHEEP_API_KEY="your_holy_sheep_api_key_here"
实现：Tardis 数据获取
La première étape consiste à récupérer les données historiques d'orderbook via l'API Tardis. J'ai choisi cette API parce qu'elle propose des données rejouables pour FTX-Restart — une exact replica de l'historique de l exchange original avant sa fermeture — ce qui est invaluable pour backtester des stratégies qui auraient fonctionné sur l'original.
#!/usr/bin/env python3
"""
Récupération des données Orderbook via Tardis API
Support: FTX-Restart, Backpack, Aevo
"""
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class OrderBookSnapshot:
exchange: str
symbol: str
timestamp: datetime
bids: List[tuple] # [(price, size), ...]
asks: List[tuple] # [(price, size), ...]
class TardisOrderbookFetcher:
"""Classe pour récupérer les données historiques d'orderbook depuis Tardis"""
BASE_URL = "https://api.tardis.dev/v1"
# Exchange IDs supportés par Tardis
SUPPORTED_EXCHANGES = {
'ftx-restart': {
'id': 'ftx-restart',
'name': 'FTX Restart',
'symbols': ['BTC-PERP', 'ETH-PERP', 'SOL-PERP', 'AVAX-PERP'],
'has_orderbook': True,
'has_trades': True
},
'backpack': {
'id': 'backpack',
'name': 'Backpack Exchange',
'symbols': ['SOL-PERP', 'BTC-PERP', 'ETH-PERP'],
'has_orderbook': True,
'has_trades': True
},
'aevo': {
'id': 'aevo',
'name': 'Aevo L2 Perpetuals',
'symbols': ['ETH-PERP', 'BTC-PERP', 'LINK-PERP'],
'has_orderbook': True,
'has_trades': True
}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def get_orderbook_snapshot(
self,
exchange: str,
symbol: str,
timestamp: datetime
) -> Optional[OrderBookSnapshot]:
"""
Récupère un snapshot d'orderbook à un timestamp donné.
Granularité: 1 seconde pour FTX-Restart, 100ms pour Backpack/Aevo
"""
endpoint = f"{self.BASE_URL}/orderbooks/{exchange}/{symbol}"
params = {
'from': timestamp.isoformat(),
'to': (timestamp + timedelta(seconds=1)).isoformat(),
'limit': 1000, # Profondeur max du book
'format': 'array' # Format optimisé pour le parsing
}
try:
response = self.session.get(endpoint, params=params, timeout=30)
response.raise_for_status()
data = response.json()
return OrderBookSnapshot(
exchange=exchange,
symbol=symbol,
timestamp=timestamp,
bids=data.get('bids', []),
asks=data.get('asks', [])
)
except requests.exceptions.RequestException as e:
print(f"❌ Erreur Tardis API: {e}")
return None
def fetch_historical_orderbook(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
interval_seconds: int = 60
) -> List[OrderBookSnapshot]:
"""
Télécharge une série temporelle complète d'orderbooks.
Coût approximatif: $0.10/1M ticks pour FTX-Restart
"""
snapshots = []
current = start_date
while current < end_date:
snapshot = self.get_orderbook_snapshot(
exchange, symbol, current
)
if snapshot:
snapshots.append(snapshot)
print(f"✅ {exchange}/{symbol} @ {current.isoformat()} "
f"(bids: {len(snapshot.bids)}, asks: {len(snapshot.asks)})")
current += timedelta(seconds=interval_seconds)
return snapshots
def estimate_cost(self, exchange: str, days: int) -> Dict:
"""
Estime le coût d'un téléchargement de données.
Prix Tardis 2026: $0.08-0.15/1M messages selon l'exchange
"""
# Estimation basée sur la volatilité moyenne
messages_per_day = {
'ftx-restart': 50_000_000, # Haute activité perp
'backpack': 25_000_000,
'aevo': 15_000_000
}
total_messages = messages_per_day.get(exchange, 20_000_000) * days
price_per_million = {
'ftx-restart': 0.12,
'backpack': 0.08,
'aevo': 0.10
}
cost = (total_messages / 1_000_000) * price_per_million.get(exchange, 0.10)
return {
'exchange': exchange,
'days': days,
'estimated_messages': total_messages,
'estimated_cost_usd': round(cost, 2),
'estimated_cost_cny': round(cost * 7.2, 2)
}
=============================================================================
EXEMPLE D'UTILISATION
=============================================================================
if __name__ == "__main__":
# Initialisation du fetcher
fetcher = TardisOrderbookFetcher(api_key="your_tardis_api_key")
# Estimation du coût pour 30 jours sur FTX-Restart BTC-PERP
cost_estimate = fetcher.estimate_cost('ftx-restart', days=30)
print(f"💰 Coût estimé: ${cost_estimate['estimated_cost_usd']} USD "
f"({cost_estimate['estimated_cost_cny']} CNY)")
# Téléchargement d'un sample (1 heure de données)
sample_data = fetcher.fetch_historical_orderbook(
exchange='ftx-restart',
symbol='BTC-PERP',
start_date=datetime(2025, 6, 1, 0, 0, 0),
end_date=datetime(2025, 6, 1, 1, 0, 0), # 1 heure
interval_seconds=60
)
print(f"\n📊 {len(sample_data)} snapshots récupérés")
实现：HolySheep AI 推理集成
Maintenant, passons à la partie la plus interessante — l'intégration de HolySheep pour la prédiction de slippage et l'estimation de liquidité en temps réel. J'utilise personnellement DeepSeek V3.2 pour cette tâche car son coût de $0.42/MTok me permet de traiter des millions de ticks sans surveiller ma facture. La latence moyenne de 32ms est parfaitement acceptable pour du backtesting batch.
#!/usr/bin/env python3
"""
Intégration HolySheep AI pour la prédiction de slippage et liquidité
base_url: https://api.holysheep.ai/v1
"""
import requests
import json
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import time
@dataclass
class LiquidityPrediction:
"""Résultat de la prédiction de liquidité"""
estimated_slippage_bps: float # Basis points
liquidity_score: float # 0-1, 1 = très liquide
mid_price_impact: float
confidence: float # 0-1
model_used: str
@dataclass
class OrderBookFeatures:
"""Features extraites de l'orderbook pour le modèle"""
bid_ask_spread: float
mid_price: float
total_bid_depth: float # Somme des sizes bid sur 10 niveaux
total_ask_depth: float
imbalance_ratio: float # bid_depth / ask_depth
top_10_bid_concentration: float
top_10_ask_concentration: float
volatility_estimate: float
class HolySheepMLInference:
"""
Client pour l'inférence ML via HolySheep AI.
Optimisé pour les modèles de prédiction de slippage.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
# Statistiques de latence (monitoring)
self.latencies = []
def _call_model(
self,
model: str,
prompt: str,
temperature: float = 0.1,
max_tokens: int = 512
) -> Optional[Dict]:
"""
Appel générique à un modèle HolySheep.
Latence typique: 32-150ms selon le modèle
"""
start_time = time.time()
payload = {
'model': model,
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': temperature,
'max_tokens': max_tokens
}
try:
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
latency_ms = (time.time() - start_time) * 1000
self.latencies.append(latency_ms)
return result
except requests.exceptions.RequestException as e:
print(f"❌ Erreur HolySheep API: {e}")
return None
def extract_orderbook_features(
self,
bids: List[Tuple[float, float]],
asks: List[Tuple[float, float]]
) -> OrderBookFeatures:
"""
Extrait les features nécessaires pour la prédiction de slippage.
"""
if not bids or not asks:
raise ValueError("Orderbook vide")
best_bid = max(bids, key=lambda x: x[0])[0]
best_ask = min(asks, key=lambda x: x[0])[0]
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price
# Profondeur cumulée sur 10 niveaux
bid_depth = sum(size for price, size in bids[:10])
ask_depth = sum(size for price, size in asks[:10])
# Concentration (HHI-like)
total_bid = sum(size for _, size in bids[:10])
total_ask = sum(size for _, size in asks[:10])
top_10_bid_conc = sum(
(size/total_bid)**2 for _, size in bids[:10]
) if total_bid > 0 else 0
top_10_ask_conc = sum(
(size/total_ask)**2 for _, size in asks[:10]
) if total_ask > 0 else 0
return OrderBookFeatures(
bid_ask_spread=spread,
mid_price=mid_price,
total_bid_depth=bid_depth,
total_ask_depth=ask_depth,
imbalance_ratio=bid_depth/ask_depth if ask_depth > 0 else 1,
top_10_bid_concentration=top_10_bid_conc,
top_10_ask_concentration=top_10_ask_conc,
volatility_estimate=0.001 # À calculer sur fenêtre glissante
)
def predict_slippage(
self,
features: OrderBookFeatures,
order_size_usd: float,
is_buy: bool = True,
model: str = 'deepseek-v3.2'
) -> LiquidityPrediction:
"""
Prédit le slippage attendu pour un ordre de taille donnée.
Utilise DeepSeek V3.2 ($0.42/MTok) pour l'analyse.
"""
prompt = f"""Analyse de liquidité pour orderbook:
Données de marché:
- Spread bid-ask: {features.bid_ask_spread:.6f} ({features.bid_ask_spread*10000:.2f} bps)
- Mid price: ${features.mid_price:.2f}
- Profondeur bid (10 niveaux): {features.total_bid_depth:.4f} BTC
- Profondeur ask (10 niveaux): {features.total_ask_depth:.4f} BTC
- Ratio d'imbalance: {features.imbalance_ratio:.4f}
- Concentration bid: {features.top_10_bid_concentration:.4f}
- Concentration ask: {features.top_10_ask_concentration:.4f}
Ordre à exécuter:
- Direction: {'ACHAT' if is_buy else 'VENTE'}
- Taille: ${order_size_usd:.2f}
Réponds uniquement au format JSON suivant (sans markdown):
{{"slippage_bps": float, "liquidity_score": float, "mid_price_impact": float, "confidence": float}}
"""
result = self._call_model(model, prompt, temperature=0.1, max_tokens=256)
if result and 'choices' in result:
content = result['choices'][0]['message']['content'].strip()
# Parse JSON response
try:
data = json.loads(content)
return LiquidityPrediction(
estimated_slippage_bps=data['slippage_bps'],
liquidity_score=data['liquidity_score'],
mid_price_impact=data['mid_price_impact'],
confidence=data['confidence'],
model_used=model
)
except json.JSONDecodeError:
# Fallback si le modèle ne retourne pas du JSON valide
return self._fallback_prediction(features, order_size_usd)
return self._fallback_prediction(features, order_size_usd)
def _fallback_prediction(
self,
features: OrderBookFeatures,
order_size_usd: float
) -> LiquidityPrediction:
"""
Prediction heuristique si l'API échoue.
Basée sur le modèle linéaire simplifié de slippage.
"""
size_btc = order_size_usd / features.mid_price
# Impact proportionnel à la taille relative
depth = features.total_ask_depth if True else features.total_bid_depth
relative_size = size_btc / depth if depth > 0 else 0.1
# Spread + impact de taille
slippage_bps = (features.bid_ask_spread * 10000 / 2) + (relative_size * 100)
return LiquidityPrediction(
estimated_slippage_bps=min(slippage_bps, 500), # Cap à 5%
liquidity_score=max(0, 1 - relative_size),
mid_price_impact=relative_size * 0.01,
confidence=0.6,
model_used='heuristic-fallback'
)
def batch_predict_slippage(
self,
orderbooks: List[Tuple[List, List]], # [(bids, asks), ...]
order_size_usd: float,
is_buy: bool = True
) -> List[LiquidityPrediction]:
"""
Traite un batch de snapshots orderbook.
Optimisé pour le backtesting: traitement asynchrone.
"""
predictions = []
for bids, asks in orderbooks:
try:
features = self.extract_orderbook_features(bids, asks)
pred = self.predict_slippage(features, order_size_usd, is_buy)
predictions.append(pred)
except Exception as e:
print(f"⚠️ Erreur processing: {e}")
predictions.append(None)
return predictions
def get_stats(self) -> Dict:
"""Retourne les statistiques de latence et coût"""
if not self.latencies:
return {'error': 'Aucune donnée'}
sorted_latencies = sorted(self.latencies)
return {
'total_calls': len(self.latencies),
'avg_latency_ms': sum(self.latencies) / len(self.latencies),
'p50_latency_ms': sorted_latencies[len(sorted_latencies)//2],
'p95_latency_ms': sorted_latencies[int(len(sorted_latencies)*0.95)],
'p99_latency_ms': sorted_latencies[int(len(sorted_latencies)*0.99)],
'estimated_cost_usd': len(self.latencies) * 0.00001 # ~$0.01/1000 calls
}
=============================================================================
EXEMPLE D'UTILISATION
=============================================================================
if __name__ == "__main__":
# Initialisation du client HolySheep
client = HolySheepMLInference(api_key="YOUR_HOLYSHEEP_API_KEY")
# Exemple d'orderbook (simulé pour FTX-Restart BTC-PERP)
sample_bids = [
(65000.0, 2.5), (64999.5, 1.2), (64999.0, 3.0),
(64998.5, 1.8), (64998.0, 2.1), (64997.5, 0.9),
(64997.0, 1.5), (64996.5, 2.3), (64996.0, 1.0),
(64995.5, 1.7)
]
sample_asks = [
(65001.0, 2.3), (65001.5, 1.5), (65002.0, 2.8),
(65002.5, 1.9), (65003.0, 1.2), (65003.5, 2.0),
(65004.0, 0.8), (65004.5, 1.6), (65005.0, 2.2),
(65005.5, 1.4)
]
# Extraction des features
features = client.extract_orderbook_features(sample_bids, sample_asks)
print(f"📊 Features extraites:")
print(f" - Spread: {features.bid_ask_spread*10000:.2f} bps")
print(f" - Mid price: ${features.mid_price:.2f}")
print(f" - Imbalance: {features.imbalance_ratio:.4f}")
# Prédiction de slippage pour un ordre de $100,000
prediction = client.predict_slippage(
features=features,
order_size_usd=100_000,
is_buy=True,
model='deepseek-v3.2'
)
print(f"\n🎯 Prédiction slippage:")
print(f" - Slippage estimé: {prediction.estimated_slippage_bps:.2f} bps "
f"(${prediction.estimated_slippage_bps/10000 * 100_000:.2f})")
print(f" - Score liquidité: {prediction.liquidity_score:.4f}")
print(f" - Confiance: {prediction.confidence:.2f}")
print(f" - Modèle: {prediction.model_used}")
# Statistiques de performance
stats = client.get_stats()
print(f"\n📈 Performance HolySheep:")
print(f" - Latence moyenne: {stats.get('avg_latency_ms', 0):.2f}ms")
print(f" - Latence p95: {stats.get('p95_latency_ms', 0):.2f}ms")
print(f" - Coût estimé: ${stats.get('estimated_cost_usd', 0):.4f}")
完整回测引擎实现
#!/usr/bin/env python3
"""
Backtest Engine pour Stratégies Derivatives
Combine: Tardis (données) + HolySheep (ML) + Calcul local (PnL)
Exchange supportés: FTX-Restart, Backpack, Aevo
"""
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime, timedelta
from enum import Enum
import json
class Exchange(Enum):
FTX_RESTART = 'ftx-restart'
BACKPACK = 'backpack'
AEVO = 'aevo'
@dataclass
class TradeSignal:
"""Signal de trading généré par la stratégie"""
timestamp: datetime
direction: int # 1 = long, -1 = short, 0 = flat
size_usd: float
entry_price: float
stop_loss: Optional[float] = None
take_profit: Optional[float] = None
confidence: float = 1.0
@dataclass
class BacktestTrade:
"""Trade exécuté en backtest"""
entry_time: datetime
exit_time: datetime
direction: int
entry_price: float
exit_price: float
size_usd: float
pnl_usd: float
pnl_pct: float
slippage_bps: float
fees_usd: float
hold_duration_minutes: float
@dataclass
class BacktestResult:
"""Résultat complet du backtest"""
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl_usd: float
max_drawdown_pct: float
sharpe_ratio: float
sortino_ratio: float
avg_trade_pnl: float
avg_slippage_bps: float
avg_fees_usd: float
avg_hold_time_minutes: float
trades: List[BacktestTrade]
def to_dataframe(self) -> pd.DataFrame:
"""Convertit les résultats en DataFrame pandas"""
return pd.DataFrame([
{
'entry_time': t.entry_time,
'exit_time': t.exit_time,
'direction': 'Long' if t.direction == 1 else 'Short',
'entry_price': t.entry_price,
'exit_price': t.exit_price,
'size_usd': t.size_usd,
'pnl_usd': t.pnl_usd,
'pnl_pct': t.pnl_pct,
'slippage_bps': t.slippage_bps,
'fees_usd': t.fees_usd,
'hold_minutes': t.hold_duration_minutes
}
for t in self.trades
])
class DerivativesBacktester:
"""
Moteur de backtesting haute performance pour stratégies derivatives.
"""
# Frais de trading par exchange (maker + taker moyens)
EXCHANGE_FEES = {
Exchange.FTX_RESTART: {'maker': 0.0002, 'taker': 0.0007},
Exchange.BACKPACK: {'maker': 0.0001, 'taker': 0.0005},
Exchange.AEVO: {'maker': 0.0000, 'taker': 0.0003} # Frais L2 réduits
}
def __init__(
self,
tardis_fetcher,
holy_sheep_client,
initial_balance_usd: float = 100_000,
max_position_size_pct: float = 0.1, # 10% max du capital
max_slippage_bps: float = 50, # Reject si slippage > 50bps
):
self.tardis = tardis_fetcher
self.holy_sheep = holy_sheep_client
self.initial_balance = initial_balance_usd
self.max_position_pct = max_position_size_pct
self.max_slippage = max_slippage_bps
self.balance = initial_balance_usd
self.position = 0
self.position_side = 0 # 1 = long, -1 = short, 0 = flat
self.trades: List[BacktestTrade] = []
self.equity_curve = [initial_balance_usd]
def run_backtest(
self,
exchange: Exchange,
symbol: str,
start_date: datetime,
end_date: datetime,
strategy_fn: Callable, # Fonction de génération de signaux
order_interval_seconds: int = 60
) -> BacktestResult:
"""
Exécute le backtest complet.
Args:
exchange: Exchange à backtester
symbol: Paire de trading (ex: 'BTC-PERP')
start_date: Date de début
end_date: Date de fin
strategy_fn: Fonction qui prend (timestamp, orderbook_features)
et retourne un TradeSignal ou None
order_interval_seconds: Intervalle entre chaque signal généré
"""
print(f"🚀 Démarrage backtest: {exchange.value}/{symbol}")
print(f" Période: {start_date.date()} → {end_date.date()}")
print(f" Capital initial: ${self.initial_balance:,.2f}")
# Reset state
self.balance = self.initial_balance
self.position = 0
self.position_side = 0
self.trades = []
self.equity_curve = [self.initial_balance]
current = start_date
pending_signal: Optional[TradeSignal] = None
while current < end_date:
# 1. Récupération des données orderbook
snapshot = self.tardis.get_orderbook_snapshot(
exchange.value, symbol, current
)
if snapshot is None:
current += timedelta(seconds=order_interval_seconds)
continue
# 2. Feature extraction + prédiction ML
try:
features = self.holy_sheep.extract_orderbook_features(
snapshot.bids, snapshot.asks
)
# Taille de position basée sur la liquidité
max_size = self.balance * self.max_position_pct
liquidity_factor = features.total_bid_depth / (features.total_ask_depth + 1)
adjusted_size = min(max_size, features.mid_price * liquidity_factor * 10)
# 3. Génération du signal via stratégie
signal = strategy_fn(current, features, adjusted_size)
if signal and self.position_side == 0:
# Calcul du slippage prédit
pred = self.holy_sheep.predict_slippage(
features, signal.size_usd, signal.direction == 1
)
# Reject si slippage trop élevé
if pred.estimated_slippage_bps > self.max_slippage:
print(f"⚠️ Slippage {pred.estimated_slippage_bps:.1f}bps "
f"> max {self.max_slippage}bps - trade skippé")
current += timedelta(seconds=order_interval_seconds)
continue
# 4. Exécution du trade (entry)
entry_price = features.mid_price * (1 + 0.0001 * signal.direction)
fees = self.EXCHANGE_FEES[exchange]['taker'] * signal.size_usd
pending_signal = TradeSignal(
timestamp=current,
direction=signal.direction,
size_usd=signal.size_usd,
entry_price=entry_price,
slippage_predicted=pred.estimated_slippage_bps
)
self.position = signal.size_usd / entry_price * signal.direction
self.position_side = signal.direction
self.balance -= fees
print(f"📈 Entry {signal.direction} @ ${entry_price:.2f} "
f"(slippage: {pred.estimated_slippage_bps:.2f}bps)")
elif pending_signal and self._check_exit_conditions(
features, pending_signal
):
# 5. Exit du trade
exit_price = features.mid_price
fees = self.EXCHANGE_FEES[exchange]['taker'] * abs(self.position * exit_price)
pnl = (exit_price - pending_signal.entry_price) * self.position
pnl -= fees
pnl -= abs(pending_signal.slippage_predicted / 10000 * pending_signal.size_usd)
trade = BacktestTrade(
entry_time=pending_signal.timestamp,
exit_time=current,
direction=pending_signal.direction,
entry_price=pending_signal.entry_price,
exit_price=exit_price,
size_usd=pending_signal.size_usd,
pnl_usd=pnl,
pnl_pct=pnl / pending_signal.size_usd * 100,
slippage_bps=pending_signal.slippage_predicted,
fees_usd=fees,
hold_duration_minutes=(
current - pending_signal.timestamp
).total_seconds() / 60
)
self.trades.append(trade)
self.balance += pnl
self.equity_curve.append(self.balance)
print(f"📉 Exit @ ${exit_price:.2f} | PnL: ${pnl:.2f} "
f"({trade.pnl_pct:.2f}%)")
self.position = 0
self.position_side = 0
pending_signal = None
except Exception as e:
print(f"❌ Erreur: {e}")
current += timedelta(seconds=order_interval_seconds)
# Log every 1000 iterations
if (current - start_date).total_seconds() % 100000 < order_interval_seconds:
print(f" Progress: {(current - start_date).days} jours | "
f"Balance: ${self.balance:,.2f}")
return self._compute_results()
def _check_exit_conditions(
self,
features,
signal: TradeSignal
) -> bool:
"""Vérifie les conditions de sortie"""
if signal.stop_loss and signal.direction == 1:
if features.mid_price <= signal.stop_loss:
return True
if signal.take_profit and signal.direction == 1:
if features.mid_price >= signal.take_profit:
return True
return False
def _compute_results(self) -> BacktestResult:
"""Calcule les métriques finales du backtest"""
if not self.trades:
return BacktestResult(
total_trades=0, winning_trades=0, losing_trades=0,
win_rate=0, total_pnl_usd=0, max_drawdown_pct=0,
sharpe_ratio=0, sortino_ratio=0, avg_trade_pnl=0,
avg_slippage_bps=0, avg_fees_usd=0, avg_hold_time_minutes=0,
trades=[]
)
df = pd.DataFrame([
{'pnl': t.pnl_usd, 'slippage': t.slippage_bps,