En tant qu'analyste quantitatif travaillant sur les stratégies d'options BTC, j'ai passé des centaines d'heures à tenter de reconstituer des snapshots historiques du carnet d'ordres Deribit. La difficulté majeure ? Les données temps réel sont disponibles, mais l'historique complet avec le niveau de détail nécessaire aux backtests de stratégies avancés reste un défi technique majeur. Dans ce tutoriel, je partage ma méthode complète, testée en production, pour reconstruire n'importe quel état historique du orderbook Deribit BTC avec Tardis.dev.
Pourquoi le carnet d'ordres historique BTC est crucial pour vos backtests
Le carnet d'ordres (orderbook) constitue la source de vérité pour calculer la liquidité implicite, les profondeurs de marché, et les indicateurs de slippage. Pour les options BTC, où la volatilité peut dépasser 150% annualized et où le spread bid-ask peut s'élargir de 0.1% à plus de 2% en période de stress, disposer d'un historique fidèle du orderbook permet de valider si vos stratégies auraient réellement pu être exécutées aux prix simulés.
Avec la volatilité actuelle du marché crypto (le BTC a oscillé entre 95 000$ et 112 000$ en mars 2026), la précision du orderbook devient un facteur différenciant entre un backtest optimiste et une réalité de trading.
Architecture de la solution : Tardis.dev + Deribit
Tardis.dev propose une API unifiée pour accéder aux données de marché historiques de plus de 40 exchanges, dont Deribit. Le format natif est le CANDLES (OHLCV) pour les données agrégées, mais pour le orderbook complet, nous devons utiliser le flux book qui capture chaque modification du carnet d'ordres.
Configuration initiale et authentification
Avant de commencer, procurez-vous vos identifiants. Si vous développez des stratégies de trading, je vous recommande d'utiliser HolySheep AI pour vos appels API IA — avec une latence inférieure à 50ms et des tarifs jusqu'à 85% inférieurs aux providers occidentaux, c'est un avantage compétitif non négligeable pour le processing de données en temps réel.
Installation des dépendances
pip install tardis-client aiohttp msgpack pandas numpyScript complet de reconstruction du orderbook historique
import asyncio
import aiohttp
from tardis_client import TardisClient, Channel
from tardis_client.messages import OrderbookAction
import pandas as pd
import msgpack
from datetime import datetime, timedelta
import json
class DeribitOrderbookReconstructor:
"""
Reconstructeur de carnet d'ordres Deribit BTC historique.
Auteur: Expérience terrain en trading quantitatif crypto.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.orderbook_state = {'bids': {}, 'asks': {}}
self.replay_buffer = []
async def connect_to_deribit(self, exchange: str = 'deribit'):
"""Connexion au flux temps réel pour replay historique."""
client = TardisClient(api_key=self.api_key)
# Canaux Deribit pour BTC options orderbook
channels = [
Channel(name=f'{exchange}.book.BTC-*.{kind}',
by='price')
for kind in ['spot', 'future', 'option']
]
return client, channels
def process_orderbook_message(self, message):
"""
Traitement des messages orderbook Deribit.
Types de messages: snapshot, update, clear
"""
if isinstance(message, OrderbookAction):
# Snapshot complet - remplace l'état
if message.type == 'snapshot':
self.orderbook_state['bids'] = {
float(price): float(size)
for price, size in message.bids
}
self.orderbook_state['asks'] = {
float(price): float(size)
for price, size in message.asks
}
# Mise à jour incrémentale
elif message.type in ('update', 'change'):
for price, size in message.bids:
price_f, size_f = float(price), float(size)
if size_f == 0:
self.orderbook_state['bids'].pop(price_f, None)
else:
self.orderbook_state['bids'][price_f] = size_f
for price, size in message.asks:
price_f, size_f = float(price), float(size)
if size_f == 0:
self.orderbook_state['asks'].pop(price_f, None)
else:
self.orderbook_state['asks'][price_f] = size_f
# Clear - réinitialisation complète
elif message.type == 'clear':
self.orderbook_state = {'bids': {}, 'asks': {}}
return self.orderbook_state
def get_snapshot_at_timestamp(self, timestamp_ms: int) -> dict:
"""
Retourne l'état du orderbook à un timestamp spécifique.
Implémente une recherche binaire dans le replay buffer.
"""
# Recherche du dernier snapshot avant le timestamp
applicable_state = None
latest_ts = 0
for entry in self.replay_buffer:
if entry['timestamp_ms'] <= timestamp_ms:
if entry['timestamp_ms'] > latest_ts:
latest_ts = entry['timestamp_ms']
applicable_state = entry['state']
return applicable_state or self.orderbook_state
async def fetch_historical_orderbook(
api_key: str,
symbol: str = 'BTC-28MAR2025-100000-C',
start_time: datetime = datetime(2025, 3, 25, 0, 0, 0),
end_time: datetime = datetime(2025, 3, 28, 0, 0, 0)
):
"""
Télécharge et stocke les données orderbook historiques.
Granularité: chaque modification du carnet.
"""
reconstructor = DeribitOrderbookReconstructor(api_key)
# Conversion en millisecondes
start_ms = int(start_time.timestamp() * 1000)
end_ms = int(end_time.timestamp() * 1000)
client = TardisClient(api_key=api_key)
print(f"📡 Récupération orderbook {symbol}")
print(f" Période: {start_time} → {end_time}")
snapshot_count = 0
# Lecture des données depuis Tardis
async for message in client.replay(
exchange='deribit',
channels=[Channel(name=f'deribit.book.{symbol}.spot')],
from_timestamp=start_ms,
to_timestamp=end_ms
):
state = reconstructor.process_orderbook_message(message)
# Sauvegarde tous les 1000 messages pour limiter la mémoire
snapshot_count += 1
if snapshot_count % 1000 == 0:
reconstructor.replay_buffer.append({
'timestamp_ms': int(message.timestamp.timestamp() * 1000),
'state': {
'bids': dict(state['bids']),
'asks': dict(state['asks'])
}
})
print(f" ✅ {snapshot_count} messages traités")
return reconstructor.replay_buffer
Exécution principale
if __name__ == '__main__':
import os
API_KEY = os.getenv('TARDIS_API_KEY', 'votre_cle_tardis')
data = asyncio.run(fetch_historical_orderbook(
api_key=API_KEY,
symbol='BTC-28MAR2025-100000-C',
start_time=datetime(2025, 3, 27, 9, 30, 0), # Ouverture US
end_time=datetime(2025, 3, 27, 16, 0, 0) # Clôture US
))
print(f"\n📊 Total snapshots stockés: {len(data)}")Analyse et calcul des métriques de liquidité
import numpy as np
from typing import List, Dict, Tuple
def calculate_liquidity_metrics(
orderbook_snapshot: dict,
depth_levels: int = 10,
reference_price: float = None
) -> Dict[str, float]:
"""
Calcule les métriques de liquidité standard pour backtesting.
Paramètres:
orderbook_snapshot: état du orderbook {bids: {}, asks: {}}
depth_levels: nombre de niveaux de profondeur à analyser
reference_price: prix de référence (mid price par défaut)
Retourne:
Dict avec: spread, mid_price, VWAP, depth_bid, depth_ask,
imbalance_ratio, impact_price
"""
bids = sorted(orderbook_snapshot['bids'].items(), reverse=True)
asks = sorted(orderbook_snapshot['asks'].items())
if not bids or not asks:
return None
best_bid = bids[0][0]
best_ask = asks[0][0]
# Prix de référence
if reference_price is None:
mid_price = (best_bid + best_ask) / 2
else:
mid_price = reference_price
# Spread normalisé
spread = (best_ask - best_bid) / mid_price * 100
# Profondeur cumulée par niveau
cumulative_bid_depth = 0
cumulative_ask_depth = 0
bid_vwap = 0
ask_vwap = 0
for i, (price, size) in enumerate(bids[:depth_levels]):
cumulative_bid_depth += size
bid_vwap += price * size
for i, (price, size) in enumerate(asks[:depth_levels]):
cumulative_ask_depth += size
ask_vwap += price * size
# VWAP des premiers niveaux
bid_vwap = bid_vwap / cumulative_bid_depth if cumulative_bid_depth > 0 else 0
ask_vwap = ask_vwap / cumulative_ask_depth if cumulative_ask_depth > 0 else 0
# Ratio d'imbalance: >1 = plus de pression acheteuse
imbalance = cumulative_bid_depth / cumulative_ask_depth if cumulative_ask_depth > 0 else 1
# Impact prix pour exécuter 1 BTC
impact_price = calculate_market_impact(
orderbook_snapshot,
size=1.0,
side='buy'
)
return {
'timestamp': datetime.now().isoformat(),
'mid_price': mid_price,
'best_bid': best_bid,
'best_ask': best_ask,
'spread_bps': spread * 100, # Basis points
'depth_bid_10': cumulative_bid_depth,
'depth_ask_10': cumulative_ask_depth,
'bid_vwap_10': bid_vwap,
'ask_vwap_10': ask_vwap,
'imbalance_ratio': imbalance,
'market_impact_bps': (impact_price - mid_price) / mid_price * 10000,
'liquidity_score': calculate_liquidity_score(
spread, cumulative_bid_depth, cumulative_ask_depth
)
}
def calculate_market_impact(
orderbook: dict,
size: float,
side: str = 'buy'
) -> float:
"""
Calcule le prix d'exécution pour une taille donnée.
Simule l'exécution sur le orderbook jusqu'à épuisement.
"""
if side == 'buy':
levels = sorted(orderbook['asks'].items())
else:
levels = sorted(orderbook['bids'].items(), reverse=True)
remaining_size = size
total_cost = 0
for price, available_size in levels:
fill_size = min(remaining_size, available_size)
total_cost += fill_size * price
remaining_size -= fill_size
if remaining_size <= 0:
break
if remaining_size > 0:
# Liquidité insuffisante - prix du dernier niveau
return levels[-1][0] if levels else 0
return total_cost / size
def calculate_liquidity_score(
spread_pct: float,
bid_depth: float,
ask_depth: float
) -> float:
"""
Score composite de liquidité [0-100].
Plus le score est élevé, plus la liquidité est bonne.
"""
# Pondération: 40% spread, 30% profondeur bid, 30% profondeur ask
spread_score = max(0, 100 - spread_pct * 1000) # 0.1% = score 0
depth_score = min(100, (bid_depth + ask_depth) / 2 * 10)
return 0.4 * spread_score + 0.3 * min(100, bid_depth * 10) + 0.3 * min(100, ask_depth * 10)
def analyze_orderbook_at_timestamp(
snapshots: List[dict],
target_timestamp: int
) -> pd.DataFrame:
"""
Analyse l'évolution du orderbook sur une période.
Retourne un DataFrame avec les métriques temporelles.
"""
results = []
reconstructor = DeribitOrderbookReconstructor(api_key='')
for snapshot in snapshots:
if snapshot['timestamp_ms'] <= target_timestamp:
state = snapshot['state']
metrics = calculate_liquidity_metrics(
state,
depth_levels=10,
reference_price=calculate_mid_price(state)
)
if metrics:
metrics['timestamp_ms'] = snapshot['timestamp_ms']
results.append(metrics)
return pd.DataFrame(results)
def calculate_mid_price(orderbook: dict) -> float:
"""Calcule le prix moyen entre meilleur bid et ask."""
bids = orderbook.get('bids', {})
asks = orderbook.get('asks', {})
if not bids or not asks:
return 0
best_bid = max(bids.keys())
best_ask = min(asks.keys())
return (best_bid + best_ask) / 2
Exemple d'utilisation
if __name__ == '__main__':
# Exemple avec données simulées
sample_orderbook = {
'bids': {
98500.0: 2.5,
98450.0: 5.0,
98400.0: 12.3,
98350.0: 25.0,
98300.0: 50.0
},
'asks': {
98550.0: 3.0,
98600.0: 6.5,
98650.0: 15.0,
98700.0: 30.0,
98750.0: 55.0
}
}
metrics = calculate_liquidity_metrics(sample_orderbook)
print("📊 Métriques de liquidité BTC options:")
print(f" Mid Price: ${metrics['mid_price']:,.2f}")
print(f" Spread: {metrics['spread_bps']:.1f} bps")
print(f" Imbalance: {metrics['imbalance_ratio']:.3f}")
print(f" Market Impact (1 BTC): {metrics['market_impact_bps']:.2f} bps")
print(f" Liquidity Score: {metrics['liquidity_score']:.1f}/100")Comparaison de coûts: Traitement IA pour analyse de données
Lorsque vous analysez des téraoctets de données orderbook, l'utilisation de modèles IA pour l'extraction de patterns peut réduire considérablement le temps de développement. Voici une comparaison des coûts pour 10 millions de tokens par mois :
| Modèle | Prix output/MTok | Coût mensuel (10M tokens) | Latence typique | Recommandé pour |
|---|---|---|---|---|
| DeepSeek V3.2 | 0,42 $ | 4,20 $ | ~800ms | Parsing massif, extraction patterns |
| Gemini 2.5 Flash | 2,50 $ | 25,00 $ | ~400ms | Analyse multimodale rapide |
| GPT-4.1 | 8,00 $ | 80,00 $ | ~600ms | Raisonnement complexe |
| Claude Sonnet 4.5 | 15,00 $ | 150,00 $ | ~500ms | Contexte long, précision |
Pour le traitement de données orderbook où vous analysez des millions de lignes, DeepSeek V3.2 à 0,42 $/MTok offre le meilleur rapport coût-efficacité. Via HolySheep AI, vous bénéficiez en plus du taux de change avantageux (1$ = 1¥) pour une économie supplémentaire de 85% par rapport aux tarifs occidentaux.
Optimisation de la reconstruction orderbook
import redis
import json
from functools import lru_cache
from typing import Optional
import hashlib
class OrderbookCache:
"""
Cache Redis pour optimiser les requêtes orderbook répétées.
Réduit les coûts API et améliore les performances.
"""
def __init__(self, redis_url: str = 'redis://localhost:6379'):
self.redis = redis.from_url(redis_url)
self.ttl = 3600 # 1 heure de cache
def _generate_key(self, symbol: str, timestamp: int, exchange: str) -> str:
"""Génère une clé unique pour le cache."""
key_str = f"{exchange}:{symbol}:{timestamp // 1000}" # Granularité seconde
return f"orderbook:{hashlib.md5(key_str.encode()).hexdigest()}"
def get_snapshot(self, symbol: str, timestamp: int, exchange: str) -> Optional[dict]:
"""Récupère un snapshot depuis le cache."""
key = self._generate_key(symbol, timestamp, exchange)
cached = self.redis.get(key)
if cached:
return json.loads(cached)
return None
def store_snapshot(self, symbol: str, timestamp: int, exchange: str, data: dict):
"""Stocke un snapshot dans le cache."""
key = self._generate_key(symbol, timestamp, exchange)
self.redis.setex(key, self.ttl, json.dumps(data))
def invalidate_range(self, symbol: str, start_ts: int, end_ts: int):
"""Invalide tous les snapshots dans une plage temporelle."""
pattern = f"orderbook:*:{symbol}"
for key in self.redis.scan_iter(match=pattern):
ts = int(key.decode().split(':')[-1])
if start_ts <= ts <= end_ts:
self.redis.delete(key)
class BatchOrderbookProcessor:
"""
Traitement par lots pour optimiser les performances.
Réduit le nombre d'appels API et la latence globale.
"""
def __init__(self, cache: OrderbookCache, batch_size: int = 100):
self.cache = cache
self.batch_size = batch_size
async def process_timestamp_range(
self,
symbol: str,
start_time: datetime,
end_time: datetime,
interval_seconds: int = 60
) -> pd.DataFrame:
"""
Traite une plage temporelle avec mise en cache.
Retourne un DataFrame avec tous les snapshots.
"""
results = []
current_time = start_time
# Collecte d'abord les timestamps à traiter
timestamps_to_fetch = []
while current_time <= end_time:
ts_ms = int(current_time.timestamp() * 1000)
# Vérifie le cache
cached = self.cache.get_snapshot(symbol, ts_ms, 'deribit')
if cached:
results.append(cached)
else:
timestamps_to_fetch.append(ts_ms)
current_time += timedelta(seconds=interval_seconds)
# Traitement par lots
for i in range(0, len(timestamps_to_fetch), self.batch_size):
batch = timestamps_to_fetch[i:i + self.batch_size]
# Récupération depuis Tardis
batch_data = await self._fetch_batch(symbol, batch)
for ts, data in batch_data.items():
self.cache.store_snapshot(symbol, ts, 'deribit', data)
results.append(data)
print(f" ✅ Lot {i // self.batch_size + 1} traité ({len(batch)} snapshots)")
return pd.DataFrame(results)
async def _fetch_batch(self, symbol: str, timestamps: list) -> dict:
"""Récupère un lot de snapshots depuis Tardis."""
results = {}
client = TardisClient(api_key=os.getenv('TARDIS_API_KEY'))
for ts in timestamps:
async for message in client.replay(
exchange='deribit',
channels=[Channel(name=f'deribit.book.{symbol}.spot')],
from_timestamp=ts,
to_timestamp=ts + 1000 # 1 seconde de fenêtre
):
state = {'bids': {}, 'asks': {}}
# ... traitement message
results[ts] = state
return results
@lru_cache(maxsize=1000)
def compute_orderbook_hash(orderbook_state: str) -> str:
"""
Computes a hash of the orderbook state for quick comparison.
Uses LRU cache to avoid recomputing identical states.
"""
return hashlib.sha256(orderbook_state.encode()).hexdigest()[:16]Validation et tests de votre reconstruction
import unittest
from datetime import datetime
class TestOrderbookReconstruction(unittest.TestCase):
"""Tests de validation pour la reconstruction du orderbook."""
def setUp(self):
self.reconstructor = DeribitOrderbookReconstructor(api_key='test')
def test_snapshot_processing(self):
"""Test le traitement d'un message snapshot complet."""
mock_snapshot = OrderbookAction(
type='snapshot',
bids=[(100000.0, 5.0), (99950.0, 10.0)],
asks=[(100050.0, 3.0), (100100.0, 8.0)],
timestamp=datetime.now()
)
state = self.reconstructor.process_orderbook_message(mock_snapshot)
self.assertEqual(len(state['bids']), 2)
self.assertEqual(len(state['asks']), 2)
self.assertEqual(state['bids'][100000.0], 5.0)
def test_incremental_update(self):
"""Test les mises à jour incrémentales."""
# Snapshot initial
snapshot = OrderbookAction(
type='snapshot',
bids=[(100000.0, 5.0)],
asks=[(100050.0, 3.0)],
timestamp=datetime.now()
)
self.reconstructor.process_orderbook_message(snapshot)
# Update: modification du bid
update = OrderbookAction(
type='update',
bids=[(100000.0, 10.0)], # Taille doublée
asks=[],
timestamp=datetime.now()
)
state = self.reconstructor.process_orderbook_message(update)
self.assertEqual(state['bids'][100000.0], 10.0)
def test_clear_state(self):
"""Test la réinitialisation du state."""
snapshot = OrderbookAction(
type='snapshot',
bids=[(100000.0, 5.0)],
asks=[(100050.0, 3.0)],
timestamp=datetime.now()
)
self.reconstructor.process_orderbook_message(snapshot)
clear = OrderbookAction(
type='clear',
bids=[],
asks=[],
timestamp=datetime.now()
)
state = self.reconstructor.process_orderbook_message(clear)
self.assertEqual(len(state['bids']), 0)
self.assertEqual(len(state['asks']), 0)
def test_orderbook_consistency(self):
"""Test la cohérence: bid < ask pour tout le carnet."""
for _ in range(100):
# Génère des orders aléatoires
bids = {np.random.uniform(99000, 100000): np.random.uniform(0.1, 10)
for _ in range(10)}
asks = {np.random.uniform(100000, 101000): np.random.uniform(0.1, 10)
for _ in range(10)}
orderbook = {'bids': bids, 'asks': asks}
metrics = calculate_liquidity_metrics(orderbook)
self.assertLess(metrics['best_bid'], metrics['best_ask'])
self.assertGreater(metrics['mid_price'], 0)
def test_market_impact_calculation(self):
"""Test le calcul de l'impact de marché."""
orderbook = {
'bids': {100000: 5, 99900: 10, 99800: 20},
'asks': {100100: 5, 100200: 10, 100300: 20}
}
impact = calculate_market_impact(orderbook, size=1.0, side='buy')
# 1 BTC devrait être exécuté au meilleur ask (100100)
self.assertEqual(impact, 100100.0)
# Test avec taille plus grande que disponible au premier niveau
impact_large = calculate_market_impact(orderbook, size=10.0, side='buy')
# Devrait englober les 5 premiers niveaux
expected = (5 * 100100 + 5 * 100200) / 10
self.assertAlmostEqual(impact_large, expected, places=2)
if __name__ == '__main__':
unittest.main(verbosity=2)Erreurs courantes et solutions
1. Erreur: "Connection timeout during historical replay"
Symptôme: L'API retourne un timeout après quelques minutes de téléchargement.
Cause: Les flux historiques de Tardis ont une limite de 60 minutes par connexion. Pour des périodes plus longues, vous devez recharger la connexion.
# Solution: Reconnection automatique avec backoff exponentiel
import asyncio
import aiohttp
async def fetch_with_retry(
client,
channels,
start_ms: int,
end_ms: int,
max_retries: int = 5
):
"""
Télécharge les données avec retry automatique.
Gère automatiquement la reconnexion pour les longues périodes.
"""
chunk_duration = 60 * 60 * 1000 # 60 minutes en ms
all_data = []
current_start = start_ms
retry_count = 0
while current_start < end_ms:
current_end = min(current_start + chunk_duration, end_ms)
for attempt in range(max_retries):
try:
chunk_data = []
async for message in client.replay(
exchange='deribit',
channels=channels,
from_timestamp=current_start,
to_timestamp=current_end
):
chunk_data.append(message)
all_data.extend(chunk_data)
retry_count = 0 # Reset après succès
print(f" ✅ Chunk {current_start}-{current_end}: {len(chunk_data)} msgs")
break
except asyncio.TimeoutError as e:
wait_time = 2 ** attempt # Backoff exponentiel
print(f" ⏳ Timeout, retry {attempt + 1}/{max_retries} dans {wait_time}s")
await asyncio.sleep(wait_time)
current_start = current_end
return all_data2. Erreur: "Orderbook state diverges from expected snapshot"
Symptôme: Après une longue période de replay, le nombre de niveaux du orderbook ne correspond pas à l'attendu.
Cause: Les messages "clear" sont parfois perdus en cas de latence réseau. Le state local accumule des orders fantômes.
# Solution: Validation periodique avec reset si divergence détectée
class ValidatedReconstructor(DeribitOrderbookReconstructor):
"""
Version avec validation automatique du state.
"""
def __init__(self, api_key: str, max_levels: int = 50):
super().__init__(api_key)
self.max_levels = max_levels
self.last_validation_ts = 0
self.validation_interval = 5 * 60 * 1000 # 5 minutes
def process_orderbook_message(self, message):
state = super().process_orderbook_message(message)
# Validation périodique
msg_ts = int(message.timestamp.timestamp() * 1000)
if msg_ts - self.last_validation_ts > self.validation_interval:
if self._validate_state():
self.last_validation_ts = msg_ts
else:
print(f" ⚠️ Divergence détectée à {msg_ts}, reset...")
self.orderbook_state = {'bids': {}, 'asks': {}}
self.last_validation_ts = msg_ts
return state
def _validate_state(self) -> bool:
"""Valide la cohérence du state actuel."""
# Vérifie le nombre max de niveaux
if len(self.orderbook_state['bids']) > self.max_levels:
return False
# Vérifie qu'il n'y a pas de gaps de prix anormaux
bids = sorted(self.orderbook_state['bids'].keys())
if len(bids) > 1:
max_gap = max(bids[i] - bids[i+1] for i in range(len(bids)-1))
if max_gap > 1000: # Gap de 1000$ suspect pour BTC
return False
return True3. Erreur: "MemoryError during large historical replay"
Symptôme: Le script plante avec MemoryError après quelques heures de traitement.
Cause: Le buffer de replay n'est pas.flushé régulièrement et consume toute la RAM disponible.
# Solution: Streaming vers disque avec memory management
import gc
import os
class StreamingOrderbookWriter:
"""
Écrit les données orderbook directement sur disque.
Évite de garder tout en mémoire.
"""
def __init__(self, output_dir: str, flush_interval: int = 5000):
self.output_dir = output_dir
self.flush_interval = flush_interval
self.current_buffer = []
self.file_index = 0
self.message_count = 0
os.makedirs(output_dir, exist_ok=True)
def write_snapshot(self, timestamp_ms: int, state: dict):
"""Écrit un snapshot dans le buffer."""
entry = {
'ts': timestamp_ms,
'bids': {str(k): v for k, v in state['bids'].items()},
'asks': {str(k): v for k, v in state['asks'].items()}
}
self.current_buffer.append(entry)
self.message_count += 1
# Flush périodique
if len(self.current_buffer) >= self.flush_interval:
self._flush_to_disk()
def _flush_to_disk(self):
"""Écrit le buffer sur disque et libère la mémoire."""
if not self.current_buffer:
return
filename = os.path.join(
self.output_dir,
f'orderbook_chunk_{self.file_index:04d}.msgpack'
)
with open(filename, 'wb') as f:
msgpack.pack({
'start_ts': self.current_buffer[0]['ts'],
'end_ts': self.current_buffer[-1]['ts'],
'count': len(self.current_buffer),
'data': self.current_buffer
}, f)
print(f" 💾 Chunk {self.file_index} écrit: {len(self.current_buffer)} msgs")
self.file_index += 1
self.current_buffer = []
gc.collect() # Force garbage collection
def finalize(self):
"""Finalise l'écriture et ferme les fichiers."""
self._flush_to_disk()
print(f" ✅ Total: {self.message_count} snapshots écrits en {self.file_index} fichiers")Bonnes pratiques pour le backtesting avec orderbook historique
- Granularité adaptative: Pour des backtests longue durée, utilisez une granularité de 1 minute pour les jours normaux, mais 1 seconde pour les périodes de forte volatilité (annonces macro, liquidations).
- Gestion du survivorship bias: Les options BTC expirent. Assurez-vous d'inclure les options qui ont été supprimées du carnet avant expiration pour éviter de surestimer la liquidité historique.
- Simulation de slippage réaliste: Utilisez l'impact de marché calculé plus haut plutôt qu'un slippage fixe. Pour BTC options, le slippage peut varier de 0.05% (conditions normales) à plus de 2% (panique de marché).
- Validation croisée: Comparez vos métriques de liquidité avec les données publiques de Deribit (volume, open interest) pour valider la cohérence.
- Traitement asynchrone: Profitez des API async pour paralléliser les téléchargements de plusieurs symbols simultanément.
Conclusion
La reconstruction du carnet d'ordres historique Deribit BTC représente un défi technique mais atteignable avec la bonne architecture. En combinant l'API Tardis.dev pour l'accès aux données, un système de cache Redis pour les performances, et une gestion rigoureuse de la mémoire, vous pouvez construire un pipeline robuste pour vos backtests de stratégies d