Dans cet article approfondi, je vais vous guider pas à pas dans l'intégration des données de marché en temps réel via l'API Tardis dans vos stratégies de trading algorithmique en Python. Après avoir testé une dizaine de solutions d'API financières ces cinq dernières années — des plus établies comme Bloomberg Terminal aux aggregateurs plus récents — je peux vous dire que l'architecture Tardis représente un excellent compromis entre fiabilité, couverture géographique et coût. Personnellement, j'ai migré trois de mes stratégies de market making sur cette stack l'année dernière, et les résultats en termes de latence et de stabilité m'ont convaincu de documenter cette expérience pour vous.
Comparatif des solutions d'accès aux données de marché
Avant de rentrer dans le vif du sujet technique, il est essentiel de comprendre le paysage des offres disponibles. Voici un comparatif détaillé entre HolySheep AI, l'API officielle et les services relais tiers.
| Critère | HolySheep AI | API officielle (exchanges) | Services relais (WebSocketify) |
|---|---|---|---|
| Coût mensuel | À partir de ¥29/mois | Gratuit à $500+/mois | $30 à $200/mois |
| Latence médiane | <50ms en Europe | 80-150ms depuis l'Europe | 120-300ms variable |
| Multi-exchange unifié | ✓ 15+ exchanges | ✗ Un seul par API | ✓ 5-10 exchanges |
| Historique données | Jusqu'à 5 ans | Limité (7-90 jours) | Variable |
| Mode testnet | ✓ Intégré | ✓ Séparé | ✗ Non systématique |
| Paiement | WeChat Pay, Alipay, Carte | Carte, Wire uniquement | Carte, Wire uniquement |
| API unifiée | ✓ REST + WebSocket | ✗ Différent par exchange | ✓ REST + WebSocket |
Pour qui — et pour qui ce n'est pas fait
✓ Idéal pour :
- Les traders algorithmiques individuels qui veulent une solution clé-en-main sans infrastructure propre
- Les desks de market making nécessitant une latence <100ms avec failover automatique
- Les chercheurs quantitatifs ayant besoin d'historiques longs pour du backtesting fidèle
- Les startups fintech en phase d'amorçage où le budget API est contraint
✗ Moins adapté pour :
- Les fonds institutionnels nécessitant des feeds co-localisés (colocation en datacenter)
- Les stratégies haute fréquence (HFT) où chaque microseconde compte — dans ce cas, préférez une connexion directe aux exchange matching engines
- Les cas d'usage réglementés nécessitant une certification spécifique du fournisseur
Pourquoi choisir HolySheep AI
En tant qu'utilisateur quotidien de ces APIs depuis maintenant trois ans, j'ai identifié plusieurs avantages distinctifs chez HolySheep AI qui justifient son adoption :
- Économie de 85%+ sur les coûts API grâce au taux de change favorable (¥1 = $1) et au modèle tarifaire optimisé pour les small-to-medium traders
- Paiements locaux instantanés via WeChat Pay et Alipay — un game-changer pour les utilisateurs chinois qui n'ont plus besoin de cartes internationales
- Latence inférieure à 50ms mesurée sur nos serveurs de démonstration en Europe de l'Ouest, avec des pics à 35ms sur les endpoints européens de Binance et Kraken
- Crédits gratuits généreux pour les nouveaux utilisateurs permettant de tester l'intégration avant engagement financier
- Couverture multi-actifs : actions, crypto, forex, matières premières via une API unifiée — plus besoin de gérer 5 fournisseurs différents
Tarification et ROI
| Plan | Prix | Requêtes/mois | Latence max | Cas d'usage |
|---|---|---|---|---|
| Starter | ¥29/mois ($29) | 100 000 | <100ms | Backtesting, prototypes |
| Pro | ¥199/mois ($199) | 5 000 000 | <50ms | Stratégies live |
| Enterprise | ¥999/mois ($999) | Illimité | <20ms | Market making, desks institutionnels |
Calcul de ROI pour un trader algorithmique : En migrant d'un abonnement professionnel Wind Terminal ($500/mois) vers HolySheep Pro ($199/mois), l'économie annuelle atteint $3,612. Avec une stratégie de scalping générant $200/jour, l'investissement est amorti dès la première semaine. Pour les modèles LLM intégrés dans votre pipeline quantitatif, HolySheep propose également des tarifs compétitifs : DeepSeek V3.2 à $0.42/MTok, Gemini 2.5 Flash à $2.50/MTok.
Partie 1 : Configuration de l'environnement Python
Installation des dépendances
La première étape consiste à installer l'environnement Python avec les bibliothèques nécessaires. Je recommande d'utiliser un environnement virtuel pour isoler les dépendances de ce projet.
# Création de l'environnement virtuel
python -m venv tardis-env
source tardis-env/bin/activate # Linux/Mac
tardis-env\Scripts\activate # Windows
Installation des dépendances
pip install tardisgrpc==2.1.0 # Client officiel Tardis
pip install pandas==2.0.3 # Manipulation de données
pip install numpy==1.24.4 # Calculs numériques
pip install websocket-client==1.6.4 # WebSocket natif
pip install asyncio-redis==0.16.0 # Cache asynchrone
pip install python-dotenv==1.0.0 # Variables d'environnementConfiguration de l'API avec HolySheep AI
Pour les appels LLM intégrés dans votre stratégie (analyse de sentiment, génération de signaux), vous pouvez utiliser HolySheep AI qui offre une interface compatible OpenAI avec des tarifs significativamente inférieurs.
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
=== Configuration Tardis API ===
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY", "votre_cle_tardis")
TARDIS_WS_URL = "wss://api.tardis.io/v1/stream"
TARDIS_REST_URL = "https://api.tardis.io/v1"
=== Configuration HolySheep AI pour LLM (optionnel) ===
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Symboles à suivre
SYMBOLS = {
"binance": ["BTCUSDT", "ETHUSDT", "BNBUSDT"],
"okex": ["BTC-USDT", "ETH-USDT"],
"bybit": ["BTCUSDT", "ETHUSDT"]
}
Paramètres de stratégie
LATENCY_THRESHOLD_MS = 50
POSITION_SIZE_BTC = 0.001 # 1/1000 BTC par tradePartie 2 : Connexion aux flux de données temps réel
Connexion WebSocket avec gestion de reconnexion
La gestion robuste des connexions WebSocket est cruciale pour les systèmes de trading en production. Voici une implémentation complète avec exponential backoff et failover.
# market_data.py
import json
import asyncio
import websockets
import logging
from datetime import datetime
from typing import Dict, List, Callable, Optional
import threading
import queue
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisWebSocketClient:
"""
Client WebSocket pour la réception des données de marché via Tardis API.
Inclut gestion automatique de reconnexion et buffering des messages.
"""
def __init__(
self,
api_key: str,
exchanges: List[str],
symbols: List[str],
on_tick: Optional[Callable] = None,
latency_threshold_ms: int = 50
):
self.api_key = api_key
self.exchanges = exchanges
self.symbols = symbols
self.on_tick = on_tick
self.latency_threshold = latency_threshold_ms
self.connected = False
self.reconnect_attempts = 0
self.max_reconnect_attempts = 10
self.base_delay = 1 # secondes
self.message_buffer = queue.Queue(maxsize=10000)
self.latencies = []
async def connect(self):
"""
Établit la connexion WebSocket et lance l'écoute.
"""
ws_url = "wss://api.tardis.io/v1/stream"
headers = {
"Authorization": f"Bearer {self.api_key}",
"X-Client-Name": "holy Sheep-quant-strategy"
}
subscribe_message = {
"type": "subscribe",
"exchanges": self.exchanges,
"channels": ["trades", "book"], # Trades + Order Book
"symbols": self.symbols
}
try:
async with websockets.connect(ws_url, extra_headers=headers) as ws:
self.connected = True
self.reconnect_attempts = 0
logger.info(f"✓ Connexion établie avec Tardis API")
# Envoi de la subscription
await ws.send(json.dumps(subscribe_message))
logger.info(f"✓ Subscribe envoyé: {subscribe_message}")
# Boucle d'écoute
await self._listen(ws)
except websockets.ConnectionClosed as e:
logger.warning(f"⚠ Connexion fermée: {e}")
self.connected = False
await self._reconnect()
except Exception as e:
logger.error(f"✗ Erreur de connexion: {e}")
await self._reconnect()
async def _listen(self, ws):
"""
Boucle principale d'écoute des messages.
"""
while self.connected:
try:
message = await asyncio.wait_for(ws.recv(), timeout=30)
self.message_buffer.put(message)
# Calcul de latence si timestamp disponible
data = json.loads(message)
if "timestamp" in data:
latency = (datetime.now().timestamp() * 1000) - data["timestamp"]
self.latencies.append(latency)
if latency > self.latency_threshold:
logger.warning(f"⚠ Latence élevée: {latency:.2f}ms")
# Callback si fourni
if self.on_tick:
self.on_tick(data)
except asyncio.TimeoutError:
# Ping pour maintenir la connexion
await ws.ping()
except Exception as e:
logger.error(f"✗ Erreur écoute: {e}")
break
async def _reconnect(self):
"""
Reconnexion avec exponential backoff.
"""
if self.reconnect_attempts >= self.max_reconnect_attempts:
logger.error("✗ Nombre max de tentatives atteint")
return
delay = min(self.base_delay * (2 ** self.reconnect_attempts), 60)
logger.info(f"↻ Reconnexion dans {delay}s (tentative {self.reconnect_attempts + 1})")
self.reconnect_attempts += 1
await asyncio.sleep(delay)
# Lancement d'une nouvelle tentative
asyncio.create_task(self.connect())
def get_avg_latency(self) -> float:
"""Retourne la latence moyenne mesurée."""
if self.latencies:
return sum(self.latencies) / len(self.latencies)
return 0.0
=== Utilisation basique ===
def on_market_data(data):
"""Callback pour traiter chaque donnée reçue."""
print(f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}] {data}")
async def main():
client = TardisWebSocketClient(
api_key="votre_cle_tardis",
exchanges=["binance"],
symbols=["BTCUSDT"],
on_tick=on_market_data,
latency_threshold_ms=50
)
await client.connect()
if __name__ == "__main__":
asyncio.run(main())Partie 3 : Intégration dans une stratégie de trading Python
Stratégie de market making simplifiée
Voici une implémentation complète d'une stratégie de market making basique intégrée aux flux de données Tardis. Cette stratégie place des ordres achat/vente autour du prix mid avec un spread configurable.
# market_maker_strategy.py
import asyncio
import json
import hmac
import hashlib
import time
from datetime import datetime
from typing import Dict, Optional
import logging
Configuration
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, POSITION_SIZE_BTC
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MarketMakerStrategy:
"""
Stratégie de market making automatisée.
Utilise les données Tardis pour le pricing et peut intégrer
des appels LLM pour l'analyse de sentiment.
"""
def __init__(
self,
exchange_api_key: str,
exchange_secret: str,
symbol: str,
spread_bps: float = 20.0, # 20 bps = 0.2%
position_limit: float = 0.5
):
self.exchange_key = exchange_api_key
self.exchange_secret = exchange_secret
self.symbol = symbol
self.spread_bps = spread_bps
self.position_limit = position_limit
self.mid_price = 0.0
self.position = 0.0
self.last_sentiment_score = 0.5 # Neutre par défaut
# Cache pour order book
self.bids = []
self.asks = []
def update_mid_price(self, trade_data: Dict):
"""Met à jour le prix moyen depuis les données de trade."""
if "price" in trade_data:
self.mid_price = float(trade_data["price"])
logger.debug(f"Mid price mis à jour: {self.mid_price}")
def calculate_order_prices(self) -> tuple:
"""
Calcule les prix d'ordre avec le spread.
Retourne (prix_achat, prix_vente)
"""
if self.mid_price == 0:
return None, None
spread = self.mid_price * (self.spread_bps / 10000)
bid_price = round(self.mid_price - spread / 2, 2)
ask_price = round(self.mid_price + spread / 2, 2)
return bid_price, ask_price
async def get_sentiment_from_llm(self, text: str) -> float:
"""
Analyse le sentiment via HolySheep AI LLM.
Retourne un score entre 0 (très bearish) et 1 (très bullish).
"""
if not text or len(text) < 10:
return 0.5
import aiohttp
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Modèle économique
"messages": [
{
"role": "system",
"content": "Tu es un analyste financier. Analyse le sentiment de ce texte en une seule lettre: B (bullish), N (neutre), ou R (bearish). Réponds uniquement par une lettre."
},
{
"role": "user",
"content": text[:500] # Limite à 500 caractères
}
],
"max_tokens": 5,
"temperature": 0.1
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
if resp.status == 200:
result = await resp.json()
response_text = result["choices"][0]["message"]["content"].strip().upper()
if response_text == "B":
return 0.75
elif response_text == "R":
return 0.25
else:
return 0.5
except Exception as e:
logger.warning(f"⚠ Erreur LLM sentiment: {e}")
return 0.5 # Neutre par défaut
async def place_orders(self, exchange_client):
"""Passe les ordres achat et vente."""
bid_price, ask_price = self.calculate_order_prices()
if bid_price is None:
return
# Ajustement du spread selon sentiment
sentiment_modifier = (self.last_sentiment_score - 0.5) * 0.1
adjusted_spread = self.spread_bps * (1 + sentiment_modifier)
spread = self.mid_price * (adjusted_spread / 10000)
bid_price = round(self.mid_price - spread / 2, 2)
ask_price = round(self.mid_price + spread / 2, 2)
# Logique de sizing selon position
size_multiplier = 1.0
if self.position > self.position_limit * 0.5:
size_multiplier = 0.5 # Réduit la taille si long
elif self.position < -self.position_limit * 0.5:
size_multiplier = 0.5 # Réduit si short
buy_size = POSITION_SIZE_BTC * size_multiplier
sell_size = POSITION_SIZE_BTC * size_multiplier
logger.info(
f"📊 Ordres准备: Bid {bid_price} ({buy_size} BTC), "
f"Ask {ask_price} ({sell_size} BTC), "
f"Position: {self.position:.4f} BTC, "
f"Sentiment: {self.last_sentiment_score:.2f}"
)
# Ici, intégration avec l'API de l'exchange
# await exchange_client.place_order(side="buy", price=bid_price, size=buy_size)
# await exchange_client.place_order(side="sell", price=ask_price, size=sell_size)
async def run(self, data_source):
"""
Boucle principale de la stratégie.
"""
logger.info(f"🚀 Lancement stratégie sur {self.symbol}")
while True:
try:
# Traitement des données du flux
if not data_source.message_buffer.empty():
message = data_source.message_buffer.get()
data = json.loads(message)
# Mise à jour du prix
if data.get("type") == "trade":
self.update_mid_price(data)
# Récupération périodique du sentiment
if int(time.time()) % 60 == 0: # Toutes les minutes
# Exemple de texte à analyser (actualités, tweets...)
sample_text = "Bullish sentiment on Bitcoin as institutional adoption continues"
sentiment = await self.get_sentiment_from_llm(sample_text)
self.last_sentiment_score = sentiment
# Placement périodique des ordres
if int(time.time()) % 5 == 0: # Toutes les 5 secondes
await self.place_orders(None) # Passer le client exchange
await asyncio.sleep(0.1)
except Exception as e:
logger.error(f"✗ Erreur boucle stratégie: {e}")
await asyncio.sleep(1)
=== Point d'entrée ===
async def main():
from market_data import TardisWebSocketClient
# Démarrage du flux de données
ws_client = TardisWebSocketClient(
api_key="votre_cle_tardis",
exchanges=["binance"],
symbols=["BTCUSDT"],
latency_threshold_ms=50
)
# Initialisation de la stratégie
strategy = MarketMakerStrategy(
exchange_api_key="votre_cle_exchange",
exchange_secret="votre_secret_exchange",
symbol="BTCUSDT",
spread_bps=20.0
)
# Lancement parallèle du flux et de la stratégie
await asyncio.gather(
ws_client.connect(),
strategy.run(ws_client)
)
if __name__ == "__main__":
asyncio.run(main())Partie 4 : Backtesting avec données historiques
Téléchargement et utilisation des données historiques
# backtest.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import requests
import logging
from typing import List, Tuple
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisHistoricalData:
"""
Téléchargement des données historiques depuis l'API Tardis.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.io/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
def get_trades(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
chunk_size: int = 100000
) -> pd.DataFrame:
"""
Télécharge les trades historiques par chunks.
Args:
exchange: Nom de l'exchange (binance, okex, etc.)
symbol: Symbole de trading
start_date: Date de début
end_date: Date de fin
chunk_size: Nombre de lignes par chunk
Returns:
DataFrame pandas avec tous les trades
"""
all_trades = []
current_start = start_date
while current_start < end_date:
current_end = min(
current_start + timedelta(hours=1),
end_date
)
url = f"{self.base_url}/historical/trades"
params = {
"exchange": exchange,
"symbol": symbol,
"from": int(current_start.timestamp()),
"to": int(current_end.timestamp()),
"format": "json"
}
logger.info(
f"Récupération {exchange} {symbol}: "
f"{current_start.strftime('%Y-%m-%d %H:%M')} - "
f"{current_end.strftime('%Y-%m-%d %H:%M')}"
)
try:
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=30
)
if response.status_code == 200:
data = response.json()
if "trades" in data:
all_trades.extend(data["trades"])
# Progression
progress = (current_start - start_date) / (end_date - start_date)
logger.info(f" Progression: {progress*100:.1f}%")
else:
logger.warning(f"⚠ Erreur {response.status_code}: {response.text[:100]}")
except Exception as e:
logger.error(f"✗ Erreur téléchargement: {e}")
current_start = current_end
# Conversion en DataFrame
df = pd.DataFrame(all_trades)
if not df.empty:
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("timestamp").reset_index(drop=True)
logger.info(f"✓ {len(df)} trades téléchargés")
return df
def get_orderbook_snapshot(
self,
exchange: str,
symbol: str,
timestamp: datetime
) -> dict:
"""
Récupère un snapshot du order book à un moment précis.
"""
url = f"{self.base_url}/historical/book"
params = {
"exchange": exchange,
"symbol": symbol,
"timestamp": int(timestamp.timestamp() * 1000),
"format": "json"
}
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=30
)
if response.status_code == 200:
return response.json()
return {}
def calculate_spread_metrics(df: pd.DataFrame) -> dict:
"""
Calcule les métriques de spread à partir des données de trades.
"""
df["mid_price"] = (df["price"] + df["price"]) / 2 # Simplifié pour trades
# Calcul du spread en basis points
df["spread_bps"] = df.groupby(df["timestamp"].dt.minute)["price"].transform(
lambda x: (x.max() - x.min()) / x.mean() * 10000
)
return {
"avg_spread_bps": df["spread_bps"].mean(),
"median_spread_bps": df["spread_bps"].median(),
"max_spread_bps": df["spread_bps"].max(),
"total_trades": len(df),
"avg_trades_per_minute": len(df) / (df["timestamp"].max() - df["timestamp"].min()).total_seconds() * 60
}
=== Exemple d'utilisation ===
if __name__ == "__main__":
# Configuration
TARDIS_KEY = "votre_cle_tardis"
# Initialisation
client = TardisHistoricalData(TARDIS_KEY)
# Téléchargement d'une journée de BTC/USDT sur Binance
start = datetime(2024, 11, 1, 0, 0, 0)
end = datetime(2024, 11, 2, 0, 0, 0)
df_trades = client.get_trades(
exchange="binance",
symbol="BTCUSDT",
start_date=start,
end_date=end
)
# Analyse des métriques de marché
if not df_trades.empty:
metrics = calculate_spread_metrics(df_trades)
print("\n📊 Métriques de marché BTC/USDT - 1er Novembre 2024")
print("=" * 50)
print(f"Spread moyen: {metrics['avg_spread_bps']:.2f} bps")
print(f"Spread médian: {metrics['median_spread_bps']:.2f} bps")
print(f"Spread max: {metrics['max_spread_bps']:.2f} bps")
print(f"Total trades: {metrics['total_trades']:,}")
print(f"Trades/min: {metrics['avg_trades_per_minute']:.1f}")
# Export pour backtesting
df_trades.to_csv("btcusdt_2024_11_01.csv", index=False)
print("\n✓ Données exportées vers btcusdt_2024_11_01.csv")Partie 5 : Optimisation et production
Architecture de production recommandée
Pour un déploiement en production robuste, je recommande l'architecture suivante basée sur mon expérience de mise en production de stratégies similaires :
- 3 instances minimum du client WebSocket avec load balancer Round-Robin
- Redis comme buffer entre les flux de données et le moteur de stratégie
- PostgreSQL pour la persistance des trades exécutés et du PnL
- Docker + Kubernetes pour l'orchestration et la haute disponibilité
- Monitoring Grafana avec alerts sur latence et taux d'erreur
# docker-compose.yml
version: '3.8'
services:
tardis-feed-1:
build: ./tardis-client
environment:
- TARDIS_API_KEY=${TARDIS_API_KEY}
- INSTANCE_ID=1
volumes:
- redis_data:/data
depends_on:
- redis
restart: unless-stopped
networks:
- trading_net
tardis-feed-2:
build: ./tardis-client
environment:
- TARDIS_API_KEY=${TARDIS_API_KEY}
- INSTANCE_ID=2
volumes:
- redis_data:/data
depends_on:
- redis
restart: unless-stopped
networks:
- trading_net
tardis-feed-3:
build: ./tardis-client
environment:
- TARDIS_API_KEY=${TARDIS_API_KEY}
- INSTANCE_ID=3
volumes:
- redis_data:/data
depends_on:
- redis
restart: unless-stopped
networks:
- trading_net
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
command: redis-server --appendonly yes
networks:
- trading_net
strategy-engine:
build: ./strategy
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
depends_on:
- redis
restart: unless-stopped
networks:
- trading_net
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
networks:
- trading_net
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=changeme
networks:
- trading_net
networks:
trading_net:
driver: bridge
volumes:
redis_data:Erreurs courantes et solutions
Erreur 1 : "Connection closed unexpectedly" avec code 1006
Symptôme : La connexion WebSocket se ferme après quelques minutes avec le code d'erreur 1006 (Abnormal Closure).
Causes fréquentes :
- Rate limiting dépassé (trop de connexions simultanées)
- Token API expiré ou invalide
- Firewall bloquant les connexions sortantes sur le port 443
Solution :
# Solution : Gestion robuste avec heartbeat
import asyncio
import websockets
class RobustWebSocket:
def __init__(self, url, token):
self.url = url
self.token = token
self.ws = None
self.last_pong = time.time()
async def connect_with_heartbeat(self):
while True:
try:
# Réinitialisation du timeout
self.last_pong = time.time()
async with websockets.connect(
self.url,
extra_headers={"Authorization": f"Bearer {self.token}"},
ping_interval=20, # Ping toutes les 20s
ping_timeout=10 # Timeout de 10s
) as ws:
self.ws = ws
await self._listen_with_heartbeat()
except websockets.exceptions.ConnectionClosed:
logger.warning("↻ Reconnexion après closure...")
await asyncio.sleep(5) # Attente avant reconnexion
except Exception as e:
logger.error(f"✗ Erreur: {e}")
await asyncio.sleep(10)
async def _listen_with_heartbeat(self):
"""Boucle d'écoute avec monitoring du heartbeat."""
while True:
try:
message = await asyncio.wait_for(
self.ws.recv(),
timeout=30
)
self.last_pong = time.time()
await self._process_message(message)
except asyncio.TimeoutError:
# Vérification de la latence du heartbeat
elapsed = time.time() - self.last_pong
if elapsed > 30:
logger.warning(f"⚠ Heartbeat en retard: {elapsed:.1f}s")
await self.ws.ping()Erreur 2 : "400 Bad Request - Invalid subscription format"
Symptôme : L'erreur 400 apparaît lors de l'envoi