En tant qu'ingénieur spécialisé dans le trading algorithmique depuis plus de cinq ans, j'ai passé d'innombrables heures à maîtriser les flux de données des carnets d'ordres. Le Order Book L2 de Bybit représente l'un des défis les plus intéressants en matière de reconstruction de données temps réel. Dans ce tutoriel complet, je vais vous guider pas à pas à travers le processus de parsing des données incrémentales et leur reconstruction en un carnet d'ordres complet et fonctionnel.
Comprendre le Order Book L2 de Bybit
Le Order Book L2 (Level 2) de Bybit pour les contrats perpétuels expose deux types de messages complémentaires : les instantanés (snapshots) et les mises à jour incrémentales (deltas). La documentation officielle de Bybit détaille un protocole WebSocket utilisant le format orderbook.200.$symbol pour les abonnements.
La latence moyenne observée sur les flux WebSocket de Bybit se situe entre 5ms et 15ms selon la localisation du serveur. Cependant, le vrai défi réside dans la reconstruction efficace du carnet d'ordres côté client, qui nécessite une gestion rigoureuse des opérations de suppression, insertion et modification des niveaux de prix.
Architecture du protocole WebSocket Bybit
Le flux de données L2 de Bybit utilise un format binaire compact pour optimiser la bande passante. Chaque message contient un en-tête de type, un timestamp Unix en millisecondes, et un corps variant selon qu'il s'agit d'un snapshot ou d'un delta. La structure implique que le client doit maintenir un état interne synchronisé avec le serveur.
Code Python pour la connexion WebSocket
import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from sortedcontainers import SortedDict
import websockets
@dataclass
class OrderBookLevel:
"""Représente un niveau de prix dans le carnet d'ordres."""
price: float
quantity: float
side: str # 'Buy' ou 'Sell'
class BybitL2OrderBook:
"""Classe de reconstruction du Order Book L2 pour Bybit."""
def __init__(self, symbol: str = "BTCUSDT"):
self.symbol = symbol
# SortedDict pour maintient automatique des prix ordonnés
self.bids: SortedDict = SortedDict() # Prix -> Quantité (achats)
self.asks: SortedDict = SortedDict() # Prix -> Quantité (ventes)
self.last_update_id: int = 0
self.sequence_number: int = 0
self.is_snapshot_received: bool = False
def apply_snapshot(self, data: dict) -> None:
"""Applique un instantané complet du carnet d'ordres."""
if "data" not in data:
return
snapshot_data = data["data"]
self.last_update_id = snapshot_data.get("u", 0)
# Vider les carnets existants
self.bids.clear()
self.asks.clear()
# Appliquer les niveaux d'achat
for level in snapshot_data.get("b", []):
price = float(level[0])
qty = float(level[1])
if qty > 0:
self.bids[price] = qty
# Appliquer les niveaux de vente
for level in snapshot_data.get("a", []):
price = float(level[0])
qty = float(level[1])
if qty > 0:
self.asks[price] = qty
self.is_snapshot_received = True
self.sequence_number = self.last_update_id
def apply_delta(self, data: dict) -> bool:
"""Applique une mise à jour incrémentale au carnet d'ordres."""
if "data" not in data:
return False
delta_data = data["data"]
update_id = delta_data.get("u", 0)
# Ignorer les mises à jour obsolètes
if update_id <= self.last_update_id:
return False
# Traiter les mises à jour d'achat
for update in delta_data.get("b", []):
price = float(update[0])
qty = float(update[1])
if qty == 0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
# Traiter les mises à jour de vente
for update in delta_data.get("a", []):
price = float(update[0])
qty = float(update[1])
if qty == 0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
self.last_update_id = update_id
self.sequence_number += 1
return True
def get_best_bid_ask(self) -> tuple:
"""Retourne le meilleur prix d'achat et de vente."""
best_bid = self.bids.peekitem(-1)[0] if self.bids else None
best_ask = self.asks.peekitem(0)[0] if self.asks else None
return (best_bid, best_ask)
def get_spread(self) -> Optional[float]:
"""Calcule le spread entre meilleur achat et vente."""
best_bid, best_ask = self.get_best_bid_ask()
if best_bid and best_ask:
return (best_ask - best_bid) / best_bid * 100
return None
def display_top_levels(self, depth: int = 5) -> None:
"""Affiche les N meilleurs niveaux de chaque côté."""
print(f"\n=== Order Book pour {self.symbol} ===")
print(f"Meilleur acheteur: {self.get_best_bid_ask()[0]}")
print(f"Meilleur vendeur: {self.get_best_bid_ask()[1]}")
print(f"Spread: {self.get_spread():.4f}%" if self.get_spread() else "Spread: N/A")
print(f"\n--- Top {depth} Achats (Bids) ---")
for i, (price, qty) in enumerate(reversed(list(self.bids.items())[-depth:])):
print(f" {i+1}. Prix: {price:,.2f} | Qté: {qty:.4f}")
print(f"\n--- Top {depth} Ventes (Asks) ---")
for i, (price, qty) in enumerate(list(self.asks.items())[:depth]):
print(f" {i+1}. Prix: {price:,.2f} | Qté: {qty:.4f}")
async def connect_bybit_orderbook():
"""Connexion au flux WebSocket L2 de Bybit."""
url = "wss://stream.bybit.com/v5/public/linear"
symbol = "BTCUSDT"
orderbook = BybitL2OrderBook(symbol)
subscribed = False
async with websockets.connect(url) as ws:
# S'abonner au flux order book
subscribe_msg = {
"op": "subscribe",
"args": [f"orderbook.200.{symbol}"]
}
await ws.send(json.dumps(subscribe