Il y a trois mois, lors de l'intégration de trois exchanges (Binance, OKX et Bybit) dans notre système de trading algorithmique, j'ai rencontré une erreur qui m'a coûté six heures de debugging :
ValueError: Timestamp mismatch: Binance says 1708310400.000, OKX says 1708310402.000, diff=2s > threshold=1sCette erreur provenait d'un décalage horaire entre les serveurs des exchanges. Après des semaines de recherche, j'ai développé une architecture robuste pour synchroniser les données temporelles. Aujourd'hui, je partage avec vous ma solution complète et éprouvée en production.
Le problème fondamental : pourquoi les timestamps divergent
Chaque exchange utilise son propre serveur NTP avec des décalages variables. Les causes principales :
- Décalage horaire du serveur (pas toujours en UTC)
- Dérive de l'horloge interne ( Clock Skew )
- Latence réseau variable selon la région géographique
- Fuseaux horaires mal configurés dans les API
Architecture de synchronisation UTC
1. Serveur NTP centralisé
#!/usr/bin/env python3
"""
Synchroniseur UTC centralisé pour données multi-exchanges
Version: 2.1.0
"""
import asyncio
import time
from datetime import datetime, timezone
from typing import Dict, Optional
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ExchangeTimeOffset:
exchange: str
offset_ms: float
latency_ms: float
last_sync: datetime
confidence: float # 0.0 - 1.0
class UTCSyncronizer:
"""
Synchroniseur UTC multi-exchanges avec correction de dérive
Latence mesurée: <50ms avec cache intelligent
"""
# Serveurs NTP de référence
NTP_SERVERS = [
"pool.ntp.org",
"time.google.com",
"time.cloudflare.com"
]
def __init__(self, cache_duration: int = 300):
self.cache_duration = cache_duration
self._offset_cache: Dict[str, ExchangeTimeOffset] = {}
self._sync_count = 0
async def get_unified_timestamp(self, exchange: str, exchange_timestamp: int) -> float:
"""
Convertit un timestamp d'exchange en UTC canonique
Args:
exchange: Nom de l'exchange (binance, okx, bybit, etc.)
exchange_timestamp: Timestamp en millisecondes depuis l'exchange
Returns:
Timestamp UTC float (secondes avec décimales)
"""
offset = await self.get_time_offset(exchange)
# Correction du décalage + normalisation UTC
corrected = (exchange_timestamp / 1000) + (offset.offset_ms / 1000)
logger.debug(
f"[{exchange}] raw={exchange_timestamp} -> UTC={corrected:.3f} "
f"(offset={offset.offset_ms}ms, latency={offset.latency_ms}ms)"
)
return corrected
async def get_time_offset(self, exchange: str) -> ExchangeTimeOffset:
"""
Calcule le décalage horaire d'un exchange par rapport à UTC
Utilise un cache pour minimiser les appels réseau
"""
current_time = datetime.now(timezone.utc)
# Vérifier le cache
if exchange in self._offset_cache:
cached = self._offset_cache[exchange]
age = (current_time - cached.last_sync).total_seconds()
if age < self.cache_duration:
logger.debug(f"[{exchange}] Using cached offset: {cached.offset_ms}ms")
return cached
# Effectuer la synchronisation
offset = await self._sync_exchange_time(exchange, current_time)
self._offset_cache[exchange] = offset
self._sync_count += 1
return offset
async def _sync_exchange_time(self, exchange: str, utc_now: datetime) -> ExchangeTimeOffset:
"""
Synchronise avec le serveur de l'exchange via plusieurs méthodes
"""
exchange_endpoints = {
"binance": "https://api.binance.com/api/v3/time",
"okx": "https://www.okx.com/api/v5/market/time",
"bybit": "https://api.bybit.com/v5/market/time",
"kucoin": "https://api.kucoin.com/api/v1/time"
}
if exchange not in exchange_endpoints:
# Retourner un offset par défaut pour exchanges inconnus
return ExchangeTimeOffset(
exchange=exchange,
offset_ms=0.0,
latency_ms=0.0,
last_sync=utc_now,
confidence=0.5
)
start = time.perf_counter()
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(exchange_endpoints[exchange], timeout=5) as resp:
data = await resp.json()
round_trip = (time.perf_counter() - start) * 1000
# Extraire le timestamp selon le format de l'exchange
if exchange == "binance":
server_time = data["serverTime"]
elif exchange == "okx":
server_time = int(data["data"][0]["ts"])
elif exchange == "bybit":
server_time = int(data["time"]) if "time" in data else int(data["list"][0]["timeSec"])
elif exchange == "kucoin":
server_time = int(data["data"]["serverTime"]) * 1000
else:
server_time = 0
# Calculer le décalage (moitié du RTT pour approximation)
local_now = time.time() * 1000
estimated_server_now = local_now + (round_trip / 2)
offset_ms = server_time - estimated_server_now
return ExchangeTimeOffset(
exchange=exchange,
offset_ms=offset_ms,
latency_ms=round_trip,
last_sync=utc_now,
confidence=min(0.95, 1.0 - (round_trip / 2000))
)
except Exception as e:
logger.warning(f"[{exchange}] Sync failed: {e}, using cached/default")
# Fallback: utiliser le dernier offset connu ou 0
if exchange in self._offset_cache:
return self._offset_cache[exchange]
return ExchangeTimeOffset(
exchange=exchange,
offset_ms=0.0,
latency_ms=0.0,
last_sync=utc_now,
confidence=0.0
)
Exemple d'utilisation
async def main():
syncer = UTCSyncronizer(cache_duration=60)
# Symboles de test
test_timestamps = {
"binance": 1708310400000, # 17:00:00 UTC
"okx": 1708310402000, # +2 secondes
"bybit": 1708310398000 # -2 secondes
}
print("=== Test de synchronisation UTC ===\n")
for exchange, ts in test_timestamps.items():
unified = await syncer.get_unified_timestamp(exchange, ts)
dt = datetime.fromtimestamp(unified, tz=timezone.utc)
print(f"{exchange:10} -> {dt.isoformat()} (ts: {ts})")
print(f"\nTotal synchronisations: {syncer._sync_count}")
if __name__ == "__main__":
asyncio.run(main())2. Alignement intelligent des klines (candlesticks)
#!/usr/bin/env python3
"""
Aligneur de données multi-sources avec buffer glissant
Résout le problème des timestamps décalés entre exchanges
"""
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Tuple, Optional, Callable
import heapq
@dataclass
class Candle:
"""Représentation unifiée d'une bougie OHLCV"""
timestamp: float # UTC Unix
open: float
high: float
low: float
close: float
volume: float
exchange: str
original_timestamp: int # Timestamp original de l'exchange
@dataclass
class AlignedCandle:
"""Bougie alignée depuis plusieurs sources"""
timestamp: float # Timestamp UTC du slot
candles: Dict[str, Candle] = field(default_factory=dict)
@property
def sources_count(self) -> int:
return len(self.candles)
def get_average_price(self, price_type: str = "close") -> Optional[float]:
"""Calcule le prix moyen pondéré par le volume"""
prices = []
volumes = []
for ex, candle in self.candles.items():
price = getattr(candle, price_type)
if price and candle.volume > 0:
prices.append(price)
volumes.append(candle.volume)
if not prices:
return None
# Moyenne pondérée par le volume
total_volume = sum(volumes)
weighted_sum = sum(p * v for p, v in zip(prices, volumes))
return weighted_sum / total_volume
class MultiExchangeAligner:
"""
Aligneur temporel intelligent pour données multi-exchanges
Caractéristiques:
- Buffer glissant pour吸收 les décalages
- Tolérance configurable (défaut: ±500ms)
- Merge intelligent des données de plusieurs sources
"""
def __init__(self, interval_seconds: int = 60, tolerance_ms: int = 500):
self.interval = interval_seconds
self.tolerance = tolerance_ms / 1000 # Conversion en secondes
# Buffers par exchange et par symbole
self._buffers: Dict[str, Dict[str, List[Candle]]] = defaultdict(
lambda: defaultdict(list)
)
# Slots alignés en attente de validation
self._aligned_slots: Dict[str, List[Tuple[float, Dict[str, Candle]]]] = defaultdict(list)
# Callbacks pour données alignées
self._on_aligned: Optional[Callable] = None
def add_candle(self, exchange: str, symbol: str, candle: Candle) -> List[AlignedCandle]:
"""
Ajoute une bougie et retourne les bougies alignées disponibles
Args:
exchange: Nom de l'exchange
symbol: Symbole de trading (ex: BTC-USDT)
candle: Données de la bougie
Returns:
Liste des bougies alignées (peut être vide)
"""
buffer = self._buffers[exchange][symbol]
# Ajouter au buffer
buffer.append(candle)
# Maintenir une taille de buffer raisonnable
if len(buffer) > 100:
buffer.pop(0)
# Essayer d'aligner
return self._try_align(symbol)
def _try_align(self, symbol: str) -> List[AlignedCandle]:
"""Tente d'aligner les bougies en buffer"""
aligned = []
for exchange, buffer in self._buffers.items():
if symbol not in buffer:
continue
# Grouper par slot temporel
slots: Dict[int, List[Candle]] = defaultdict(list)
for candle in buffer:
# Calculer le slot UTC (arrondi à l'intervalle)
slot_ts = self._get_slot_timestamp(candle.timestamp)
slots[slot_ts].append(candle)
# Pour chaque slot avec plusieurs exchanges
for slot_ts, candles in slots.items():
if len(candles) >= 1: # Au moins une bougie
# Trouver les bougies des autres exchanges dans la tolérance
slot_candles: Dict[str, Candle] = {}
for ex, buf in self._buffers.items():
if symbol not in buf:
continue
for c in buf:
c_slot = self._get_slot_timestamp(c.timestamp)
if abs(c_slot - slot_ts) <= self.tolerance:
slot_candles[ex] = c
break
if len(slot_candles) > 1: # Données de plusieurs exchanges
aligned_candle = AlignedCandle(
timestamp=slot_ts,
candles=slot_candles
)
aligned.append(aligned_candle)
# Notifier si callback défini
if self._on_aligned:
self._on_aligned(symbol, aligned_candle)
return aligned
def _get_slot_timestamp(self, utc_ts: float) -> int:
"""Calcule le timestamp du slot pour un timestamp UTC"""
return int(utc_ts // self.interval * self.interval)
def get_aligned_data(
self,
symbol: str,
start_time: datetime,
end_time: datetime
) -> List[AlignedCandle]:
"""
Récupère les données alignées pour une période donnée
Args:
symbol: Symbole de trading
start_time: Début de la période (datetime UTC)
end_time: Fin de la période
Returns:
Liste des bougies alignées triées par timestamp
"""
result = []
start_ts = start_time.timestamp()
end_ts = end_time.timestamp()
for exchange, buffer in self._buffers.items():
if symbol not in buffer:
continue
for candle in buffer:
if start_ts <= candle.timestamp <= end_ts:
# Vérifier si d'autres exchanges ont des données dans la tolérance
slot_candles = {}
for ex, buf in self._buffers.items():
if symbol not in buf:
continue
for c in buf:
if abs(c.timestamp - candle.timestamp) <= self.tolerance:
slot_candles[ex] = c
break
if len(slot_candles) > 1:
aligned = AlignedCandle(
timestamp=self._get_slot_timestamp(candle.timestamp),
candles=slot_candles
)
result.append(aligned)
# Dédupliquer et trier
seen = set()
unique = []
for a in result:
key = (a.timestamp, tuple(sorted(a.candles.keys())))
if key not in seen:
seen.add(key)
unique.append(a)
return sorted(unique, key=lambda x: x.timestamp)
Intégration avec HolySheep AI pour analyse intelligente
async def analyze_with_holysheep(
aligned_data: List[AlignedCandle],
api_key: str
):
"""
Utilise HolySheep AI pour analyser les données alignées
HolySheep propose <50ms de latence et 85%+ d'économie vs OpenAI
Prix: DeepSeek V3.2 à $0.42/MTok vs GPT-4.1 à $8/MTok
"""
import aiohttp
base_url = "https://api.holysheep.ai/v1"
# Préparer les données pour l'analyse
summary = {
"total_candles": len(aligned_data),
"exchanges": list(set(
ex for candle in aligned_data
for ex in candle.candles.keys()
)),
"price_range": {
"min": min(
c.close for candle in aligned_data
for c in candle.candles.values()
),
"max": max(
c.close for candle in aligned_data
for c in candle.candles.values()
)
},
"volume_discrepancy_pct": []
}
# Calculer les écarts de volume entre exchanges
for aligned in aligned_data:
if aligned.sources_count >= 2:
prices = [c.volume for c in aligned.candles.values()]
avg = sum(prices) / len(prices)
max_diff = max(abs(p - avg) / avg * 100 for p in prices)
summary["volume_discrepancy_pct"].append(max_diff)
# Envoyer vers HolySheep AI pour analyse
prompt = f"""Analyse des données de trading multi-exchanges:
Données: {summary}
Identifie:
1. Les anomalies de prix entre exchanges
2. Les opportunités d'arbitrage
3. Les problèmes de qualité de données
4. Recommandations d'ajustement des offsets
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3
}
) as resp:
result = await resp.json()
return result.get("choices", [{}])[0].get("message", {}).get("content", "")
if __name__ == "__main__":
# Test de l'aligneur
aligner = MultiExchangeAligner(interval_seconds=60, tolerance_ms=500)
base_ts = datetime(2024, 2, 18, 17, 0, 0, tzinfo=timezone.utc).timestamp()
# Simuler des bougies de différents exchanges avec décalages
test_candles = [
Candle(base_ts, 42000, 42100, 41950, 42050, 100, "binance", 1708310400000),
Candle(base_ts + 0.5, 42010, 42120, 41960, 42060, 98, "okx", 1708310402500),
Candle(base_ts + 1.2, 42000, 42100, 41950, 42055, 102, "bybit", 1708310403200),
Candle(base_ts + 60, 42055, 42150, 42000, 42100, 110, "binance", 1708310460000),
Candle(base_ts + 60.3, 42060, 42160, 42010, 42110, 108, "okx", 1708310462300),
]
print("=== Test d'alignement ===\n")
for candle in test_candles:
aligned = aligner.add_candle(candle.exchange, "BTC-USDT", candle)
if aligned:
print(f"Nouvelles bougies alignées: {len(aligned)}")
for a in aligned:
print(f" Slot UTC: {datetime.fromtimestamp(a.timestamp, tz=timezone.utc).isoformat()}")
print(f" Sources: {list(a.candles.keys())}")
print(f" Prix moyen: ${a.get_average_price():.2f}\n")3. Intégration complète avec gestion des erreurs
#!/usr/bin/env python3
"""
Intégrateur complet multi-exchanges avec retry et fallbacks
Version production-ready avec gestion d'erreurs avancée
"""
import asyncio
import aiohttp
import logging
from datetime import datetime, timezone
from typing import Dict, List, Optional, Any
from enum import Enum
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ExchangeStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
DOWN = "down"
UNKNOWN = "unknown"
@dataclass
class ExchangeHealth:
name: str
status: ExchangeStatus
latency_ms: float
error_count: int
last_success: datetime
consecutive_failures: int = 0
class MultiExchangeIntegrator:
"""
Intégrateur multi-exchanges complet
Fonctionnalités:
- Health checks automatique
- Fallback intelligent entre exchanges
- Retry exponentiel avec jitter
- Rate limiting adaptatif
"""
EXCHANGE_APIS = {
"binance": {
"rest": "https://api.binance.com",
"klines": "/api/v3/klines",
"time": "/api/v3/time"
},
"okx": {
"rest": "https://www.okx.com",
"klines": "/api/v5/market/candles",
"time": "/api/v5/market/time"
},
"bybit": {
"rest": "https://api.bybit.com",
"klines": "/v5/market/kline",
"time": "/v5/market/time"
},
"kucoin": {
"rest": "https://api.kucoin.com",
"klines": "/api/v1/market/candles",
"time": "/api/v1/time"
}
}
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
timeout: float = 10.0
):
self.max_retries = max_retries
self.base_delay = base_delay
self.timeout = timeout
self._health: Dict[str, ExchangeHealth] = {}
self._rate_limits: Dict[str, Dict[str, Any]] = {}
# Initialiser le health tracking
for name in self.EXCHANGE_APIS:
self._health[name] = ExchangeHealth(
name=name,
status=ExchangeStatus.UNKNOWN,
latency_ms=0,
error_count=0,
last_success=datetime.min.replace(tzinfo=timezone.utc)
)
async def fetch_klines_with_fallback(
self,
symbol: str,
interval: str = "1m",
limit: int = 100,
preferred_exchange: Optional[str] = None
) -> Dict[str, List[Dict]]:
"""
Récupère les klines depuis plusieurs exchanges avec fallback
Args:
symbol: Symbole de trading (ex: BTCUSDT)
interval: Intervalle (1m, 5m, 1h, etc.)
limit: Nombre de bougies
preferred_exchange: Exchange préféré (si disponible)
Returns:
Dict mapping exchange -> liste de klines
"""
results = {}
errors = []
# Déterminer l'ordre de priorité
if preferred_exchange and preferred_exchange in self.EXCHANGE_APIS:
priority = [preferred_exchange] + [
ex for ex in self.EXCHANGE_APIS if ex != preferred_exchange
]
else:
priority = list(self.EXCHANGE_APIS.keys())
# Tenter chaque exchange
for exchange in priority:
try:
klines = await self._fetch_klines_safe(exchange, symbol, interval, limit)
if klines:
results[exchange] = klines
self._update_health(exchange, success=True)
else:
errors.append(f"{exchange}: returned empty data")
self._update_health(exchange, success=False)
except Exception as e:
errors.append(f"{exchange}: {type(e).__name__}: {str(e)}")
self._update_health(exchange, success=False)
logger.warning(f"[{exchange}] Failed to fetch klines: {e}")
if not results:
raise RuntimeError(
f"Aucun exchange disponible. Erreurs: {'; '.join(errors)}"
)
return results
async def _fetch_klines_safe(
self,
exchange: str,
symbol: str,
interval: str,
limit: int
) -> List[Dict]:
"""Fetch avec retry et gestion d'erreurs"""
# Vérifier le rate limit
if self._is_rate_limited(exchange):
logger.warning(f"[{exchange}] Rate limited, skipping")
return []
# Map intervalle selon l'exchange
interval_map = self._get_interval_mapping(exchange, interval)
for attempt in range(self.max_retries):
try:
start = datetime.now(timezone.utc)
if exchange == "binance":
data = await self._binance_klines(symbol, interval_map, limit)
elif exchange == "okx":
data = await self._okx_klines(symbol, interval_map, limit)
elif exchange == "bybit":
data = await self._bybit_klines(symbol, interval_map, limit)
elif exchange == "kucoin":
data = await self._kucoin_klines(symbol, interval_map, limit)
else:
return []
elapsed = (datetime.now(timezone.utc) - start).total_seconds() * 1000
logger.info(f"[{exchange}] Fetched {len(data)} klines in {elapsed:.0f}ms")
return data
except aiohttp.ClientResponseError as e:
if e.status == 429:
self._set_rate_limit(exchange)
logger.warning(f"[{exchange}] Rate limited (429)")
return []
elif e.status >= 500:
# Erreur serveur, retry
delay = self._get_retry_delay(attempt)
await asyncio.sleep(delay)
else:
# Erreur client, ne pas retry
raise
except asyncio.TimeoutError:
delay = self._get_retry_delay(attempt)
logger.warning(f"[{exchange}] Timeout, retry in {delay}s")
await asyncio.sleep(delay)
except Exception as e:
if attempt == self.max_retries - 1:
raise
delay = self._get_retry_delay(attempt)
await asyncio.sleep(delay)
return []
def _get_retry_delay(self, attempt: int) -> float:
"""Calcule le délai de retry avec exponential backoff et jitter"""
import random
base = self.base_delay * (2 ** attempt)
jitter = random.uniform(0, 0.5)
return min(base + jitter, 30.0) # Max 30 secondes
async def _binance_klines(self, symbol: str, interval: str, limit: int) -> List[Dict]:
"""Récupère les klines depuis Binance"""
base = self.EXCHANGE_APIS["binance"]["rest"]
url = f"{base}{self.EXCHANGE_APIS['binance']['klines']}"
params = {"symbol": symbol, "interval": interval, "limit": limit}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=self.timeout) as resp:
data = await resp.json()
if not isinstance(data, list):
raise ValueError(f"Unexpected Binance response: {data}")
return [{
"timestamp": int(k[0]),
"open": float(k[1]),
"high": float(k[2]),
"low": float(k[3]),
"close": float(k[4]),
"volume": float(k[5]),
"exchange": "binance"
} for k in data]
async def _okx_klines(self, symbol: str, interval: str, limit: int) -> List[Dict]:
"""Récupère les klines depuis OKX"""
base = self.EXCHANGE_APIS["okx"]["rest"]
url = f"{base}{self.EXCHANGE_APIS['okx']['klines']}"
params = {
"instId": symbol.replace("USDT", "-USDT"),
"bar": interval,
"limit": str(limit)
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=self.timeout) as resp:
data = await resp.json()
if data.get("code") != "0":
raise ValueError(f"OKX error: {data.get('msg')}")
candles = data.get("data", [])
# OKX retourne les données en ordre inverse
return [{
"timestamp": int(c[0]),
"open": float(c[1]),
"high": float(c[2]),
"low": float(c[3]),
"close": float(c[4]),
"volume": float(c[6]),
"exchange": "okx"
} for c in reversed(candles)]
async def _bybit_klines(self, symbol: str, interval: str, limit: int) -> List[Dict]:
"""Récupère les klines depuis Bybit"""
base = self.EXCHANGE_APIS["bybit"]["rest"]
url = f"{base}{self.EXCHANGE_APIS['bybit']['klines']}"
params = {
"category": "spot",
"symbol": symbol,
"interval": interval,
"limit": str(limit)
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=self.timeout) as resp:
data = await resp.json()
if data.get("retCode") != 0:
raise ValueError(f"Bybit error: {data.get('retMsg')}")
candles = data.get("result", {}).get("list", [])
# Bybit retourne les données en ordre inverse
return [{
"timestamp": int(c[0]),
"open": float(c[1]),
"high": float(c[2]),
"low": float(c[3]),
"close": float(c[4]),
"volume": float(c[5]),
"exchange": "bybit"
} for c in reversed(candles)]
async def _kucoin_klines(self, symbol: str, interval: str, limit: int) -> List[Dict]:
"""Récupère les klines depuis KuCoin"""
base = self.EXCHANGE_APIS["kucoin"]["rest"]
url = f"{base}{self.EXCHANGE_APIS['kucoin']['klines']}"
# KuCoin utilise un format d'intervalle différent
interval_map = {"1m": "1min", "5m": "5min", "1h": "1hour", "1d": "1day"}
kline_interval = interval_map.get(interval, interval)
params = {
"symbol": symbol,
"type": kline_interval
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=self.timeout) as resp:
data = await resp.json()
if data.get("code") != "200000":
raise ValueError(f"KuCoin error: {data.get('msg')}")
candles = data.get("data", [])
return [{
"timestamp": int(c[0]) * 1000, # KuCoin utilise des secondes
"open": float(c[1]),
"high": float(c[3]),
"low": float(c[4]),
"close": float(c[2]),
"volume": float(c[5]),
"exchange": "kucoin"
} for c in candles]
def _get_interval_mapping(self, exchange: str, interval: str) -> str:
"""Mappe l'intervalle standard vers le format de l'exchange"""
mappings = {
"binance": {"1m": "1m", "5m": "5m", "1h": "1h", "1d": "1d"},
"okx": {"1m": "1m", "5m": "5m", "1h": "1h", "1d": "1D"},
"bybit": {"1m": "1", "5m": "5", "1h": "60", "1d": "D"},
"kucoin": {"1m": "1min", "5m": "5min", "1h": "1hour", "1d": "1day"}
}
return mappings.get(exchange, {}).get(interval, interval)
def _is_rate_limited(self, exchange: str) -> bool:
if exchange not in self._rate_limits:
return False
import time
limit_data = self._rate_limits[exchange]
reset_time = limit_data.get("reset_at", 0)
if time.time() > reset_time:
del self._rate_limits[exchange]
return False
return True
def _set_rate_limit(self, exchange: str, retry_after: int = 60):
"""Configure le rate limit pour un exchange"""
import time
self._rate_limits[exchange] = {
"reset_at": time.time() + retry_after,
"limit": 1200 # Requêtes par minute (exemple)
}
def _update_health(self, exchange: str, success: bool):
"""Met à jour le statut de santé d'un exchange"""
health = self._health[exchange]
if success:
health.consecutive_failures = 0
health.status = ExchangeStatus.HEALTHY
health.last_success = datetime.now(timezone.utc)
else:
health.consecutive_failures += 1
health.error_count += 1
if health.consecutive_failures >= 3:
health.status = ExchangeStatus.DOWN
elif health.consecutive_failures >= 1:
health.status = ExchangeStatus.DEGRADED
def get_health_report(self) -> Dict[str, Dict]:
"""Génère un rapport de santé des exchanges"""
return {
name: {
"status": health.status.value,
"latency_ms": health.latency_ms