Introduction
Dans l'univers du trading algorithmique et du développement de robots de trading, la capacité à recevoir et traiter les mises à jour du carnet d'ordres en temps réel constitue un avantage compétitif majeur. L'API Tardis propose une solution WebSocket robuste pour consommner ces données de marché avec une latence minimale. Dans ce tutoriel exhaustif, nous allons explorer en profondeur la mécanique de parsing des messages WebSocket, depuis la connexion initiale jusqu'au traitement efficace des données de niveau 2 du marché.
Ce guide s'adresse aux développeurs Python et JavaScript/TypeScript souhaitant intégrer des flux de données de carnet d'ordres dans leurs applications de trading. Nous aborderons les aspects techniques du protocole WebSocket, les stratégies d'optimisation du parsing, et les pièges courants à éviter pour garantir une consommation fiable et performante des données de marché.
Comprendre l'architecture WebSocket de Tardis
Protocole de connexion et authentification
Le service WebSocket de Tardis utilise une connexion persistante basée sur le protocole WSS (WebSocket Secure) pour garantir la confidentialité et l'intégrité des données transitant entre le serveur et le client. La phase d'authentification s'effectue via un jeton Bearer dans l'en-tête de la requête de connexion initiale, permettant un contrôle d'accès granulaire aux différents flux de données de marché.
La gestion de la reconnexion automatique constitue un aspect critique de l'architecture cliente. En cas de déconnexion réseau ou de timeout serveur, le client doit implémenter un mécanisme de backoff exponentiel avec jitter pour éviter de surcharger le serveur lors de pics de connectivité instable. Les recommandations officielles suggèrent un délai initial de 1 seconde, avec un maximum de 30 secondes entre chaque tentative de reconnexion.
Structure des messages et types de données
Les messages WebSocket de Tardis sont encodés au format JSON, structurés selon une hiérarchie précise reflétant les différents types d'événements de marché. Chaque message contient un champ type permettant d'identifier la nature de l'événement, accompagné de métadonnées temporelles et contextuelles essentielles au traitement correct des données.
Les types d'événements principaux incluent les mises à jour du carnet d'ordres (orderbook_update), les transactions exécutées (trade), et les modifications d'état du carnet (orderbook_snapshot). Comprendre cette taxonomie s'avère fundamental pour implémenter un parser capable de router correctement chaque message vers le bon handler de traitement.
Implémentation du Parser en Python
Configuration initiale et dépendances
Pour реализовать un client WebSocket robuste avec Python, nous recommandons l'utilisation de la bibliothèque websockets version 13.0 ou supérieure, offrant un support natif des coroutines asyncio et une gestion optimisée du heartbeat pour maintenir la connexion活跃.
# Installation des dépendances requises
pip install websockets==13.0 aiofiles==23.2.1
Configuration du client WebSocket avec gestion de reconnexion
import asyncio
import json
import logging
from typing import Callable, Dict, Any, Optional
from websockets import WebSocketClientProtocol, connect
from dataclasses import dataclass, field
from collections import defaultdict
import time
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
@dataclass
class OrderBookLevel:
"""Représente un niveau de prix dans le carnet d'ordres."""
price: float
quantity: float
side: str # 'bid' ou 'ask'
timestamp: int
@dataclass
class OrderBookState:
"""État local du carnet d'ordres avec mise à jour incrémentale."""
bids: Dict[float, float] = field(default_factory=dict)
asks: Dict[float, float] = field(default_factory=dict)
last_update_id: int = 0
sequence_number: int = 0
class TardisWebSocketParser:
"""
Parseur haute performance pour les messages WebSocket de l'API Tardis.
Gère la reconnexion automatique et le maintien de l'état du carnet d'ordres.
"""
def __init__(
self,
api_key: str,
exchanges: list[str],
symbols: list[str],
channels: list[str] = None
):
self.api_key = api_key
self.exchanges = exchanges
self.symbols = [s.upper() for s in symbols]
self.channels = channels or ["orderbook-raw"]
self.orderbooks: Dict[str, OrderBookState] = {}
self._running = False
self._reconnect_delay = 1.0
self._max_reconnect_delay = 30.0
def _build_subscribe_message(self) -> dict:
"""Construit le message de subscription pour Tardis."""
return {
"type": "subscribe",
"exchanges": self.exchanges,
"channels": self.channels,
"symbols": self.symbols
}
def _parse_orderbook_snapshot(self, data: dict) -> None:
"""Parse un snapshot complet du carnet d'ordres."""
symbol = data.get("symbol", "")
if symbol not in self.orderbooks:
self.orderbooks[symbol] = OrderBookState()
ob = self.orderbooks[symbol]
ob.bids = {float(p): float(q) for p, q in data.get("bids", [])}
ob.asks = {float(p): float(q) for p, q in data.get("asks", [])}
ob.last_update_id = data.get("lastUpdateId", 0)
ob.sequence_number = data.get("sequenceId", 0)
logger.info(f"Snapshot reçu pour {symbol}: {len(ob.bids)} bids, {len(ob.asks)} asks")
def _parse_orderbook_update(self, data: dict, symbol: str) -> None:
"""
Parse une mise à jour incrémentale du carnet d'ordres.
Applique les modifications au состоя local de manière atomique.
"""
if symbol not in self.orderbooks:
logger.warning(f"Mise à jour reçue sans snapshot: {symbol}")
return
ob = self.orderbooks[symbol]
update_seq = data.get("sequenceId", 0)
# Vérification de la séquence pour détecter les pertes de messages
if update_seq <= ob.sequence_number:
logger.debug(f"Mise à jour hors séquence: {update_seq} <= {ob.sequence_number}")
return
# Traitement des modifications de prix
for side, levels in [("bid", "bids"), ("ask", "asks")]:
for price_str, quantity_str in data.get(levels, []):
price = float(price_str)
quantity = float(quantity_str)
book = ob.bids if side == "bid" else ob.asks
if quantity == 0:
book.pop(price, None)
else:
book[price] = quantity
ob.last_update_id = data.get("lastUpdateId", ob.last_update_id)
ob.sequence_number = update_seq
def _parse_trade(self, data: dict) -> OrderBookLevel:
"""Parse un événement de transaction."""
return OrderBookLevel(
price=float(data["price"]),
quantity=float(data["quantity"]),
side="buy" if data.get("side", "").lower() == "buy" else "sell",
timestamp=data.get("timestamp", 0)
)
async def _handle_message(self, message: str, handlers: Dict[str, Callable]) -> None:
"""
Route le message vers le bon parser selon le type d'événement.
Inclut la gestion des erreurs de parsing pour éviter les crashs.
"""
try:
data = json.loads(message)
msg_type = data.get("type", "")
if msg_type == "orderbook-snapshot":
self._parse_orderbook_snapshot(data)
if "snapshot" in handlers:
await handlers["snapshot"](data)
elif msg_type == "orderbook-update":
symbol = data.get("symbol", "")
self._parse_orderbook_update(data, symbol)
if "update" in handlers:
await handlers["update"](data, self.orderbooks.get(symbol))
elif msg_type == "trade":
trade = self._parse_trade(data)
if "trade" in handlers:
await handlers["trade"](trade)
elif msg_type == "subscribed":
logger.info(f"Subscription confirmée: {data}")
elif msg_type == "error":
logger.error(f"Erreur serveur: {data.get('message', 'Unknown')}")
except json.JSONDecodeError as e:
logger.error(f"JSON invalide: {e}, message: {message[:200]}")
except KeyError as e:
logger.error(f"Champ manquant: {e}")
except Exception as e:
logger.exception(f"Erreur de parsing: {e}")
async def connect(self, url: str, handlers: Dict[str, Callable]) -> None:
"""
Établit la connexion WebSocket et lance le consumer de messages.
Gère automatiquement la reconnexion en cas de déconnexion.
"""
self._running = True
reconnect_attempts = 0
while self._running:
try:
headers = {"Authorization": f"Bearer {self.api_key}"}
async with connect(url, extra_headers=headers) as ws:
logger.info("Connexion WebSocket établie")
reconnect_attempts = 0
self._reconnect_delay = 1.0
# Envoi du message de subscription
subscribe_msg = self._build_subscribe_message()
await ws.send(json.dumps(subscribe_msg))
logger.info(f"Subscription envoyée: {subscribe_msg}")
# Boucle de consommation des messages
async for raw_message in ws:
await self._handle_message(raw_message, handlers)
except WebSocketException as e:
logger.error(f"Erreur WebSocket: {e}")
except Exception as e:
logger.exception(f"Connexion interrompue: {e}")
finally:
if self._running:
reconnect_attempts += 1
delay = min(
self._reconnect_delay * (2 ** reconnect_attempts) + random.uniform(0, 1),
self._max_reconnect_delay
)
logger.info(f"Reconnexion dans {delay:.1f}s (tentative {reconnect_attempts})")
await asyncio.sleep(delay)
def stop(self) -> None:
"""Arrête le consumer de manière élégante."""
self._running = False
logger.info("Arrêt du consumer WebSocket")
Exemple d'utilisation avec gestion du carnet d'ordres
import asyncio
import random
async def main():
"""Exemple complet d'utilisation du parseur Tardis."""
# Configuration des handlers pour chaque type d'événement
async def on_snapshot(data: dict):
symbol = data.get("symbol", "")
bids_count = len(data.get("bids", []))
asks_count = len(data.get("asks", []))
print(f"📊 Snapshot {symbol}: {bids_count} bids / {asks_count} asks")
async def on_update(data: dict, orderbook):
if orderbook is None:
return
symbol = data.get("symbol", "")
best_bid = max(orderbook.bids.keys()) if orderbook.bids else None
best_ask = min(orderbook.asks.keys()) if orderbook.asks else None
spread = best_ask - best_bid if best_bid and best_ask else 0
print(f"📈 {symbol} | Bid: {best_bid} | Ask: {best_ask} | Spread: {spread:.2f}")
async def on_trade(trade):
direction = "🟢 ACHAT" if trade.side == "buy" else "🔴 VENTE"
print(f"{direction} | Prix: {trade.price} | Qté: {trade.quantity} | {trade.timestamp}")
handlers = {
"snapshot": on_snapshot,
"update": on_update,
"trade": on_trade
}
# Initialisation et lancement du parseur
parser = TardisWebSocketParser(
api_key="YOUR_TARDIS_API_KEY",
exchanges=["binance", "bybit"],
symbols=["BTC-USDT", "ETH-USDT"],
channels=["orderbook-raw", "trade"]
)
# Connexion au serveur WebSocket de Tardis
tardis_url = "wss://api.tardis.dev/v1/feed"
await parser.connect(tardis_url, handlers)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nArrêt demandé par l'utilisateur")
Implémentation JavaScript/TypeScript
Client WebSocket avec gestion d'état
/**
* Client WebSocket haute performance pour l'API Tardis
* Version TypeScript avec typage fort et gestion avancée des erreurs
*/
interface OrderBookLevel {
price: number;
quantity: number;
}
interface OrderBookState {
bids: Map;
asks: Map;
lastUpdateId: number;
sequenceNumber: number;
}
interface TardisMessage {
type: 'orderbook-snapshot' | 'orderbook-update' | 'trade' | 'subscribed' | 'error';
symbol?: string;
[key: string]: unknown;
}
interface WebSocketConfig {
apiKey: string;
exchanges: string[];
symbols: string[];
channels?: string[];
onOrderbookUpdate?: (symbol: string, state: OrderBookState) => void;
onTrade?: (trade: { price: number; quantity: number; side: string; timestamp: number }) => void;
onConnectionChange?: (connected: boolean) => void;
onError?: (error: Error) => void;
}
class TardisWebSocketClient {
private ws: WebSocket | null = null;
private config: Required;
private orderbooks: Map = new Map();
private reconnectAttempts = 0;
private reconnectDelay = 1000;
private maxReconnectDelay = 30000;
private shouldReconnect = true;
private messageQueue: string[] = [];
private isProcessingQueue = false;
constructor(config: WebSocketConfig) {
this.config = {
channels: ['orderbook-raw'],
onOrderbookUpdate: () => {},
onTrade: () => {},
onConnectionChange: () => {},
onError: () => {},
...config
};
}
/**
* Établit la connexion WebSocket avec gestion automatique de la reconnexion
*/
connect(url: string = 'wss://api.tardis.dev/v1/feed'): void {
this.shouldReconnect = true;
this.createConnection(url);
}
private createConnection(url: string): void {
try {
this.ws = new WebSocket(url);
this.setupEventHandlers();
} catch (error) {
this.handleConnectionError(error as Error);
}
}
private setupEventHandlers(): void {
if (!this.ws) return;
this.ws.onopen = () => {
console.log('[Tardis] Connexion établie');
this.config.onConnectionChange(true);
this.reconnectAttempts = 0;
this.reconnectDelay = 1000;
this.sendSubscription();
this.processQueuedMessages();
};
this.ws.onmessage = (event) => {
this.processMessage(event.data);
};
this.ws.onclose = (event) => {
console.log([Tardis] Connexion fermée: ${event.code} - ${event.reason});
this.config.onConnectionChange(false);
if (this.shouldReconnect) {
this.scheduleReconnect();
}
};
this.ws.onerror = (event) => {
const error = new Error('Erreur WebSocket');
console.error('[Tardis] Erreur:', error);
this.config.onError(error);
};
}
private sendSubscription(): void {
const subscription = {
type: 'subscribe',
exchanges: this.config.exchanges,
channels: this.config.channels,
symbols: this.config.symbols.map(s => s.toUpperCase())
};
this.send(JSON.stringify(subscription));
console.log('[Tardis] Subscription envoyée:', subscription);
}
private send(data: string): void {
if (this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(data);
} else {
// Mise en queue pour envoi ultérieur
this.messageQueue.push(data);
}
}
private async processQueuedMessages(): Promise {
if (this.isProcessingQueue) return;
this.isProcessingQueue = true;
while (this.messageQueue.length > 0 && this.ws?.readyState === WebSocket.OPEN) {
const message = this.messageQueue.shift();
if (message) {
this.ws.send(message);
}
}
this.isProcessingQueue = false;
}
private processMessage(rawMessage: string): void {
try {
const message: TardisMessage = JSON.parse(rawMessage);
this.handleMessage(message);
} catch (error) {
console.error('[Tardis] Erreur de parsing:', error);
}
}
private handleMessage(message: TardisMessage): void {
switch (message.type) {
case 'orderbook-snapshot':
this.handleSnapshot(message);
break;
case 'orderbook-update':
this.handleUpdate(message);
break;
case 'trade':
this.handleTrade(message);
break;
case 'subscribed':
console.log('[Tardis] Subscription confirmée:', message);
break;
case 'error':
console.error('[Tardis] Erreur serveur:', message);
break;
default:
console.warn('[Tardis] Type de message inconnu:', message.type);
}
}
private handleSnapshot(message: TardisMessage): void {
const symbol = message.symbol as string;
const state: OrderBookState = {
bids: new Map(),
asks: new Map(),
lastUpdateId: message.lastUpdateId as number || 0,
sequenceNumber: message.sequenceId as number || 0
};
// Parsing des bids
const bids = message.bids as Array<[string, string]> || [];
for (const [price, quantity] of bids) {
state.bids.set(parseFloat(price), parseFloat(quantity));
}
// Parsing des asks
const asks = message.asks as Array<[string, string]> || [];
for (const [price, quantity] of asks) {
state.asks.set(parseFloat(price), parseFloat(quantity));
}
this.orderbooks.set(symbol, state);
console.log([Tardis] Snapshot ${symbol}: ${state.bids.size} bids, ${state.asks.size} asks);
this.config.onOrderbookUpdate(symbol, state);
}
private handleUpdate(message: TardisMessage): void {
const symbol = message.symbol as string;
const state = this.orderbooks.get(symbol);
if (!state) {
console.warn([Tardis] Mise à jour sans snapshot: ${symbol});
return;
}
const updateSeq = message.sequenceId as number;
if (updateSeq <= state.sequenceNumber) {
console.debug([Tardis] Message hors séquence: ${updateSeq} <= ${state.sequenceNumber});
return;
}
// Application des mises à jour de bids
const bids = message.bids as Array<[string, string]> || [];
for (const [priceStr, quantityStr] of bids) {
const price = parseFloat(priceStr);
const quantity = parseFloat(quantityStr);
if (quantity === 0) {
state.bids.delete(price);
} else {
state.bids.set(price, quantity);
}
}
// Application des mises à jour de asks
const asks = message.asks as Array<[string, string]> || [];
for (const [priceStr, quantityStr] of asks) {
const price = parseFloat(priceStr);
const quantity = parseFloat(quantityStr);
if (quantity === 0) {
state.asks.delete(price);
} else {
state.asks.set(price, quantity);
}
}
state.lastUpdateId = message.lastUpdateId as number;
state.sequenceNumber = updateSeq;
this.config.onOrderbookUpdate(symbol, state);
}
private handleTrade(message: TardisMessage): void {
this.config.onTrade({
price: parseFloat(message.price as string),
quantity: parseFloat(message.quantity as string),
side: message.side as string,
timestamp: message.timestamp as number
});
}
private handleConnectionError(error: Error): void {
console.error('[Tardis] Erreur de connexion:', error);
this.config.onError(error);
this.scheduleReconnect();
}
private scheduleReconnect(): void {
const delay = Math.min(
this.reconnectDelay * Math.pow(2, this.reconnectAttempts) + Math.random(),
this.maxReconnectDelay
);
console.log([Tardis] Reconnexion dans ${Math.round(delay)}ms (tentative ${this.reconnectAttempts + 1}));
setTimeout(() => {
this.reconnectAttempts++;
this.createConnection('wss://api.tardis.dev/v1/feed');
}, delay);
}
/**
* Récupère l'état actuel du carnet d'ordres pour un symbole
*/
getOrderBook(symbol: string): OrderBookState | undefined {
return this.orderbooks.get(symbol);
}
/**
* Calcule le prix moyen pondéré par le volume (VWAP) pour un côté du livre
*/
calculateVWAP(symbol: string, side: 'bid' | 'ask', levels: number = 10): number | null {
const state = this.orderbooks.get(symbol);
if (!state) return null;
const book = side === 'bid' ? state.bids : state.asks;
const sortedPrices = Array.from(book.entries()).sort((a, b) =>
side === 'bid' ? b[0] - a[0] : a[0] - b[0]
).slice(0, levels);
if (sortedPrices.length === 0) return null;
let totalVolume = 0;
let weightedSum = 0;
for (const [price, quantity] of sortedPrices) {
weightedSum += price * quantity;
totalVolume += quantity;
}
return totalVolume > 0 ? weightedSum / totalVolume : null;
}
/**
* Ferme la connexion proprement
*/
disconnect(): void {
this.shouldReconnect = false;
this.messageQueue = [];
if (this.ws) {
this.ws.close(1000, 'Déconnexion client');
this.ws = null;
}
}
}
// Export pour usage en module
export { TardisWebSocketClient, WebSocketConfig, OrderBookState };
Exemple d'utilisation côté client
import { TardisWebSocketClient } from './tardis-client';
/**
* Exemple d'implémentation d'un afficheur de carnet d'ordres en temps réel
*/
function initializeOrderBookFeed() {
const client = new TardisWebSocketClient({
apiKey: 'YOUR_TARDIS_API_KEY',
exchanges: ['binance', 'coinbase'],
symbols: ['BTC-USDT', 'ETH-USDT'],
channels: ['orderbook-raw'],
onConnectionChange: (connected) => {
console.log(connected ? '✅ Connecté au flux Tardis' : '❌ Déconnecté');
updateConnectionStatus(connected);
},
onOrderbookUpdate: (symbol, state) => {
// Mise à jour de l'interface utilisateur
renderOrderBook(symbol, state);
// Calcul du spread
const bestBid = Math.max(...state.bids.keys());
const bestAsk = Math.min(...state.asks.keys());
const spread = ((bestAsk - bestBid) / bestBid) * 100;
updateSpreadDisplay(symbol, spread.toFixed(3) + '%');
},
onTrade: (trade) => {
addTradeToFeed({
...trade,
timestamp: new Date(trade.timestamp).toLocaleTimeString()
});
},
onError: (error) => {
console.error('Erreur:', error);
showErrorNotification(error.message);
}
});
// Démarrage de la connexion
client.connect();
// Exemple de calcul VWAP toutes les secondes
setInterval(() => {
const btcVwap = client.calculateVWAP('BTC-USDT', 'bid', 5);
if (btcVwap) {
updateVWAPDisplay('BTC-USDT', btcVwap);
}
}, 1000);
return client;
}
// Fonctions utilitaires pour l'interface
function renderOrderBook(symbol: string, state: any) {
const bidsContainer = document.getElementById(${symbol}-bids);
const asksContainer = document.getElementById(${symbol}-asks);
if (!bidsContainer || !asksContainer) return;
// Tri et limitation des niveaux affichés
const sortedBids = Array.from(state.bids.entries())
.sort((a, b) => b[0] - a[0])
.slice(0, 15);
const sortedAsks = Array.from(state.asks.entries())
.sort((a, b) => a[0] - b[0])
.slice(0, 15);
// Rendu des bids (en vert)
bidsContainer.innerHTML = sortedBids.map(([price, qty]) =>
`
${price.toFixed(2)}
${qty.toFixed(4)}
`
).join('');
// Rendu des asks (en rouge)
asksContainer.innerHTML = sortedAsks.map(([price, qty]) =>
`
${price.toFixed(2)}
${qty.toFixed(4)}
`
).join('');
}
// Démarrage au chargement de la page
document.addEventListener('DOMContentLoaded', () => {
const client = initializeOrderBookFeed();
// Nettoyage à la fermeture de la page
window.addEventListener('beforeunload', () => {
client.disconnect();
});
});
Optimisation des performances
Stratégies de buffering et debouncing
Pour les applications nécessitant un rendu visuel fluide, l'implémentation d'un buffer de messages s'avère bénéfique. Au lieu de traiter chaque mise à jour individuellement, le buffer accumule les messages pendant un intervalle fixe (généralement 50-100ms) puis applique l'ensemble des modifications en une seule opération de rendu. Cette approche réduit significativement la charge CPU sur les applications frontend tout en maintenant une latence perçue acceptable.
Le batching côté serveur représente une alternative élégante lorsque vous contrôlez l'infrastructure de consommation. En regroupant plusieurs mises à jour dans un seul message WebSocket, on réduit le nombre de tours réseau nécessaires au traitement complet du carnet d'ordres. Les gains en performance varient selon la volatilité du marché, avec des améliorations pouvant atteindre 40% en période de forte activité.
Gestion de la mémoire et garbage collection
Le maintien d'un état local du carnet d'ordres implique une gestion rigoureuse de la mémoire. Les structures de données comme les Maps et les objets doivent être nettoyés périodiquement pour éviter l'accumulation de clés orphelines. Une stratégie efficace consiste à reconstruire périodiquement l'état local à partir d'un nouveau snapshot plutôt que de tenter de nettoyer les entrées individuelles.
Pour les applications JavaScript, l'utilisation de WeakMap pour les références temporaires et le évitant les closures circulaires contribue à réduire la pression sur le garbage collector. Les tests de performance montrent qu'une application bien optimisée peut traiter plus de 10 000 messages par seconde sur un serveur modeste, avec une empreinte mémoire stable inférieure à 512 Mo.
Erreurs courantes et solutions
Erreur 1: Messages hors séquence et divergence d'état
Symptôme: Le carnet d'ordres local diverge progressivement du état réel du marché, avec des prix affichés qui ne correspondent plus aux meilleures offres disponibles.
# Solution Python: Implémentation d'un verificateur de séquence robuste
class SequenceVerifier:
"""
Vérifie l'intégrité des séquences de messages pour détecter
les pertes de messages ou les désynchronisations.
"""
def __init__(self, max_gap: int = 1000):
self.max_gap = max_gap
self.last_sequence: Dict[str, int] = {}
self.missing_count: Dict[str, int] = defaultdict(int)
def verify(self, symbol: str, sequence_id: int) -> tuple[bool, Optional[int]]:
"""
Vérifie si le message est dans la bonne séquence.
Retourne (valide,Sequence attendue)
"""
if symbol not in self.last_sequence:
self.last_sequence[symbol] = sequence_id - 1
return True, None
expected = self.last_sequence[symbol] + 1
gap = sequence_id - expected
if gap == 0:
self.last_sequence[symbol] = sequence_id
self.missing_count[symbol] = 0
return True, None
if gap > 0 and gap <= self.max_gap:
self.missing_count[symbol] += gap
logger.warning(
f"Séquence interrompue pour {symbol}: "
f"{gap} messages manquants (reçu {sequence_id}, attendu {expected})"
)
self.last_sequence[symbol] = sequence_id
return False, expected
if gap < 0:
logger.warning(f"Message du passé pour {symbol}: {sequence_id} < {expected}")
return False, expected
# Gap trop important: resynchronisation nécessaire
logger.error(
f"Décalage de séquence critique pour {symbol}: "
f"gap de {gap} messages"
)
return False, None
def request_resync(self, symbol: str) -> dict:
"""Demande un nouveau snapshot pour resynchroniser."""
return {
"type": "resync",
"symbol": symbol,
"reason": "sequence_gap_exceeded"
}
Utilisation dans le parseur
verifier = SequenceVerifier(max_gap=1000)
def _parse_orderbook_update(self, data: dict, symbol: str) -> None:
sequence_id = data.get("sequenceId", 0)
valid, expected = verifier.verify(symbol, sequence_id)
if not valid:
if expected is None:
# Demande de resynchronisation
logger.error(f"Resync requise pour {symbol}")
self._request_resync(symbol)
else:
logger.warning(
f"Message ignoré: seq={sequence_id}, attendu={expected}"
)
return
# Traitement normal si séquence valide
self._apply_update(data, symbol)
Erreur 2: Fuite mémoire par accumulation d'états
Symptôme: L'utilisation mémoire de l'application croît progressivement jusqu'à atteindre les limites du système,伴有 une dégradation progressive des performances.
# Solution: Nettoyage périodique et reconstruction du carnet
class MemoryOptimizedOrderBook:
"""
Carnet d'ordres avec gestion optimisée de la mémoire.
Reconstruction périodique et limitation de la profondeur.
"""
def __init__(self, max_levels: int = 100, rebuild_interval: int = 3600):
self.max_levels = max_levels
self.rebuild_interval = rebuild_interval
self.bids: Dict[float, float] = {}
self.asks: Dict[float, float] = {}
self._last_rebuild = time.time()
self._update_count = 0
def update(self, side: str, price: float, quantity: float) -> None:
"""Mise à jour avec limitation de profondeur."""
book = self.bids if side == "bid" else self.asks
if quantity == 0:
book.pop(price, None)
else:
book[price] = quantity
self._enforce_limit(side)
self._update_count += 1
self._check_rebuild()
def _enforce_limit(self, side: str) -> None:
"""Supprime les niveaux excessifs."""
book = self.bids if side == "bid" else self.asks
if len(book) <= self.max_levels:
return
# Tri et truncation
if side == "bid":
sorted_prices = sorted(book.keys(), reverse=True)
else:
sorted_prices = sorted(book.keys())
# Suppression des niveaux les plus éloignés du meilleur prix
prices_to_remove = sorted_prices[self.max_levels:]
for price in prices_to_remove:
del book[price]
def _check_rebuild(self) -> None:
"""Déclenche une reconstruction si nécessaire."""
current_time = time.time()
if current_time - self._last_rebuild >= self.rebuild_interval:
self._rebuild()
# Reconstruction aussi si trop de mises à jour accumulées
if self._update_count >= 10000:
logger.info("Reconstruction préventive: 10000 mises à jour")
self._rebuild()
def _rebuild(self) -> None:
"""
Reconstruction complète du carnet.
Réinitialise les compteurs et libère la mémoire.
"""
logger.info(
f"Reconstruction du carnet: "
f"bids={len(self.bids)}, asks={len(self.asks)}"
)
# Reconstruction à partir des données actuelles
# Cela force le garbage collector à libérer les références
self.bids = dict(sorted(self.bids.items(), reverse=True)[:self.max_levels])
self.asks = dict(sorted(self.asks.items())[:self.max_levels])
self._last_rebuild = time.time()
self._update_count = 0