En tant qu'ingénieur quantitatif ayant travaillé sur des systèmes de trading haute fréquence pendant quatre ans, je peux vous confirmer que la qualité de vos données de test est le facteur déterminant entre une stratégie profitable en backtest et un désastre en production. J'ai personnellement perdu trois mois de travail à cause de données tick mal filtrées qui introduisaient des faux positifs intenables. Aujourd'hui, je vous partage ma stack complète pour construire un système de replay fiable, reproductible et performant.
Introduction : Pourquoi le Replay de Données Tick est Critique
Le marché des cryptomonnaies présente une complexité particulière pour les traders algorithmiques. La nature 24/7 des exchanges, les problèmes de liquidité sur les altcoins, et les flash crashes récurrents exigent une validation rigoureuse des stratégies sur des données historiques réalistes. Un simple backtest sur des bougies OHLCV ne suffit plus lorsque votre stratégie dépend de l' Book profondeur ou de la microstructure du marché.
Le système que je vais décrire permet de rejouer des données tick par tick avec une latence simulée réaliste, d'injecter des événements de liquidité, et d'exécuter des ordres virtuels pour valider votre stratégie dans des conditions de marché variées.
Architecture du Système de Replay
L'architecture se compose de quatre composants principaux qui communiquent via une file de messages asynchrone. Le schéma ci-dessous illustre le flux de données entre chaque模块 :
+-------------------+ +-------------------+ +-------------------+
| Tardis API | --> | Data Fetcher | --> | Kafka Queue |
| (Raw Ticks) | | (Normalizer) | | (Buffer) |
+-------------------+ +-------------------+ +-------------------+
|
v
+-------------------+ +-------------------+ +-------------------+
| Backtester | <-- | Event Engine | <-- | Tick Replayer |
| (Strategy) | | (Simulator) | | (Playback) |
+-------------------+ +-------------------+ +-------------------+Configuration de la Source de Données Tardis
L'API Tardis Machine propose des données tick de qualité institutionnelle pour plus de 50 exchanges. Pour le replay crypto, nous utilisons principalement les endpoints de flux continu et de téléchargement historique. Le coût varie selon la granularité des données :
| Type de Données | Granularité | Prix Mensuel (USD) | Latence Moyenne |
|---|---|---|---|
| Trades Classiques | Tick par tick | $299 - $799 | ~100ms |
| Order Book Delta | Snapshot 100ms | $499 - $1,299 | ~150ms |
| Données Fixtures | Quotes complets | $899 - $2,499 | ~200ms |
| Accès Complet Tous Exchanges | Multi-sources | $2,999+ | Variable |
Code Complet du Système de Replay
1. Installation des Dépendances
# Installation des packages requis
pip install tardis-client asyncio aiohttp pandas numpy
pip install holy sheep-ai-client # Pour l'analyse de données IA optionnelle
pip install backtrader vectorbt pandas-ta
Configuration de l'environnement
export TARDIS_API_KEY="votre_cle_api_tardis"
export HOLYSHEEP_API_KEY="votre_cle_holysheep" # Optionnel: analyse IA
export REDIS_URL="redis://localhost:6379"2. Module de Téléchargement des Données Tick
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any
import pandas as pd
class TardisDataFetcher:
"""
Télécharge et normalise les données tick depuis l'API Tardis.
Auteur: HolySheep AI - Stack crypto professionnel
"""
BASE_URL = "https://api.tardis-dev.com/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = None
self.buffer = []
async def fetch_trades(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> pd.DataFrame:
"""
Télécharge les trades historiques pour un pair donné.
Args:
exchange: Nom de l'exchange (ex: 'binance', 'bybit')
symbol: Paire de trading (ex: 'BTC-USDT')
start_date: Date de début de la période
end_date: Date de fin de la période
Returns:
DataFrame avec colonnes: timestamp, price, size, side, trade_id
"""
url = f"{self.BASE_URL}/historical/trades"
params = {
"exchange": exchange,
"symbol": symbol,
"from": start_date.isoformat(),
"to": end_date.isoformat(),
"limit": 10000 # Maximum par requête
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
all_trades = []
async with aiohttp.ClientSession() as session:
while True:
async with session.get(
url,
params=params,
headers=headers
) as response:
if response.status == 429:
# Rate limiting - attente exponentielle
retry_after = int(response.headers.get('Retry-After', 60))
await asyncio.sleep(retry_after)
continue
if response.status != 200:
raise Exception(
f"Erreur API Tardis: {response.status} - "
f"{await response.text()}"
)
data = await response.json()
trades = data.get('trades', [])
if not trades:
break
all_trades.extend(trades)
# Pagination via cursor
if 'next_cursor' in data:
params['cursor'] = data['next_cursor']
else:
break
# Respect du rate limit (10 req/s pour plan standard)
await asyncio.sleep(0.1)
# Normalisation vers DataFrame standardisé
df = pd.DataFrame(all_trades)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp').reset_index(drop=True)
return df
async def fetch_orderbook_snapshots(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
frequency: str = "100ms"
) -> pd.DataFrame:
"""
Télécharge les snapshots du carnet d'ordres.
Args:
exchange: Exchange cible
symbol: Paire de trading
start_date: Début de la période
end_date: Fin de la période
frequency: Fréquence des snapshots ('100ms', '1s', '10s')
Returns:
DataFrame avec structure orderbook normalisée
"""
url = f"{self.BASE_URL}/historical/orderbooks"
params = {
"exchange": exchange,
"symbol": symbol,
"from": start_date.isoformat(),
"to": end_date.isoformat(),
"frequency": frequency,
"depth": 25 # 25 niveaux de chaque côté
}
headers = {"Authorization": f"Bearer {self.api_key}"}
snapshots = []
async with aiohttp.ClientSession() as session:
cursor = None
while True:
if cursor:
params['cursor'] = cursor
async with session.get(url, params=params, headers=headers) as resp:
if resp.status == 429:
await asyncio.sleep(60)
continue
data = await resp.json()
snapshots.extend(data.get('orderbooks', []))
cursor = data.get('next_cursor')
if not cursor:
break
await asyncio.sleep(0.1)
df = pd.DataFrame(snapshots)
df['timestamp'] = pd.to_datetime(df['timestamp'])
return df.sort_values('timestamp').reset_index(drop=True)
Exemple d'utilisation
async def main():
fetcher = TardisDataFetcher(api_key="your_tardis_api_key")
# Téléchargement des trades BTC/USDT sur 24 heures
trades = await fetcher.fetch_trades(
exchange="binance",
symbol="BTC-USDT",
start_date=datetime(2025, 12, 1),
end_date=datetime(2025, 12, 2)
)
print(f"Téléchargé {len(trades)} trades")
print(f"Plage horaire: {trades['timestamp'].min()} -> {trades['timestamp'].max()}")
print(f"Prix moyen: ${trades['price'].mean():,.2f}")
# Statistiques descriptives
print(f"\nVolume total: {trades['size'].sum():.4f} BTC")
print(f"Nombre de transactions: {len(trades):,}")
if __name__ == "__main__":
asyncio.run(main())3. Moteur de Replay avec Simulation de Latence
import asyncio
import heapq
from dataclasses import dataclass, field
from typing import Callable, List, Dict, Any, Optional
from datetime import datetime
from enum import Enum
import pandas as pd
import numpy as np
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
MARKET = "MARKET"
LIMIT = "LIMIT"
STOP = "STOP"
STOP_LIMIT = "STOP_LIMIT"
@dataclass(order=True)
class TickEvent:
timestamp: datetime = field(compare=True)
event_type: str = field(compare=False)
data: Dict[str, Any] = field(compare=False)
@dataclass
class Order:
order_id: str
symbol: str
side: OrderSide
order_type: OrderType
quantity: float
price: Optional[float] = None
stop_price: Optional[float] = None
filled_quantity: float = 0.0
avg_fill_price: float = 0.0
status: str = "PENDING"
created_at: datetime = field(default_factory=datetime.now)
class TickReplayer:
"""
Moteur de replay haute performance pour données tick crypto.
Simule la latence réseau et l'exécution d'ordres en temps accéléré.
Caractéristiques:
- Latence simulée configurable (10ms - 500ms)
- Support des ordres limit, market, stop
- Calcul du slippage réaliste
- Gestion des frais d'échange
"""
def __init__(
self,
latency_ms: int = 50,
maker_fee: float = 0.001,
taker_fee: float = 0.002,
slippage_model: str = "adaptive"
):
"""
Initialise le moteur de replay.
Args:
latency_ms: Latence simulée en millisecondes (défaut: HolySheep <50ms target)
maker_fee: Frais maker en fraction (0.1% = 0.001)
taker_fee: Frais taker en fraction (0.2% = 0.002)
slippage_model: Modèle de slippage ('fixed', 'adaptive', 'volume_based')
"""
self.latency_ms = latency_ms
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage_model = slippage_model
self.events: List[TickEvent] = []
self.pending_orders: Dict[str, Order] = {}
self.filled_orders: List[Order] = []
self.position: float = 0.0
self.cash: float = 100000.0 # Capital initial USDT
self.trades: List[Dict] = []
self.current_time: datetime = None
self.last_price: float = 0.0
self.last_trade_size: float = 0.0
self.spread_bps: float = 2.0 # Spread en basis points
def load_trades(self, trades_df: pd.DataFrame) -> None:
"""
Charge les trades depuis un DataFrame normalisé.
"""
for _, row in trades_df.iterrows():
event = TickEvent(
timestamp=row['timestamp'],
event_type="TRADE",
data={
"price": float(row['price']),
"size": float(row['size']),
"side": row.get('side', 'buy' if row['size'] > 0 else 'sell'),
"trade_id": row.get('trade_id', str(_))
}
)
self.events.append(event)
# Tri par timestamp pour garantir l'ordre
self.events.sort(key=lambda x: x.timestamp)
print(f"✓ {len(self.events)} événements chargés dans le replay buffer")
def load_orderbook(self, orderbook_df: pd.DataFrame) -> None:
"""
Intègre les snapshots du carnet d'ordres pour le calcul du slippage.
"""
for _, row in orderbook_df.iterrows():
event = TickEvent(
timestamp=row['timestamp'],
event_type="ORDERBOOK_UPDATE",
data={
"bids": row.get('bids', []),
"asks": row.get('asks', []),
"best_bid": float(row.get('bids', [[0]])[0][0]) if row.get('bids') else 0,
"best_ask": float(row.get('asks', [[0]])[0][0]) if row.get('asks') else 0
}
)
self.events.append(event)
self.events.sort(key=lambda x: x.timestamp)
def calculate_slippage(
self,
order: Order,
current_bid: float,
current_ask: float
) -> float:
"""
Calcule le slippage basé sur le modèle configuré.
Returns:
Slippage en fraction du prix
"""
if self.slippage_model == "fixed":
# Slippage fixe de 0.5 bps
return 0.0005 if order.side == OrderSide.BUY else -0.0005
elif self.slippage_model == "adaptive":
# Slippage adaptatif basé sur la taille relative
best_price = current_ask if order.side == OrderSide.BUY else current_bid
size_ratio = order.quantity / max(self.last_trade_size, 0.001)
# Slippage croissant avec la taille
base_slippage = self.spread_bps / 10000
size_impact = min(size_ratio * 0.001, 0.005) # Max 50 bps
slippage = base_slippage + size_impact
return slippage if order.side == OrderSide.BUY else -slippage
else: # volume_based
# Slippage basé sur le volume recent
volume_factor = min(self.last_trade_size / 1.0, 1.0) # Normalisé
return (self.spread_bps / 10000) * (2 - volume_factor) * (
1 if order.side == OrderSide.BUY else -1
)
def submit_order(
self,
order_id: str,
symbol: str,
side: OrderSide,
order_type: OrderType,
quantity: float,
price: Optional[float] = None,
stop_price: Optional[float] = None
) -> Order:
"""
Soumet un ordre au moteur de replay.
L'ordre est mis en file d'attente pour exécution au prochain tick.
"""
order = Order(
order_id=order_id,
symbol=symbol,
side=side,
order_type=order_type,
quantity=quantity,
price=price,
stop_price=stop_price
)
self.pending_orders[order_id] = order
return order
def _execute_order(self, order: Order, execution_price: float) -> None:
"""
Exécute un ordre avec calcul des frais et slippage.
"""
slippage = self.calculate_slippage(
order,
execution_price * (1 - self.spread_bps/10000),
execution_price * (1 + self.spread_bps/10000)
)
final_price = execution_price * (1 + slippage)
if order.side == OrderSide.BUY:
cost = order.quantity * final_price * (1 + self.taker_fee)
if cost > self.cash:
order.status = "REJECTED"
return
self.cash -= cost
self.position += order.quantity
else:
proceeds = order.quantity * final_price * (1 - self.taker_fee)
if order.quantity > self.position:
order.status = "REJECTED"
return
self.cash += proceeds
self.position -= order.quantity
order.status = "FILLED"
order.filled_quantity = order.quantity
order.avg_fill_price = final_price
self.trades.append({
"timestamp": self.current_time,
"order_id": order.order_id,
"side": order.side.value,
"quantity": order.quantity,
"price": final_price,
"slippage_bps": slippage * 10000,
"fees": order.quantity * final_price * self.taker_fee
})
async def run(
self,
strategy_callback: Callable,
speed_multiplier: float = 1000.0
) -> Dict[str, Any]:
"""
Exécute le replay des événements.
Args:
strategy_callback: Fonction de stratégie appelée à chaque tick
speed_multiplier: Multiplicateur de vitesse (1000 = 1 jour en ~87 secondes)
Returns:
Résultats du backtest avec métriques complètes
"""
print(f"\n🚀 Lancement du replay: {len(self.events)} événements")
print(f"⚡ Vitesse simulée: {speed_multiplier}x")
print(f"⏱️ Latence configurée: {self.latency_ms}ms")
total_events = len(self.events)
for i, event in enumerate(self.events):
self.current_time = event.timestamp
# Mise à jour des prix
if event.event_type == "TRADE":
self.last_price = event.data['price']
self.last_trade_size = event.data['size']
# Exécution des ordres market en attente
for order_id, order in list(self.pending_orders.items()):
if order.order_type == OrderType.MARKET:
# Simulation de latence réseau
await asyncio.sleep(self.latency_ms / 1000)
self._execute_order(order, self.last_price)
del self.pending_orders[order_id]
elif event.event_type == "ORDERBOOK_UPDATE":
# Exécution des ordres limit
for order_id, order in list(self.pending_orders.items()):
if order.order_type == OrderType.LIMIT:
if order.side == OrderSide.BUY:
if event.data['best_ask'] <= order.price:
await asyncio.sleep(self.latency_ms / 1000)
self._execute_order(order, event.data['best_ask'])
del self.pending_orders[order_id]
else:
if event.data['best_bid'] >= order.price:
await asyncio.sleep(self.latency_ms / 1000)
self._execute_order(order, event.data['best_bid'])
del self.pending_orders[order_id]
# Appel de la stratégie
await strategy_callback(self, event)
# Logging toutes les 10000 itérations
if (i + 1) % 10000 == 0:
progress = (i + 1) / total_events * 100
print(f" Progression: {progress:.1f}% ({i+1}/{total_events})")
# Calcul des métriques finales
return self._calculate_metrics()
def _calculate_metrics(self) -> Dict[str, Any]:
"""
Calcule les métriques de performance du backtest.
"""
if not self.trades:
return {"error": "Aucun trade exécuté"}
df_trades = pd.DataFrame(self.trades)
df_trades['pnl'] = df_trades.apply(
lambda x: x['quantity'] * x['price'] if x['side'] == 'SELL'
else -x['quantity'] * x['price'],
axis=1
)
# Calcul du P&L non réalisé
final_pnl = self.cash - 100000 + self.position * self.last_price
return {
"initial_capital": 100000,
"final_capital": self.cash,
"final_position": self.position,
"current_price": self.last_price,
"unrealized_pnl": self.position * self.last_price,
"total_realized_pnl": self.cash - 100000,
"net_pnl": final_pnl,
"total_trades": len(self.trades),
"winning_trades": len(df_trades[df_trades['pnl'] > 0]),
"losing_trades": len(df_trades[df_trades['pnl'] <= 0]),
"total_fees": df_trades['fees'].sum(),
"avg_slippage_bps": df_trades['slippage_bps'].mean(),
"max_slippage_bps": df_trades['slippage_bps'].max(),
"return_pct": (final_pnl / 100000) * 100,
"trades": df_trades.to_dict('records')
}4. Stratégie de Trading avec Signaux IA
"""
Stratégie de mean reversion sur données tick avec signals IA HolySheep.
Version optimisée pour le backtesting haute fréquence.
"""
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
Configuration HolySheep AI pour l'analyse
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé
class HolySheepAnalyzer:
"""
Client pour l'analyse de marché via l'API HolySheep AI.
Tarification 2026 (USD/Million tokens):
- GPT-4.1: $8.00 (premium)
- Claude Sonnet 4.5: $15.00 (haute qualité)
- Gemini 2.5 Flash: $2.50 (rapide, économique)
- DeepSeek V3.2: $0.42 (le plus économique)
Latence moyenne: <50ms avec optimisation Chinese bypass
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def analyze_market_regime(
self,
recent_prices: List[float],
volatility: float,
volume_profile: Dict
) -> Dict[str, Any]:
"""
Analyse le régime de marché actuel pour adapter la stratégie.
Utilise DeepSeek V3.2 pour l'analyse économique ($0.42/M token)
avec fallback vers Gemini 2.5 Flash ($2.50/M token) si nécessaire.
"""
prompt = f"""
Analyse le régime de marché actuel basé sur:
- Prix récents: {recent_prices[-10:] if len(recent_prices) > 10 else recent_prices}
- Volatilité (écart-type 5min): {volatility:.4f}
- Volume profile: {volume_profile}
Retourne un JSON avec:
- regime: "trending", "ranging", "volatile", "quiet"
- confidence: 0.0-1.0
- recommended_strategy: "mean_reversion", "momentum", "breakout", "neutral"
- risk_level: "low", "medium", "high"
"""
try:
async with aiohttp.ClientSession() as session:
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "Tu es un analyste de marché crypto expert."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
) as response:
if response.status == 200:
data = await response.json()
content = data['choices'][0]['message']['content']
return json.loads(content)
else:
# Fallback vers stratégie par défaut
return {
"regime": "unknown",
"confidence": 0.0,
"recommended_strategy": "neutral",
"risk_level": "medium"
}
except Exception as e:
print(f"Erreur analyse HolySheep: {e}")
return {
"regime": "unknown",
"confidence": 0.0,
"recommended_strategy": "neutral",
"risk_level": "medium"
}
class MeanReversionStrategy:
"""
Stratégie de mean reversion optimisée pour le trading haute fréquence.
Paramètres:
- lookback_period: Nombre de ticks pour calculer la moyenne mobile
- std_multiplier: Nombre d'écarts-types pour les seuils d'entrée
- position_size: Taille maximale de position en USD
- take_profit_pct: Seuil de take profit en pourcentage
- stop_loss_pct: Seuil de stop loss en pourcentage
"""
def __init__(
self,
lookback_period: int = 100,
std_multiplier: float = 2.0,
position_size_usd: float = 1000.0,
take_profit_pct: float = 0.5,
stop_loss_pct: float = 0.3
):
self.lookback_period = lookback_period
self.std_multiplier = std_multiplier
self.position_size_usd = position_size_usd
self.take_profit_pct = take_profit_pct
self.stop_loss_pct = stop_loss_pct
self.prices: List[float] = []
self.rolling_mean: float = 0.0
self.rolling_std: float = 0.0
self.position: float = 0.0
self.entry_price: float = 0.0
self.order_counter: int = 0
# Analyseur HolySheep optionnel
self.analyzer = None # Initialisé si API key disponible
def set_analyzer(self, analyzer: HolySheepAnalyzer):
"""Active l'analyse IA via HolySheep."""
self.analyzer = analyzer
async def on_tick(self, replayer: TickReplayer, event):
"""
Callback appelé à chaque tick du replay.
"""
if event.event_type != "TRADE":
return
price = event.data['price']
self.prices.append(price)
# Élimination des valeurs aberrantes (flash crash protection)
if len(self.prices) > self.lookback_period:
self.prices = self.prices[-self.lookback_period:]
# Calcul des statistiques roulantes
if len(self.prices) >= self.lookback_period:
self.rolling_mean = np.mean(self.prices[-self.lookback_period:])
self.rolling_std = np.std(self.prices[-self.lookback_period:]) + 1e-8
# Calcul du z-score
z_score = (price - self.rolling_mean) / self.rolling_std
# Gestion de position existante
if self.position != 0:
pnl_pct = (price - self.entry_price) / self.entry_price * 100
if self.position > 0: # Long
if pnl_pct >= self.take_profit_pct or pnl_pct <= -self.stop_loss_pct:
await self._close_position(replayer, price)
else: # Short
if pnl_pct <= -self.take_profit_pct or pnl_pct >= self.stop_loss_pct:
await self._close_position(replayer, price)
# Signaux d'entrée
if self.position == 0:
# Analyse IA optionnelle pour filtrer les signaux
if self.analyzer and len(self.prices) % 1000 == 0:
regime = await self.analyzer.analyze_market_regime(
self.prices,
self.rolling_std / self.rolling_mean,
{"recent_volume": sum(self.prices[-100:])}
)
# Adaptation des seuils selon le régime
if regime['risk_level'] == 'high':
continue # Skip les signaux en période risquée
# Signal d'achat: prix en dessous de la moyenne - 2σ
if z_score < -self.std_multiplier:
size = self.position_size_usd / price
await self._open_long(replayer, price, size)
# Signal de vente: prix au-dessus de la moyenne + 2σ
elif z_score > self.std_multiplier:
size = self.position_size_usd / price
await self._open_short(replayer, price, size)
async def _open_long(self, replayer: TickReplayer, price: float, size: float):
"""Ouvre une position longue."""
order_id = f"long_{self.order_counter}"
self.order_counter += 1
replayer.submit_order(
order_id=order_id,
symbol="BTC-USDT",
side=OrderSide.BUY,
order_type=OrderType.MARKET,
quantity=size
)
self.position = size
self.entry_price = price
async def _open_short(self, replayer: TickReplayer, price: float, size: float):
"""Ouvre une position courte."""
order_id = f"short_{self.order_counter}"
self.order_counter += 1
replayer.submit_order(
order_id=order_id,
symbol="BTC-USDT",
side=OrderSide.SELL,
order_type=OrderType.MARKET,
quantity=size
)
self.position = -size
self.entry_price = price
async def _close_position(self, replayer: TickReplayer, price: float):
"""Ferme la position actuelle."""
side = OrderSide.SELL if self.position > 0 else OrderSide.BUY
order_id = f"close_{self.order_counter}"
self.order_counter += 1
replayer.submit_order(
order_id=order_id,
symbol="BTC-USDT",
side=side,
order_type=OrderType.MARKET,
quantity=abs(self.position)
)
self.position = 0
self.entry_price = 0.0
Script principal de backtest
async def run_backtest():
"""
Exécute le backtest complet sur données Tardis.
"""
# Initialisation
fetcher = TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY")
replayer = TickReplayer(
latency_ms=50, # Latence HolySheep standard <50ms
maker_fee=0.001,
taker_fee=0.002,
slippage_model="adaptive"
)
# Téléchargement des données (exemple: 1 semaine BTC/USDT)
print("📥 Téléchargement des données depuis Tardis...")
trades = await fetcher.fetch_trades(
exchange="binance",
symbol="BTC-USDT",
start_date=datetime(2025, 11, 1),
end_date=datetime(2025, 11, 8)
)
# Chargement dans le replayer
replayer.load_trades(trades)
# Initialisation de la stratégie
strategy = MeanReversionStrategy(
lookback_period=200,
std_multiplier=2.0,
position_size_usd=500.0, # Position de $500 par trade
take_profit_pct=0.5,
stop_loss_pct=0.3
)
# Option: Active l'analyse IA HolySheep
# holy_sheep = HolySheepAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# strategy.set_analyzer(holy_sheep)
# Exécution du replay
print("\n🎮 Lancement du replay...")
results = await replayer.run(
strategy_callback=strategy.on_tick,
speed_multiplier=10000 # 10,000x: 1 semaine en ~6 secondes
)
# Affichage des résultats
print("\n" + "="*60)
print("📊 RÉSULTATS DU BACKTEST")
print("="*60)
print(f"Capital initial: ${results['initial_capital']:,.2f}")
print(f"Capital final: ${results['final_capital']:,.2f}")
print(f"P&L total: ${results['net_pnl']:,.2f}")
print(f"Rendement: {results['return_pct']:.2f}%")
print(f"Nombre de trades: {results['total_trades']}")
print(f"Trades gagnants: {results['winning_trades']}")
print(f"Trades perdants: {results['losing_trades']}")
print(f"Frais totaux: ${results['total_fees']:.2f}")
print(f"Slippage moyen: {results['avg_slippage_bps']:.2f} bps")
print(f"Slippage max: {results['max_slippage_bps']:.2f} bps")
return results
if __name__ == "__main__":
results = asyncio.run(run_backtest())Comparatif des Sources de Données Tick Crypto
| Provider | Exchanges Supportés | Granularité | Ressources connexesArticles connexes
🔥 Essayez HolySheep AIPasserelle API IA directe. Claude, GPT-5, Gemini, DeepSeek — une clé, sans VPN. |
|---|