En tant qu'ingénieur qui a passé 18 mois à construire des systèmes de trading haute fréquence sur les principales plateformes d'échange, je peux vous confirmer que la maîtrise du order book en temps réel est la compétence la plus critique pour tout projet de trading algorithmique. Après avoir testé toutes les approches possibles, je vais vous partager ma stack complète, incluant une comparaison objective des coûts d'infrastructure IA que j'utilise désormais pour l'analyse prédictive.
Comparatif des Coûts IA pour le Traitement de Données de Marché
Puisque le parsing du order book génère des volumes massifs de données à analyser (sentiment de marché, détection de manipulation, patterns de liquidité), j'utilise désormais des modèles IA pour enrichir mes analyses. Voici ma comparaison actualisée 2026 des principaux providers :
| Provider / Modèle | Prix $/MTok Output | Latence Moyenne | Coût 10M Tokens/mois | Avantage Clé |
|---|---|---|---|---|
| HolySheep - DeepSeek V3.2 | $0.42 | <50ms | $4,200 | Meilleur rapport qualité/prix |
| Google - Gemini 2.5 Flash | $2.50 | ~80ms | $25,000 | Excellente performance |
| OpenAI - GPT-4.1 | $8.00 | ~120ms | $80,000 | Écosystème mature |
| Anthropic - Claude Sonnet 4.5 | $15.00 | ~150ms | $150,000 | Reasoning avancé |
Économie réalisées avec HolySheep : 85%+ par rapport aux providers occidentaux. Pour mon usage de 10 millions de tokens/mois en analyse de order book, je passe de $80,000 à $4,200 — une différence qui change radicalement la rentabilité de mon infrastructure.
Pourquoi Connecter le Order Book OKX à une Analyse IA ?
La question n'est plus « si » mais « comment » intégrer l'intelligence artificielle dans votre pipeline de données de marché. Le order book contient :
- La microstructure du marché en temps réel
- Les concentrations de liquidité (zones de support/résistance)
- Les patterns de spoofing et wash trading
- Les anticipations de mouvement de prix
En analysant ces données via une API IA performante comme HolySheep, je peux automatiquement générer des signaux de trading avec un coût de traitement inférieur à $0.01 par bougie analysée.
Architecture de Connexion OKX WebSocket
1. Installation et Configuration Initiale
# Installation des dépendances Python
pip install websocket-client websockets pandas numpy msgpack
pip install aiohttp asyncio aiofiles
Structure du projet
trading-bot/
├── config/
│ └── settings.py
├── src/
│ ├── websocket_client.py
│ ├── orderbook_parser.py
│ ├── storage_handler.py
│ └── ai_analyzer.py
├── tests/
│ └── test_orderbook.py
└── requirements.txt2. Client WebSocket OKX — Code Production Ready
# src/websocket_client.py
import asyncio
import json
import hmac
import base64
import hashlib
import time
from typing import Dict, List, Callable, Optional
from datetime import datetime
class OKXWebSocketClient:
"""
Client WebSocket haute performance pour le order book OKX.
Supporte les connexions privées et publiques.
"""
def __init__(
self,
api_key: str = "",
api_secret: str = "",
passphrase: str = "",
sandbox: bool = True
):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.sandbox = sandbox
# URLs WebSocket OKX
self.base_url = (
"wss://wspap.okx.com:8443/ws/v5/private"
if not sandbox
else "wss://wspap.okx.com:8443/ws/v5/public"
)
self.websocket = None
self.subscriptions: List[Dict] = []
self.orderbook_data: Dict[str, Dict] = {}
self.callbacks: List[Callable] = []
self.reconnect_delay = 1
self.max_reconnect_delay = 60
def _get_timestamp(self) -> str:
"""Génère le timestamp pour l'authentification."""
return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""
Génère la signature pour l'authentification OKX.
Algorithme: HMAC-SHA256
"""
message = timestamp + method + path + body
mac = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
def _get_auth_params(self) -> Dict:
"""Prépare les paramètres d'authentification pour les connexions privées."""
timestamp = self._get_timestamp()
signature = self.sign(timestamp, "GET", "/users/self/verify")
return {
"op": "login",
"args": [
{
"apiKey": self.api_key,
"passphrase": self.passphrase,
"timestamp": timestamp,
"sign": signature
}
]
}
async def connect(self, use_auth: bool = False):
"""Établit la connexion WebSocket."""
import websockets
self.websocket = await websockets.connect(
self.base_url,
ping_interval=20,
ping_timeout=10,
max_size=10 * 1024 * 1024 # 10MB max message
)
if use_auth and self.api_key:
auth_params = self._get_auth_params()
await self.websocket.send(json.dumps(auth_params))
response = await asyncio.wait_for(
self.websocket.recv(),
timeout=10
)
resp_data = json.loads(response)
if resp_data.get("code") != "0":
raise ConnectionError(f"Auth échouée: {resp_data}")
print(f"✓ Connexion WebSocket établie: {self.base_url}")
return self
async def subscribe_orderbook(
self,
inst_id: str = "BTC-USDT",
depth: int = 400
):
"""
Souscrit au flux du order book.
Args:
inst_id: Identifiant de l'instrument (ex: BTC-USDT)
depth: Profondeur du order book (5, 25, 400, ou full)
"""
subscribe_params = {
"op": "subscribe",
"args": [
{
"channel": "books" if depth <= 25 else "books-l2-tbt",
"instId": inst_id,
"snapshot": "false" # true pour ordre complet, false pour incrémental
}
]
}
await self.websocket.send(json.dumps(subscribe_params))
self.subscriptions.append({"channel": "books", "instId": inst_id})
print(f"✓ Subscribe order book: {inst_id} (depth={depth})")
async def subscribe_trades(self, inst_id: str = "BTC-USDT"):
"""Souscrit au flux des trades en temps réel."""
subscribe_params = {
"op": "subscribe",
"args": [
{
"channel": "trades",
"instId": inst_id
}
]
}
await self.websocket.send(json.dumps(subscribe_params))
self.subscriptions.append({"channel": "trades", "instId": inst_id})
async def listen(self):
"""Boucle principale d'écoute des messages."""
try:
async for message in self.websocket:
await self._process_message(message)
except websockets.exceptions.ConnectionClosed:
print("⚠ Connexion fermée, tentative de reconnexion...")
await self._reconnect()
async def _process_message(self, raw_message: str):
"""Parse et dispatch les messages reçus."""
try:
data = json.loads(raw_message)
# Gestion des réponses d'authentification/subscription
if "event" in data:
print(f"Event reçu: {data['event']}")
return
# Message de données
if "arg" in data and "data" in data:
channel = data["arg"]["channel"]
messages = data["data"]
for msg in messages:
if channel == "books" or channel == "books-l2-tbt":
await self._process_orderbook(msg)
elif channel == "trades":
await self._process_trade(msg)
except json.JSONDecodeError as e:
print(f"Erreur parsing JSON: {e}")
async def _process_orderbook(self, data: Dict):
"""
Parse les données du order book OKX.
Structure OKX:
- asks: [[price, qty, sz], ...]
- bids: [[price, qty, sz], ...]
- ts: timestamp en millisecondes
- seqId: ID de séquence pour détection des gaps
"""
inst_id = data.get("instId", "UNKNOWN")
parsed = {
"symbol": inst_id,
"timestamp": int(data["ts"]),
"datetime": datetime.fromtimestamp(int(data["ts"]) / 1000).isoformat(),
"seqId": data.get("seqId", 0),
"asks": [[float(p), float(q)] for p, q in data.get("asks", [])],
"bids": [[float(p), float(q)] for p, q in data.get("bids", [])],
"ask_depth_1": float(data["asks"][0][0]) if data.get("asks") else None,
"bid_depth_1": float(data["bids"][0][0]) if data.get("bids") else None,
"spread": None,
"mid_price": None,
"imbalance": None
}
# Calculs supplémentaires
if parsed["ask_depth_1"] and parsed["bid_depth_1"]:
parsed["spread"] = parsed["ask_depth_1"] - parsed["bid_depth_1"]
parsed["mid_price"] = (parsed["ask_depth_1"] + parsed["bid_depth_1"]) / 2
ask_vol = sum(q for _, q in parsed["asks"][:10])
bid_vol = sum(q for _, q in parsed["bids"][:10])
total_vol = ask_vol + bid_vol
if total_vol > 0:
parsed["imbalance"] = (bid_vol - ask_vol) / total_vol
# Stockage local
self.orderbook_data[inst_id] = parsed
# Notification callbacks
for callback in self.callbacks:
await callback(parsed)
async def _process_trade(self, data: Dict):
"""Parse les données de trade."""
trade = {
"symbol": data["instId"],
"trade_id": data["tradeId"],
"timestamp": int(data["ts"]),
"price": float(data["px"]),
"quantity": float(data["sz"]),
"side": data["side"], # buy ou sell
"is_buyer_maker": data.get("instId") == "SELL"
}
for callback in self.callbacks:
await callback(trade)
def register_callback(self, callback: Callable):
"""Enregistre un callback pour recevoir les mises à jour."""
self.callbacks.append(callback)
async def _reconnect(self):
"""Gère la reconnexion automatique avec backoff exponentiel."""
delay = self.reconnect_delay
max_delay = self.max_reconnect_delay
while True:
try:
await asyncio.sleep(delay)
await self.connect(use_auth=bool(self.api_key))
# Resubscribe aux canaux
for sub in self.subscriptions:
if sub["channel"] == "books":
await self.subscribe_orderbook(sub["instId"])
elif sub["channel"] == "trades":
await self.subscribe_trades(sub["instId"])
self.reconnect_delay = 1
break
except Exception as e:
print(f"Reconnexion échouée: {e}")
delay = min(delay * 2, max_delay)
Usage example
async def main():
client = OKXWebSocketClient(sandbox=True)
# Callback pour traiter les données
async def on_orderbook_update(data):
print(f"[{data['datetime']}] {data['symbol']}")
print(f" Bid: {data['bid_depth_1']} | Ask: {data['ask_depth_1']}")
print(f" Spread: {data['spread']} | Mid: {data['mid_price']}")
print(f" Imbalance: {data['imbalance']:.4f}")
client.register_callback(on_orderbook_update)
await client.connect(use_auth=False)
await client.subscribe_orderbook("BTC-USDT", depth=25)
await client.subscribe_trades("BTC-USDT")
await client.listen()
if __name__ == "__main__":
asyncio.run(main())3. Système de Stockage avec PostgreSQL et TimescaleDB
# src/storage_handler.py
import asyncio
import asyncpg
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from contextlib import asynccontextmanager
import json
class OrderBookStorage:
"""
Gestionnaire de stockage haute performance pour le order book.
Utilise TimescaleDB pour l'insertion rapide de séries temporelles.
"""
def __init__(
self,
dsn: str = "postgresql://user:pass@localhost:5432/orderbook",
batch_size: int = 100,
flush_interval: float = 1.0
):
self.dsn = dsn
self.batch_size = batch_size
self.flush_interval = flush_interval
self.pool: Optional[asyncpg.Pool] = None
self.buffer: List[Dict] = []
self.last_flush = datetime.utcnow()
self._lock = asyncio.Lock()
async def connect(self):
"""Initialise la connexion à la base de données."""
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=5,
max_size=20,
command_timeout=60
)
await self._create_tables()
print("✓ Connexion PostgreSQL établie")
async def _create_tables(self):
"""Crée les tables hypertable pour TimescaleDB."""
async with self.pool.acquire() as conn:
# Table principale order_book_snapshots
await conn.execute("""
CREATE TABLE IF NOT EXISTS order_book_snapshots (
time TIMESTAMPTZ NOT NULL,
symbol TEXT NOT NULL,
bid_price NUMERIC(20, 8) NOT NULL,
bid_quantity NUMERIC(20, 8) NOT NULL,
ask_price NUMERIC(20, 8) NOT NULL,
ask_quantity NUMERIC(20, 8) NOT NULL,
seq_id BIGINT NOT NULL,
spread NUMERIC(20, 8),
mid_price NUMERIC(20, 8),
imbalance NUMERIC(10, 6),
raw_data JSONB
);
""")
# Création de la hypertable TimescaleDB
try:
await conn.execute("""
SELECT create_hypertable(
'order_book_snapshots',
'time',
if_not_exists => TRUE
);
""")
print("✓ Hypertable TimescaleDB créée")
except asyncpg.exceptions.DuplicateTable:
pass # Hypertable existe déjà
# Index pour requêtes fréquentes
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_orderbook_symbol_time
ON order_book_snapshots (symbol, time DESC);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_orderbook_seq
ON order_book_snapshots (symbol, seq_id);
""")
async def insert_snapshot(self, data: Dict):
"""
Insère un snapshot du order book dans le buffer.
Les données sont flushées périodiquement pour performance.
"""
async with self._lock:
snapshot = {
"time": datetime.fromtimestamp(data["timestamp"] / 1000),
"symbol": data["symbol"],
"bid_price": data.get("bid_depth_1"),
"bid_quantity": data["bids"][0][1] if data.get("bids") else 0,
"ask_price": data.get("ask_depth_1"),
"ask_quantity": data["asks"][0][1] if data.get("asks") else 0,
"seq_id": data.get("seqId", 0),
"spread": data.get("spread"),
"mid_price": data.get("mid_price"),
"imbalance": data.get("imbalance"),
"raw_data": json.dumps({
"asks": data.get("asks", [])[:10],
"bids": data.get("bids", [])[:10]
})
}
self.buffer.append(snapshot)
# Flush si buffer plein ou timeout
if len(self.buffer) >= self.batch_size:
await self._flush()
elif (datetime.utcnow() - self.last_flush).total_seconds() >= self.flush_interval:
await self._flush()
async def _flush(self):
"""Flush le buffer vers la base de données."""
if not self.buffer or not self.pool:
return
data_to_insert = self.buffer.copy()
self.buffer.clear()
self.last_flush = datetime.utcnow()
async with self.pool.acquire() as conn:
await conn.copy_records_to_table(
'order_book_snapshots',
records=[
(
r["time"], r["symbol"], r["bid_price"], r["bid_quantity"],
r["ask_price"], r["ask_quantity"], r["seq_id"],
r["spread"], r["mid_price"], r["imbalance"], r["raw_data"]
)
for r in data_to_insert
],
columns=[
'time', 'symbol', 'bid_price', 'bid_quantity',
'ask_price', 'ask_quantity', 'seq_id',
'spread', 'mid_price', 'imbalance', 'raw_data'
]
)
print(f"✓ Flush {len(data_to_insert)} snapshots vers PostgreSQL")
async def get_recent_snapshots(
self,
symbol: str,
seconds: int = 60,
limit: int = 100
) -> List[Dict]:
"""Récupère les snapshots récents pour analyse."""
async with self.pool.acquire() as conn:
rows = await conn.fetch("""
SELECT * FROM order_book_snapshots
WHERE symbol = $1
AND time > NOW() - INTERVAL '1 second' * $2
ORDER BY time DESC
LIMIT $3
""", symbol, seconds, limit)
return [dict(row) for row in rows]
async def calculate_metrics(
self,
symbol: str,
window_seconds: int = 300
) -> Dict:
"""Calcule les métriques agrégées sur une fenêtre."""
async with self.pool.acquire() as conn:
metrics = await conn.fetchrow("""
SELECT
COUNT(*) as snapshot_count,
AVG(mid_price) as avg_mid_price,
STDDEV(mid_price) as stddev_mid_price,
MIN(mid_price) as min_mid_price,
MAX(mid_price) as max_mid_price,
AVG(spread) as avg_spread,
AVG(imbalance) as avg_imbalance,
MIN(imbalance) as min_imbalance,
MAX(imbalance) as max_imbalance
FROM order_book_snapshots
WHERE symbol = $1
AND time > NOW() - INTERVAL '1 second' * $2
""", symbol, window_seconds)
return dict(metrics) if metrics else {}
async def detect_sequence_gaps(self, symbol: str) -> List[Dict]:
"""Détecte les gaps dans la séquence du order book."""
async with self.pool.acquire() as conn:
gaps = await conn.fetch("""
WITH seq_diff AS (
SELECT
seq_id,
LAG(seq_id) OVER (ORDER BY time) as prev_seq,
seq_id - LAG(seq_id) OVER (ORDER BY time) as diff
FROM order_book_snapshots
WHERE symbol = $1
AND time > NOW() - INTERVAL '1 hour'
)
SELECT * FROM seq_diff
WHERE diff > 1 OR diff < 0
ORDER BY seq_id DESC
LIMIT 100
""", symbol)
return [dict(row) for row in gaps]
async def close(self):
"""Ferme les connexions."""
if self.buffer:
await self._flush()
if self.pool:
await self.pool.close()
Intégration avec HolySheep AI pour l'analyse prédictive
class AIOrderBookAnalyzer:
"""
Utilise HolySheep AI pour analyser le order book
et détecter des patterns significatifs.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "deepseek-v3.2" # $0.42/MTok - optimal pour ce use case
async def analyze_imbalance_pattern(
self,
symbol: str,
recent_snapshots: List[Dict],
holy_sheep_api_key: str
) -> Dict:
"""
Analyse les patterns d'imbalance via l'API HolySheep.
Coût estimé: ~500 tokens par analyse = $0.00021
"""
import aiohttp
# Préparation du contexte
imbalances = [s.get("imbalance", 0) for s in recent_snapshots]
mid_prices = [float(s.get("mid_price", 0)) for s in recent_snapshots]
prompt = f"""
Analyse ce order book pour {symbol}:
Historique des imbalances (derniers {len(imbalances)} snapshots):
{imbalances[-20:]}
Prix mid (derniers prix):
{mid_prices[-10:]}
Questions:
1. Quel est le sentiment actuel du marché (bullish/bearish/neutral)?
2. Y a-t-il un pattern d'accumulation ou de distribution?
3. Quel est le risque de reversal à court terme?
Réponds en JSON avec: sentiment, pattern_type, reversal_risk (0-1), confidence (0-1)
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {holy_sheep_api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"response_format": {"type": "json_object"}
},
timeout=aiohttp.ClientTimeout(total=10)
) as response:
result = await response.json()
return {
"analysis": result.get("choices", [{}])[0].get("message", {}).get("content"),
"usage": result.get("usage", {}),
"cost": result["usage"]["total_tokens"] * 0.42 / 1_000_000
}
async def generate_trading_signals(
self,
symbol: str,
orderbook_metrics: Dict,
holy_sheep_api_key: str
) -> str:
"""Génère des signaux de trading basiques via HolySheep."""
import aiohttp
prompt = f"""
Génère un signal de trading simple pour {symbol}.
Métriques du order book:
- Volatilité (stddev mid price): {orderbook_metrics.get('stddev_mid_price', 0)}
- Imbalance moyen: {orderbook_metrics.get('avg_imbalance', 0)}
- Spread moyen: {orderbook_metrics.get('avg_spread', 0)}
Réponds uniquement avec:
- SIGNAL: BUY ou SELL ou NEUTRAL
- CONFIANCE: Haute/Moyenne/Basse
- RAISON: une phrase
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {holy_sheep_api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 100
}
) as response:
result = await response.json()
return result["choices"][0]["message"]["content"]
Pipeline complet d'utilisation
async def complete_pipeline():
"""
Pipeline complet: WebSocket → Parse → Stocke → Analyse IA
"""
# 1. Initialisation
ws_client = OKXWebSocketClient(sandbox=True)
storage = OrderBookStorage(batch_size=50, flush_interval=0.5)
analyzer = AIOrderBookAnalyzer(api_key="YOUR_HOLYSHEEP_KEY")
await storage.connect()
# 2. Callbacks
async def process_and_store(data):
await storage.insert_snapshot(data)
# Analyse toutes les 100 mises à jour
if len(storage.buffer) % 100 == 0:
snapshots = await storage.get_recent_snapshots(data["symbol"], seconds=60)
result = await analyzer.analyze_imbalance_pattern(
data["symbol"],
snapshots,
holy_sheep_api_key="YOUR_HOLYSHEEP_KEY"
)
print(f"Analyse IA: {result['analysis']}")
print(f"Coût: ${result['cost']:.6f}")
ws_client.register_callback(process_and_store)
# 3. Démarrage
await ws_client.connect()
await ws_client.subscribe_orderbook("BTC-USDT", depth=25)
try:
await ws_client.listen()
finally:
await storage.close()
if __name__ == "__main__":
asyncio.run(complete_pipeline())Erreurs Courantes et Solutions
Erreur 1 : Sequence ID Gap — Perte de Données
# Symptôme: Ordres qui semblent disparaître du order book
Erreur: "Sequence gap detected: expected 12345, got 12350"
❌ MAUVAIS: Ignorer les gaps
async def on_orderbook_update_legacy(data):
# Les données peuvent être incomplètes!
self.orderbook_data = data
✅ BON: Détection et resubscription
async def on_orderbook_update(self, data):
current_seq = data.get("seqId", 0)
previous_seq = self.last_seq_id.get(data["symbol"], 0)
if previous_seq > 0 and (current_seq - previous_seq) > 1:
print(f"⚠ GAP DÉTECTÉ: {data['symbol']}")
print(f" Dernier seq: {previous_seq} → Actuel: {current_seq}")
print(f" Gap size: {current_seq - previous_seq - 1} messages perdus")
# Option 1: Resubscribe pour obtenir le full snapshot
await self.resubscribe_full_snapshot(data["symbol"])
# Option 2: Demander le snapshot complet manuellement
await self.request_snapshot(data["symbol"])
self.last_seq_id[data["symbol"]] = current_seq
self.orderbook_data = data
async def resubscribe_full_snapshot(self, symbol: str):
"""Resubscribe avec snapshot complet pour resynchroniser."""
# Unsubscribe du canal actuel
await self.websocket.send(json.dumps({
"op": "unsubscribe",
"args": [{"channel": "books-l2-tbt", "instId": symbol}]
}))
await asyncio.sleep(0.1)
# Subscribe avec snapshot=true
await self.websocket.send(json.dumps({
"op": "subscribe",
"args": [{
"channel": "books",
"instId": symbol,
"snapshot": "true" # IMPORTANT: obtenir le full order book
}]
}))Erreur 2 : WebSocket Deconnexion Fréquente
# Symptôme: Connexions qui tombent toutes les 5-10 minutes
Erreur: "ConnectionClosed: close code 1006"
❌ CAUSES PROBABLES:
1. Pas de ping/pong heartbeat
2. Buffer trop petit pour les messages
3. Pas de gestion de reconnexion
✅ SOLUTION COMPLÈTE:
class RobustWebSocketClient:
def __init__(self):
self.heartbeat_interval = 20 # OKX requiert heartbeat < 30s
self.reconnect_base_delay = 1
self.max_reconnect_delay = 60
async def connect(self):
self.websocket = await websockets.connect(
self.url,
ping_interval=self.heartbeat_interval,
ping_timeout=10,
max_size=10 * 1024 * 1024, # 10MB pour order books profonds
close_timeout=5
)
# Démarrer heartbeat monitor
self._heartbeat_task = asyncio.create_task(self._heartbeat_monitor())
# Démarrer reconnect monitor
self._reconnect_task = asyncio.create_task(self._auto_reconnect())
async def _heartbeat_monitor(self):
"""Vérifie que les messages arrivent régulièrement."""
last_message = time.time()
while True:
await asyncio.sleep(5)
if time.time() - last_message > self.heartbeat_interval * 2:
print("⚠ Aucun message depuis longtemps, test ping...")
try:
await self.websocket.ping()
except:
pass # Sera géré par reconnect
last_message = time.time()
async def _auto_reconnect(self):
"""Reconnexion automatique avec backoff exponentiel."""
while True:
await asyncio.sleep(1)
if not self.websocket or self.websocket.closed:
delay = self.reconnect_base_delay
while delay <= self.max_reconnect_delay:
print(f"Reconnexion dans {delay}s...")
await asyncio.sleep(delay)
try:
await self.connect()
await self.resubscribe_all()
delay = self.reconnect_base_delay # Reset
print("✓ Reconnexion réussie")
break
except Exception as e:
print(f"✗ Échec: {e}")
delay = min(delay * 2, self.max_reconnect_delay)Erreur 3 : Dépassement de Mémoire avec Order Books Profonds
# Symptôme: Mémoire qui augmente continuellement
Erreur: MemoryError ou OOM killed
❌ MAUVAIS: Stocker tout l'historique en mémoire
class MemoryLeakClient:
def __init__(self):
self.all_orderbooks = [] # FUITTE MÉMOIRE!
async def on_update(self, data):
self.all_orderbooks.append(data) # Infiniment croissant
✅ BON: Fenêtre glissante avec limite de taille
class MemoryEfficientClient:
def __init__(self, max_depth: int = 400, max_history: int = 1000):
self.max_depth = max_depth
self.max_history = max_history
# Order book actuel (limité)
self.current_orderbook: Dict[str, List] = {"asks": [], "bids": []}
# Historique pour calculs (taille fixe)
from collections import deque
self.orderbook_history = deque(maxlen=self.max_history)
async def apply_snapshot(self, data: Dict):
"""Applique un snapshot complet (reset)."""
self.current_orderbook = {
"asks": [[float(p), float(q)] for p, q in data["asks"][:self.max_depth]],
"bids": [[float(p), float(q)] for p, q in data["bids"][:self.max_depth]]
}
# Gard seulement les changements pour l'historique
summary = {
"ts": data["ts"],
"best_bid": self.current_orderbook["bids"][0][0] if self.current_orderbook["bids"] else 0,
"best_ask": self.current_orderbook["asks"][0][0] if self.current_orderbook["asks"] else 0,
}
self.orderbook_history.append(summary)
async def apply_update(self, data: Dict):
"""Applique une mise à jour incrémentale."""
# Update asks: [price, qty, ...] où qty=0 = suppression
for update in data.get("asks", []):
price, qty = float(update[0]), float(update[1])
if qty == 0:
# Supprimer
self.current_orderbook["asks"] = [
[p, q] for p, q in self.current_orderbook["asks"]
if p != price
]
else:
# Update ou insert
found = False
for i, (p, q) in enumerate(self.current_orderbook["asks"]):
if p == price:
self.current_orderbook["asks"][i] = [price, qty]
found = True
break
if not found:
self.current_orderbook["asks"].append([price, qty])
# Tri et limite
self.current_orderbook["asks"].sort(key=lambda x: x[0])