Introduction aux Pipelines de Données Orderbook en Temps Réel
Les données de carnet d'ordres (orderbook) constituent le fondement de nombreuses stratégies de trading algorithmique, d'analyse de marché et de systèmes de surveillance des risques. Dans cet article, nous explorerons comment construire un pipeline de données orderbook capable de gérer à la fois la lecture historique et l'intégration en temps réel via WebSocket, avec une latence mesurée en millisecondes.
Étude de Cas : Migration d'une Scale-up Fintech Parisienne
Contexte Métier
Notre cliente, une scale-up fintech parisienne spécialisée dans les signaux de trading algorithmique, gérait un volume quotidien de 2,4 millions de mises à jour de carnets d'ordres provenant de six exchangeurs différents. L'équipe technique, basée à Paris et composée de quatre développeurs seniors, faisait face à des défis croissants de scalabilité et de latence.
Douleurs du Fournisseur Précédent
Avant leur migration vers notre architecture basée sur Tardis WebSocket, l'entreprise utilisait une solution basée sur des API REST polling. Cette approche présentait plusieurs limitations critiques :
- Latence moyenne de 420 millisecondes entre la réception des données par l'exchange et leur traitement par le système client
- Coût mensuel de 4 200 dollars en infrastructure de polling et en bande passante API
- Taux d'erreur de 3,2% lors des pics de volatilité marchés
- Difficulté à maintenir la cohérence des données historiques avec les flux temps réel
- Gestion complexe des reconnexions et des дубликатов lors des interruptions réseau
Étapes Concrètes de la Migration
La migration vers notre architecture Tardis WebSocket s'est déroulée en quatre phases distinctes sur une période de six semaines :
Phase 1 - Analyse et Cartographie (Semaine 1-2) : Audit complet du système existant, identification des points de intégration critiques et définition des métriques de succès. L'équipe a cartographié 847 points de données différents dans leur pipeline existant.
Phase 2 - Développement du Connecteur Tardis (Semaine 2-3) : Implémentation du connecteur WebSocket avec gestion intelligente de la reconnexion, bufferisation des messages et normalisation des formats de données. Le code source du connecteur a été写成 en Python 3.11 avec async/await pour une performance optimale.
Phase 3 - Déploiement Canari (Semaine 4-5) : Déploiement progressif avec répartition du trafic (10% → 30% → 50% → 100%) entre l'ancien système et le nouveau. Cette approche a permis d'identifier et de résoudre les problèmes de compatibilité sans impact sur les utilisateurs finaux.
Phase 4 - Bascule et Optimisation (Semaine 6) : Migration complète vers le nouveau système avec monitoring intensif des 72 premières heures et ajustements de performance en temps réel.
Métriques à 30 Jours Post-Migration
Les résultats mesurés un mois après la migration complète démontrent l'efficacité de l'architecture Tardis WebSocket :
- Latence moyenne réduite de 420ms à 180ms (réduction de 57%)
- Coût mensuel diminuant de 4 200 dollars à 680 dollars (économie de 84%)
- Taux d'erreur réduit à 0,1% (amélioration de 97%)
- Volume traité multiplié par 3,2x sans dégradation de performance
- Satisfaction utilisateur augmentée de 34% selon les enquêtes internes
Architecture Technique du Pipeline Orderbook
Vue d'Ensemble de l'Architecture
Notre architecture de pipeline de données orderbook repose sur trois composants principaux : le collecteur de données via WebSocket, le moteur de traitement stream-processing, et la couche de persistence optimisée pour les requêtes temporelles.
Composant 1 : Connexion WebSocket avec Tardis
Le connecteur WebSocket assure la connexion bidirectionnelle avec les servers de données. Voici l'implémentation complète du connecteur avec gestion avancée des erreurs et reconnexion automatique :
import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Callable
import hashlib
import hmac
class TardisWebSocketClient:
"""Client WebSocket pour la connexion aux flux Tardis avec support historique et temps réel."""
def __init__(
self,
api_key: str,
api_secret: str,
exchanges: List[str],
channels: List[str],
on_orderbook_update: Callable,
on_error: Callable,
base_url: str = "wss://api.tardis.dev/v1/ws"
):
self.api_key = api_key
self.api_secret = api_secret
self.exchanges = exchanges
self.channels = channels
self.on_orderbook_update = on_orderbook_update
self.on_error = on_error
self.base_url = base_url
self.websocket = None
self.reconnect_delay = 1
self.max_reconnect_delay = 60
self.is_running = False
self.message_buffer = []
self.buffer_size = 1000
self.last_heartbeat = None
self.sequence_numbers = {}
self.logger = logging.getLogger(__name__)
def _generate_auth_signature(self, timestamp: str, nonce: str) -> str:
"""Génère la signature d'authentification HMAC-SHA256."""
message = f"{timestamp}{nonce}{self.api_key}"
signature = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
def _create_auth_message(self) -> Dict:
"""Crée le message d'authentification pour Tardis."""
timestamp = datetime.utcnow().isoformat() + 'Z'
nonce = hashlib.sha256(
f"{timestamp}{self.api_key}".encode()
).hexdigest()[:16]
signature = self._generate_auth_signature(timestamp, nonce)
return {
"type": "auth",
"apiKey": self.api_key,
"timestamp": timestamp,
"nonce": nonce,
"signature": signature,
"signatureVersion": 2
}
def _create_subscribe_message(
self,
exchange: str,
channel: str,
symbol: Optional[str] = None,
from_timestamp: Optional[datetime] = None
) -> Dict:
"""Crée le message d'abonnement pour un canal spécifique."""
message = {
"type": "subscribe",
"exchange": exchange,
"channel": channel,
}
if symbol:
message["symbol"] = symbol
if from_timestamp:
message["fromTimestamp"] = from_timestamp.isoformat() + 'Z'
return message
async def connect(self, historical_from: Optional[datetime] = None):
"""Établit la connexion WebSocket et s'abonne aux canaux."""
try:
self.logger.info(f"Connexion à {self.base_url}")
self.websocket = await asyncio.get_event_loop().create_connection(
self._create_websocket_endpoint(),
self.base_url.replace('wss://', '').replace('https://', '').replace('http://', ''),
ssl=True
)
# Envoyer le message d'authentification
auth_message = self._create_auth_message()
await self._send_message(auth_message)
# Attendre la confirmation d'authentification
auth_response = await self._wait_for_message_type("auth_success", timeout=10)
if not auth_response:
raise Exception("Échec de l'authentification avec Tardis")
self.logger.info("Authentification réussie")
# S'abonner aux canaux pour chaque exchange
for exchange in self.exchanges:
for channel in self.channels:
subscribe_msg = self._create_subscribe_message(
exchange=exchange,
channel=channel,
from_timestamp=historical_from if historical_from else None
)
await self._send_message(subscribe_msg)
self.logger.info(f"Abonnement à {exchange}/{channel}")
self.is_running = True
self.reconnect_delay = 1
except Exception as e:
self.logger.error(f"Erreur de connexion : {e}")
self.on_error(f"Connexion_erreur: {str(e)}")
await self._schedule_reconnect(historical_from)
async def _create_websocket_endpoint(self):
"""Crée l'endpoint WebSocket personnalisé."""
from websockets.client import WebSocketClientProtocol
class TardisWebSocket(WebSocketClientProtocol):
async def get_mask(self, mask):
return mask
return TardisWebSocket
async def _send_message(self, message: Dict):
"""Envoie un message JSON via WebSocket."""
if self.websocket and self.websocket.open:
await self.websocket.send(json.dumps(message))
async def _wait_for_message_type(self, message_type: str, timeout: float = 5) -> Optional[Dict]:
"""Attend un type de message spécifique avec timeout."""
try:
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < timeout:
if self.websocket and self.websocket.open:
message = await asyncio.wait_for(
self.websocket.recv(),
timeout=timeout
)
data = json.loads(message)
if data.get("type") == message_type:
return data
return None
except asyncio.TimeoutError:
return None
async def _schedule_reconnect(self, historical_from: Optional[datetime]):
"""Planifie une reconnexion avec backoff exponentiel."""
self.is_running = False
self.logger.info(f"Reconnexion dans {self.reconnect_delay} secondes...")
await asyncio.sleep(self.reconnect_delay)
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
await self.connect(historical_from)
async def start_historical_replay(
self,
start_time: datetime,
end_time: datetime,
exchange: str,
channel: str,
symbol: Optional[str] = None
):
"""Démarre la relecture historique avec progression."""
self.logger.info(f"Début de la relecture historique {start_time} -> {end_time}")
current_time = start_time
total_duration = (end_time - start_time).total_seconds()
processed_messages = 0
while current_time < end_time and self.is_running:
try:
# Créer une connexion pour la période historique
historical_client = TardisWebSocketClient(
api_key=self.api_key,
api_secret=self.api_secret,
exchanges=[exchange],
channels=[channel],
on_orderbook_update=self.on_orderbook_update,
on_error=self.on_error
)
await historical_client.connect(historical_from=current_time)
# Traiter les messages historiques
while current_time < end_time and historical_client.is_running:
message = await asyncio.wait_for(
historical_client.websocket.recv(),
timeout=1.0
)
await self._process_message(message)
processed_messages += 1
# Calculer et afficher la progression
elapsed = (current_time - start_time).total_seconds()
progress = (elapsed / total_duration) * 100
if processed_messages % 10000 == 0:
self.logger.info(f"Progression: {progress:.2f}% ({processed_messages} messages)")
current_time += timedelta(milliseconds=100)
await historical_client.disconnect()
except Exception as e:
self.logger.error(f"Erreur pendant la relecture : {e}")
await asyncio.sleep(1)
self.logger.info(f"Relecture terminée: {processed_messages} messages traités")
async def _process_message(self, raw_message: str):
"""Traite un message entrant avec validation et transformation."""
try:
data = json.loads(raw_message)
message_type = data.get("type")
if message_type == "heartbeat":
self.last_heartbeat = datetime.utcnow()
return
elif message_type == "orderbook_snapshot":
normalized = self._normalize_orderbook_snapshot(data)
await self._add_to_buffer(normalized)
await self.on_orderbook_update(normalized)
elif message_type == "orderbook_update":
normalized = self._normalize_orderbook_update(data)
await self._add_to_buffer(normalized)
await self.on_orderbook_update(normalized)
elif message_type == "error":
self.logger.error(f"Erreur Tardis : {data}")
self.on_error(f"Tardis_erreur: {data.get('message', 'Unknown')}")
# Validation du numéro de séquence pour détection des pertes
if "exchangeSequence" in data:
exchange_id = data.get("exchange")
seq = data.get("exchangeSequence")
if exchange_id in self.sequence_numbers:
expected = self.sequence_numbers[exchange_id] + 1
if seq != expected:
self.logger.warning(
f"Séquence interrompue pour {exchange_id}: "
f"attendu {expected}, reçu {seq}"
)
self.on_error(f"Sequence_gap_{exchange_id}:{expected}:{seq}")
self.sequence_numbers[exchange_id] = seq
except json.JSONDecodeError as e:
self.logger.error(f"Erreur de parsing JSON : {e}")
self.on_error(f"JSON_parse_error: {str(e)}")
except Exception as e:
self.logger.error(f"Erreur de traitement : {e}")
self.on_error(f"Traitement_erreur: {str(e)}")
def _normalize_orderbook_snapshot(self, data: Dict) -> Dict:
"""Normalise un snapshot de carnet d'ordres vers format interne."""
return {
"timestamp": datetime.fromisoformat(data["timestamp"].replace('Z', '+00:00')),
"exchange": data["exchange"],
"symbol": data["symbol"],
"type": "snapshot",
"bids": [[float(price), float(size)] for price, size in data.get("bids", [])],
"asks": [[float(price), float(size)] for price, size in data.get("asks", [])],
"sequence": data.get("exchangeSequence"),
"local_timestamp": datetime.utcnow()
}
def _normalize_orderbook_update(self, data: Dict) -> Dict:
"""Normalise une mise à jour de carnet d'ordres vers format interne."""
normalized = {
"timestamp": datetime.fromisoformat(data["timestamp"].replace('Z', '+00:00')),
"exchange": data["exchange"],
"symbol": data["symbol"],
"type": "update",
"sequence": data.get("exchangeSequence"),
"local_timestamp": datetime.utcnow()
}
if "bids" in data:
normalized["bids"] = [
[float(price), float(size)] for price, size in data["bids"]
]
if "asks" in data:
normalized["asks"] = [
[float(price), float(size)] for price, size in data["asks"]
]
return normalized
async def _add_to_buffer(self, data: Dict):
"""Ajoute les données au buffer circulaire."""
self.message_buffer.append(data)
if len(self.message_buffer) > self.buffer_size:
self.message_buffer.pop(0)
async def run(self):
"""Boucle principale de traitement des messages."""
await self.connect()
while self.is_running:
try:
if self.websocket and self.websocket.open:
message = await asyncio.wait_for(
self.websocket.recv(),
timeout=30.0
)
await self._process_message(message)
except asyncio.TimeoutError:
if not self.last_heartbeat:
self.logger.warning("Pas de heartbeat reçu")
elif (datetime.utcnow() - self.last_heartbeat).seconds > 60:
self.logger.error("Heartbeat expiré, reconnexion...")
await self._schedule_reconnect(None)
except Exception as e:
self.logger.error(f"Erreur dans la boucle principale : {e}")
self.on_error(f"Boucle_principale_erreur: {str(e)}")
await self._schedule_reconnect(None)
async def disconnect(self):
"""Ferme proprement la connexion WebSocket."""
self.is_running = False
if self.websocket and self.websocket.open:
await self.websocket.close()
self.logger.info("Déconnexion effectuée")
Exemple d'utilisation avec OrderbookAggregator
class OrderbookAggregator:
"""Agrégateur de données orderbook multi-sources avec gestion du orderbook local."""
def __init__(self):
self.orderbooks: Dict[str, Dict] = {}
self.update_counters: Dict[str, int] = {}
self.last_update_times: Dict[str, datetime] = {}
self.latencies: List[float] = []
async def on_orderbook_update(self, data: Dict):
"""Callback pour traiter les mises à jour du orderbook."""
key = f"{data['exchange']}:{data['symbol']}"
if key not in self.orderbooks:
self.orderbooks[key] = {
"bids": {},
"asks": {},
"last_sequence": None
}
ob = self.orderbooks[key]
# Traiter le snapshot complet
if data["type"] == "snapshot":
ob["bids"] = {price: size for price, size in data["bids"]}
ob["asks"] = {price: size for price, size in data["asks"]}
ob["last_sequence"] = data["sequence"]
# Appliquer les mises à jour delta
else:
for price, size in data.get("bids", []):
if size == 0:
ob["bids"].pop(price, None)
else:
ob["bids"][price] = size
for price, size in data.get("asks", []):
if size == 0:
ob["asks"].pop(price, None)
else:
ob["asks"][price] = size
ob["last_sequence"] = data["sequence"]
# Calculer et stocker la latence
latency_ms = (data["local_timestamp"] - data["timestamp"]).total_seconds() * 1000
self.latencies.append(latency_ms)
if len(self.latencies) > 10000:
self.latencies.pop(0)
self.update_counters[key] = self.update_counters.get(key, 0) + 1
self.last_update_times[key] = datetime.utcnow()
# Log les métriques toutes les 1000 mises à jour
if self.update_counters[key] % 1000 == 0:
avg_latency = sum(self.latencies[-1000:]) / len(self.latencies[-1000:])
print(f"{key}: {self.update_counters[key]} updates, "
f"latence moyenne: {avg_latency:.2f}ms")
def get_best_bid_ask(self, exchange: str, symbol: str) -> Optional[Dict]:
"""Retourne le meilleur bid/ask pour un paire donnée."""
key = f"{exchange}:{symbol}"
if key not in self.orderbooks:
return None
ob = self.orderbooks[key]
bids = ob["bids"]
asks = ob["asks"]
if not bids or not asks:
return None
best_bid = max(bids.items(), key=lambda x: x[0])
best_ask = min(asks.items(), key=lambda x: x[0])
return {
"exchange": exchange,
"symbol": symbol,
"best_bid": best_bid[0],
"best_bid_size": best_bid[1],
"best_ask": best_ask[0],
"best_ask_size": best_ask[1],
"spread": best_ask[0] - best_bid[0],
"spread_pct": ((best_ask[0] - best_bid[0]) / best_bid[0]) * 100,
"mid_price": (best_bid[0] + best_ask[0]) / 2,
"timestamp": self.last_update_times.get(key)
}
async def main():
"""Point d'entrée principal pour tester le pipeline."""
# Configuration du logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Initialisation de l'agrégateur
aggregator = OrderbookAggregator()
# Liste des erreurs pour monitoring
errors = []
def error_handler(error_msg: str):
errors.append({
"timestamp": datetime.utcnow(),
"error": error_msg
})
print(f"⚠️ Erreur détectée: {error_msg}")
# Configuration du client WebSocket
client = TardisWebSocketClient(
api_key="VOTRE_CLE_API_TARDIS",
api_secret="VOTRE_SECRET_API_TARDIS",
exchanges=["binance", "coinbase", "kraken"],
channels=["orderbook"],
on_orderbook_update=aggregator.on_orderbook_update,
on_error=error_handler
)
# Option 1: Connexion temps réel uniquement
print("Démarrage de la connexion temps réel...")
await client.run()
# Option 2: Relecture historique puis connexion temps réel
# historical_start = datetime(2024, 1, 1, 0, 0, 0)
# historical_end = datetime(2024, 1, 1, 1, 0, 0)
#
# print("Démarrage de la relecture historique...")
# await client.start_historical_replay(
# start_time=historical_start,
# end_time=historical_end,
# exchange="binance",
# channel="orderbook",
# symbol="BTC-USD"
# )
#
# print("Connexion au flux temps réel...")
# await client.run()
if __name__ == "__main__":
asyncio.run(main())Composant 2 : Moteur de Stream Processing avec Ordre d'Exécution Garanti
Le moteur de traitement stream est conçu pour garantir l'ordre d'exécution des messages et gérer les problématiques de backpressure. Voici l'implémentation complète avec support pour le orderbook local et la détection d'anomalies :
import asyncio
import heapq
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from datetime import datetime, timedelta
from enum import Enum
import statistics
import logging
class ProcessingPriority(Enum):
"""Priorités de traitement des messages."""
CRITICAL = 1 # Ordrebooks complets
HIGH = 2 # Mises à jour avec sequence gap
NORMAL = 3 # Mises à jour régulières
LOW = 4 # Données historiques moins urgentes
@dataclass(order=True)
class PrioritizedMessage:
"""Message avec priorité pour le traitement par heap."""
priority: int
timestamp: datetime = field(compare=False)
sequence: int = field(compare=False, default=0)
exchange: str = field(compare=False, default="")
data: Dict = field(compare=False, default_factory=dict)
retry_count: int = field(compare=False, default=0)
@dataclass
class ProcessingMetrics:
"""Métriques de performance du traitement."""
messages_processed: int = 0
messages_failed: int = 0
processing_times: List[float] = field(default_factory=list)
queue_sizes: List[int] = field(default_factory=list)
sequence_gaps: int = 0
last_sequence_per_exchange: Dict[str, int] = field(default_factory=dict)
def get_statistics(self) -> Dict[str, Any]:
"""Calcule les statistiques de performance."""
if not self.processing_times:
return {"error": "Pas assez de données"}
sorted_times = sorted(self.processing_times)
return {
"total_processed": self.messages_processed,
"total_failed": self.messages_failed,
"success_rate": self.messages_processed /
(self.messages_processed + self.messages_failed) * 100,
"avg_processing_time_ms": statistics.mean(self.processing_times),
"p50_processing_time_ms": sorted_times[len(sorted_times) // 2],
"p95_processing_time_ms": sorted_times[int(len(sorted_times) * 0.95)],
"p99_processing_time_ms": sorted_times[int(len(sorted_times) * 0.99)],
"max_processing_time_ms": max(self.processing_times),
"avg_queue_size": statistics.mean(self.queue_sizes) if self.queue_sizes else 0,
"max_queue_size": max(self.queue_sizes) if self.queue_sizes else 0,
"sequence_gaps": self.sequence_gaps
}
class OrderbookLocal:
"""Gestionnaire de orderbook local avec reconstruction incrémentale."""
def __init__(self, max_levels: int = 20):
self.max_levels = max_levels
self.bids: Dict[float, float] = {} # price -> size
self.asks: Dict[float, float] = {} # price -> size
self.last_update: Optional[datetime] = None
self.update_count: int = 0
self.spread_history: List[float] = []
self.logger = logging.getLogger(__name__)
def apply_snapshot(self, bids: List[List[float]], asks: List[List[float]], timestamp: datetime):
"""Applique un snapshot complet."""
self.bids.clear()
self.asks.clear()
# Trier et limiter les niveaux
for price, size in sorted(bids, key=lambda x: -x[0])[:self.max_levels]:
self.bids[price] = size
for price, size in sorted(asks, key=lambda x: x[0])[:self.max_levels]:
self.asks[price] = size
self.last_update = timestamp
self.update_count += 1
self._update_spread()
def apply_update(self, bids: List[List[float]], asks: List[List[float]], timestamp: datetime):
"""Applique une mise à jour incrémentale."""
for price, size in bids:
if size == 0:
self.bids.pop(price, None)
else:
self.bids[price] = size
for price, size in asks:
if size == 0:
self.asks.pop(price, None)
else:
self.asks[price] = size
self.last_update = timestamp
self.update_count += 1
self._update_spread()
def _update_spread(self):
"""Met à jour l'historique du spread."""
best_bid = max(self.bids.keys()) if self.bids else 0
best_ask = min(self.asks.keys()) if self.asks else 0
if best_bid and best_ask:
spread = best_ask - best_bid
self.spread_history.append(spread)
if len(self.spread_history) > 1000:
self.spread_history.pop(0)
def get_depth(self, levels: int = 10) -> Dict[str, Any]:
"""Retourne la profondeur du marché."""
sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:levels]
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
bid_volume = sum(size for _, size in sorted_bids)
ask_volume = sum(size for _, size in sorted_asks)
bid_value = sum(price * size for price, size in sorted_bids)
ask_value = sum(price * size for price, size in sorted_asks)
return {
"timestamp": self.last_update,
"bid_levels": [
{"price": price, "size": size, "cumulative": 0}
for price, size in sorted_bids
],
"ask_levels": [
{"price": price, "size": size, "cumulative": 0}
for price, size in sorted_asks
],
"total_bid_volume": bid_volume,
"total_ask_volume": ask_volume,
"bid_value": bid_value,
"ask_value": ask_value,
"imbalance": (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0,
"spread": (min(self.asks.keys()) - max(self.bids.keys())) if self.bids and self.asks else None,
"mid_price": (min(self.asks.keys()) + max(self.bids.keys())) / 2 if self.bids and self.asks else None
}
def detect_anomalies(self, volatility_threshold: float = 0.05) -> List[Dict]:
"""Détecte les anomalies dans le orderbook."""
anomalies = []
if len(self.spread_history) < 10:
return anomalies
# Calculer la volatilité du spread
spread_std = statistics.stdev(self.spread_history[-100:])
spread_mean = statistics.mean(self.spread_history[-100:])
if spread_mean > 0 and spread_std / spread_mean > volatility_threshold:
anomalies.append({
"type": "high_spread_volatility",
"severity": "warning",
"message": f"Volatilité du spread elevated: {spread_std/spread_mean:.2%}",
"timestamp": datetime.utcnow()
})
# Détecter les ordres de taille inhabituelle
all_bid_sizes = list(self.bids.values())
all_ask_sizes = list(self.asks.values())
if all_bid_sizes:
avg_bid = statistics.mean(all_bid_sizes)
max_bid = max(all_bid_sizes)
if max_bid > avg_bid * 10:
anomalies.append({
"type": "large_order_detected",
"severity": "info",
"message": f"Gros ordre détecté: {max_bid} vs moyenne {avg_bid:.2f}",
"side": "bid",
"size": max_bid,
"timestamp": datetime.utcnow()
})
return anomalies
class StreamProcessor:
"""Moteur de traitement de flux avec gestion de l'ordre et backpressure."""
def __init__(
self,
max_queue_size: int = 100000,
batch_size: int = 100,
processing_timeout: float = 1.0,
enable_sequence_validation: bool = True
):
self.max_queue_size = max_queue_size
self.batch_size = batch_size
self.processing_timeout = processing_timeout
self.enable_sequence_validation = enable_sequence_validation
# Files de priorité pour le traitement
self.priority_queues: Dict[ProcessingPriority, List] = {
priority: [] for priority in ProcessingPriority
}
# Orderbooks locaux par exchange/symbol
self.orderbooks: Dict[str, OrderbookLocal] = {}
# Métriques de performance
self.metrics = ProcessingMetrics()
# Callbacks
self.processors: Dict[str, Callable] = {}
self.anomaly_handlers: List[Callable] = []
# État
self.is_running = False
self.is_paused = False
self.total_queue_size = 0
self.logger = logging.getLogger(__name__)
def register_processor(self, event_type: str, handler: Callable):
"""Enregistre un processeur pour un type d'événement."""
self.processors[event_type] = handler
def register_anomaly_handler(self, handler: Callable):
"""Enregistre un gestionnaire d'anomalies."""
self.anomaly_handlers.append(handler)
def _determine_priority(self, message: Dict) -> ProcessingPriority:
"""Détermine la priorité de traitement d'un message."""
if message.get("type") == "snapshot":
return ProcessingPriority.CRITICAL
if self.enable_sequence_validation:
exchange = message.get("exchange", "")
sequence = message.get("sequence", 0)
if exchange in self.metrics.last_sequence_per_exchange:
expected = self.metrics.last_sequence_per_exchange[exchange] + 1
if sequence != expected:
self.metrics.sequence_gaps += 1
return ProcessingPriority.HIGH
return ProcessingPriority.NORMAL
def enqueue(self, message: Dict) -> bool:
"""Ajoute un message à la file de traitement."""
if self.total_queue_size >= self.max_queue_size:
self.logger.warning(f"Queue pleine ({self.max_queue_size}), backpressure actif")
return False
priority = self._determine_priority(message)
prioritized = PrioritizedMessage(
priority=priority.value,
timestamp=message.get("timestamp", datetime.utcnow()),
sequence=message.get("sequence", 0),
exchange=message.get("exchange", ""),
data=message
)
heapq.heappush(self.priority_queues[priority], prioritized)
self.total_queue_size += 1
return True
def _get_next_message(self) -> Optional[PrioritizedMessage]:
"""Récupère