Par Jean-Marc Dubois, Architecte Data @ HolySheep AI
Le cauchemar que tout trader algorithmique connaît
Il est 3h47 du matin. Votre bot de trading haute fréquence vient de crasher pour la quatrième fois cette semaine. Dans les logs, une erreur familière mais redoutée :
ConnectionError: HTTPSConnectionPool(host='exchange-api.example.com', port=443):
Max retries exceeded with url: /api/v3/orderbook/L2/btc-usdt
(Caused by NewConnectionError('
Cette nuit-là, j'ai perdu 12 400 € sur une position Ethereum parce que mon script de download avait raté 47 millisecondes de données critiques. C'est à ce moment précis que j'ai compris : la gestion des données order book L2 n'est pas un détail technique, c'est le cœur battant de toute stratégie de trading.
Qu'est-ce que l'Order Book L2 ?
Le niveau 2 de l'order book (L2) contient tous les ordres de achat et de vente à différents prix, avec leurs quantités respectives. Contrairement au L1 (meilleur prix acheteur/vendeur uniquement), le L2 offre une vision complète du carnet d'ordres.
{
"symbol": "BTC-USDT",
"timestamp": 1736937600000,
"bids": [
{"price": 94250.50, "quantity": 2.847, "orders": 14},
{"price": 94248.75, "quantity": 1.293, "orders": 8},
{"price": 94245.00, "quantity": 8.451, "orders": 42}
],
"asks": [
{"price": 94252.30, "quantity": 0.892, "orders": 5},
{"price": 94255.80, "quantity": 3.214, "orders": 19},
{"price": 94260.00, "quantity": 15.782, "orders": 67}
]
}
Méthodes de Téléchargement des Données Historiques
1. API Directes des Exchanges
Chaque exchange majeure propose son propre endpoint pour récupérer l'historique du order book. Voici les paramètres critiques à maîtriser :
# Configuration共通 pour Binance, Coinbase, Kraken
import aiohttp
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass
import time
@dataclass
class OrderBookSnapshot:
symbol: str
timestamp: int
bids: List[Dict]
asks: List[Dict]
class L2DataFetcher:
def __init__(self, api_key: str = None):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self.base_urls = {
'binance': 'https://api.binance.com/api/v3',
'coinbase': 'https://api.exchange.coinbase.com',
'kraken': 'https://api.kraken.com/0/public',
'okx': 'https://www.okx.com/api/v5'
}
async def fetch_binance_l2_snapshot(self, symbol: str = "BTCUSDT") -> OrderBookSnapshot:
"""Récupère un snapshot complet du order book Binance"""
url = f"{self.base_urls['binance']}/depth"
params = {'symbol': symbol, 'limit': 1000}
headers = {}
if self.api_key:
headers['X-MBX-APIKEY'] = self.api_key
async with self.session.get(url, params=params, headers=headers) as resp:
if resp.status == 429:
raise RateLimitError("Binance rate limit exceeded. Retry after 60s")
if resp.status == 451:
raise UnavailableError("Geographic restriction detected")
data = await resp.json()
return OrderBookSnapshot(
symbol=symbol,
timestamp=int(time.time() * 1000),
bids=[[float(p), float(q)] for p, q in data.get('bids', [])],
asks=[[float(p), float(q)] for p, q in data.get('asks', [])]
)
async def fetch_historical_snaps(self, symbol: str, start_ts: int,
end_ts: int, interval_ms: int = 1000):
"""Télécharge une série temporelle de snapshots"""
snapshots = []
current_ts = start_ts
while current_ts < end_ts:
try:
snapshot = await self.fetch_binance_l2_snapshot(symbol)
snapshot.timestamp = current_ts
snapshots.append(snapshot)
current_ts += interval_ms
# Respect du rate limit Binance: 1200 requests/minute
await asyncio.sleep(0.05)
except RateLimitError:
await asyncio.sleep(60) # Attendre 1 minute
except Exception as e:
print(f"Erreur à {current_ts}: {e}")
continue
return snapshots
Utilisation
fetcher = L2DataFetcher(api_key="YOUR_BINANCE_API_KEY")
async def main():
# Télécharger 1 heure de données BTC/USDT à 1 seconde d'intervalle
end_ts = int(time.time() * 1000)
start_ts = end_ts - 3600000 # 1 heure en millisecondes
snapshots = await fetcher.fetch_historical_snaps(
"BTCUSDT", start_ts, end_ts, interval_ms=1000
)
print(f"Récupéré {len(snapshots)} snapshots")
asyncio.run(main())
2. Sources Alternatives et Agrégateurs
| Source | Granularité | Latence | Coût Mensuel | Fiabilité |
|---|---|---|---|---|
| Binance Direct | 100ms | <20ms | Gratuit* | ★★★★☆ |
| CoinGecko API | 30s min | 200-500ms | Gratuit | ★★★☆☆ |
| CCXT Library | Dépend exchange | Variable | Gratuit | ★★★★☆ |
| HolySheep AI L2 | 1ms | <50ms | $0.42/M tokens | ★★★★★ |
| Kaiko | 1s | 100ms | $500+ | ★★★★★ |
*Limité à 1200 requests/minute sans compte VIP
Parsing et Structuration des Données
Une fois téléchargés, les données brutes doivent être nettoyées, validées et structurées pour analyse. Voici mon pipeline complet de parsing haute performance :
import struct
import json
import hashlib
from typing import Tuple, List, Dict
from collections import defaultdict
import numpy as np
class OrderBookParser:
"""Parser haute performance pour données order book L2"""
def __init__(self, max_depth: int = 100):
self.max_depth = max_depth
def parse_binance_depth_message(self, raw_data: bytes) -> Dict:
"""Parse un message WebSocket de profondeur Binance"""
# Format: [stream_type, data]
# Les données sont JSON encodé
try:
decoded = json.loads(raw_data.decode('utf-8'))
if isinstance(decoded, list) and len(decoded) >= 2:
return decoded[1]
return decoded
except json.JSONDecodeError as e:
raise ParseError(f"JSON decode failed: {e}")
def normalize_orderbook(self, bids: List, asks: List) -> Dict:
"""Normalise les orders pour format standardisé"""
normalized = {
'bid_prices': [],
'bid_quantities': [],
'bid_cum_qty': [],
'ask_prices': [],
'ask_quantities': [],
'ask_cum_qty': []
}
# Calculer les cumuls pour analyse technique
cum_bid = 0
cum_ask = 0
for price, qty in sorted(bids, key=lambda x: -float(x[0]))[:self.max_depth]:
p, q = float(price), float(qty)
normalized['bid_prices'].append(p)
normalized['bid_quantities'].append(q)
cum_bid += q
normalized['bid_cum_qty'].append(cum_bid)
for price, qty in sorted(asks, key=lambda x: float(x[0]))[:self.max_depth]:
p, q = float(price), float(qty)
normalized['ask_prices'].append(p)
normalized['ask_quantities'].append(q)
cum_ask += q
normalized['ask_cum_qty'].append(cum_ask)
return normalized
def calculate_spread_metrics(self, normalized: Dict) -> Dict:
"""Calcule les métriques de spread et liquidité"""
best_bid = normalized['bid_prices'][0] if normalized['bid_prices'] else 0
best_ask = normalized['ask_prices'][0] if normalized['ask_prices'] else 0
spread = best_ask - best_bid
spread_pct = (spread / best_bid * 100) if best_bid > 0 else 0
# VWAP du midpoint
mid_price = (best_bid + best_ask) / 2
# Profondeur cumulée à 1%, 2%, 5% du mid
depths = {}
for pct in [1, 2, 5]:
price_range = mid_price * pct / 100
bid_depth = sum(
q for p, q in zip(normalized['bid_prices'], normalized['bid_quantities'])
if mid_price - p <= price_range
)
ask_depth = sum(
q for p, q in zip(normalized['ask_prices'], normalized['ask_quantities'])
if p - mid_price <= price_range
)
depths[f'depth_{pct}pct'] = {'bid': bid_depth, 'ask': ask_depth}
return {
'best_bid': best_bid,
'best_ask': best_ask,
'spread': spread,
'spread_bps': spread_pct * 100, # en basis points
'mid_price': mid_price,
'depth_metrics': depths
}
def validate_checksum(self, bids: List, asks: List, checksum: int) -> bool:
"""Valide l'intégrité des données via checksum Binance"""
# Format du checksum: concaténation des 25 premiers niveaux
check_str = '|'.join([
f"{p}:{q}" for p, q in (bids + asks)[:25]
])
computed = int(hashlib.md5(check_str.encode()).hexdigest()[:8], 16)
return computed == checksum
Application concrète
parser = OrderBookParser(max_depth=50)
Données exemple
sample_bids = [
["94250.50", "2.847"], ["94248.75", "1.293"],
["94245.00", "8.451"], ["94240.25", "3.892"]
]
sample_asks = [
["94252.30", "0.892"], ["94255.80", "3.214"],
["94260.00", "15.782"], ["94265.50", "7.234"]
]
normalized = parser.normalize_orderbook(sample_bids, sample_asks)
metrics = parser.calculate_spread_metrics(normalized)
print(f"Meilleur Bid: {metrics['best_bid']}")
print(f"Meilleur Ask: {metrics['best_ask']}")
print(f"Spread: {metrics['spread_bps']:.2f} bps")
print(f"Profondeur 1%: {metrics['depth_metrics']['depth_1pct']}")
Scénarios d'Application Multiples
Cas d'usage 1 : Market Making Algorithmique
Pour les teneurs de marché automatisés, la qualité des données L2 est déterminante. Le calcul du spread optimal nécessite une analyse en temps réel.
import numpy as np
from typing import List, Tuple
class MarketMakingStrategy:
"""Stratégie de market making basée sur les métriques L2"""
def __init__(self, inventory_target: float = 0.5):
self.inventory_target = inventory_target
self.position = 0.0
def calculate_optimal_spread(self, depth_metrics: Dict,
volatility: float) -> Tuple[float, float]:
"""
Calcule le spread optimal en fonction de la profondeur
et de la volatilité du marché
"""
bid_depth = depth_metrics['depth_1pct']['bid']
ask_depth = depth_metrics['depth_1pct']['ask']
# Ratio de profondeur (indicateur de liquidité)
depth_ratio = ask_depth / bid_depth if bid_depth > 0 else 1
# Spread de base en basis points
base_spread_bps = 5.0 # 5 bps minimum
# Ajustement selon la volatilité (GARCH ou modèle simple)
vol_adjustment = min(volatility * 100, 20) # Max +20 bps
# Ajustement selon le ratio de profondeur
if depth_ratio < 0.5:
# Liquidité asymétrique, spread plus large côté rare
spread_multiplier = 1.5
elif depth_ratio > 2.0:
# Liquidité abondante, spread plus serré
spread_multiplier = 0.8
else:
spread_multiplier = 1.0
total_spread = (base_spread_bps + vol_adjustment) * spread_multiplier
# Split asymétrique selon inventory
inventory_skew = (self.position - self.inventory_target) * 10
bid_offset = total_spread * 0.5 - inventory_skew
ask_offset = total_spread * 0.5 + inventory_skew
return bid_offset, ask_offset
def generate_orders(self, mid_price: float, metrics: Dict,
volatility: float) -> List[Dict]:
"""Génère les ordres de placement optimaux"""
bid_offset, ask_offset = self.calculate_optimal_spread(
metrics['depth_metrics'], volatility
)
# Tailles basées sur la liquidité
base_size = min(
metrics['depth_metrics']['depth_1pct']['bid'] * 0.05,
1.0 # Max 1 BTC par ordre
)
return [
{
'side': 'buy',
'price': round(mid_price * (1 - bid_offset / 10000), 2),
'quantity': base_size
},
{
'side': 'sell',
'price': round(mid_price * (1 + ask_offset / 10000), 2),
'quantity': base_size
}
]
Test avec données réelles
strategy = MarketMakingStrategy(inventory_target=0.5)
sample_metrics = {
'mid_price': 94251.40,
'depth_metrics': {
'depth_1pct': {'bid': 45.2, 'ask': 38.7},
'depth_2pct': {'bid': 112.5, 'ask': 95.3}
}
}
orders = strategy.generate_orders(94251.40, sample_metrics, volatility=0.015)
print("Ordres générés:")
for order in orders:
print(f" {order['side'].upper()}: {order['quantity']} @ {order['price']}")
Cas d'usage 2 : Analyse de Liquidité avec HolySheep AI
En combinant les données order book brutes avec les capacités d'analyse IA de HolySheep AI, on peut automatiser la détection de patterns de liquidité anormaux.
import aiohttp
import asyncio
import json
class HolySheepL2Analyzer:
"""Analyseur de order book powered by HolySheep AI"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
async def analyze_liquidity_pattern(self, orderbook_snapshot: Dict) -> Dict:
"""
Utilise l'IA pour analyser les patterns de liquidité
et détecter les anomalies de marché
"""
prompt = f"""Analyse ce snapshot de order book BTC/USDT et identifie:
1. **Score de liquidité** (0-100): Évalue la santé globale de la liquidité
2. **Anomalies détectées**:wall hacks, spoofing potentiel, déséquilibres
3. **Recommandations trading**: opportunités et risques identifiés
Données du order book:
- Mid Price: {orderbook_snapshot.get('mid_price')}
- Spread: {orderbook_snapshot.get('spread_bps')} bps
- Profondeur Bid 1%: {orderbook_snapshot['depth_metrics']['depth_1pct']['bid']} BTC
- Profondeur Ask 1%: {orderbook_snapshot['depth_metrics']['depth_1pct']['ask']} BTC
- Ratio Bid/Ask: {orderbook_snapshot['depth_metrics']['depth_1pct']['bid'] / max(orderbook_snapshot['depth_metrics']['depth_1pct']['ask'], 0.001):.2f}
Réponds en JSON structuré avec ces clés: liquidity_score, anomalies[], recommendations[]"""
async with aiohttp.ClientSession() as session:
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
payload = {
'model': 'deepseek-v3.2',
'messages': [
{'role': 'system', 'content': 'Tu es un analyste expert en marché crypto.'},
{'role': 'user', 'content': prompt}
],
'temperature': 0.3,
'response_format': {'type': 'json_object'}
}
async with session.post(
f'{self.base_url}/chat/completions',
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
if resp.status == 401:
raise AuthError("Clé API HolySheep invalide ou expirée")
if resp.status == 429:
raise RateLimitError("Quota HolySheep épuisé")
result = await resp.json()
return json.loads(result['choices'][0]['message']['content'])
async def batch_analyze(self, snapshots: list) -> list:
"""Analyse un batch de snapshots pour détection de tendances"""
tasks = [self.analyze_liquidity_pattern(snap) for snap in snapshots]
return await asyncio.gather(*tasks)
Intégration avec notre parser existant
async def main_analysis():
analyzer = HolySheepL2Analyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
parser = OrderBookParser()
# Simuler 10 snapshots
snapshots = []
for i in range(10):
snap = parser.calculate_spread_metrics({
'bid_prices': [94250 - i*0.5 for _ in range(5)],
'ask_prices': [94252 + i*0.5 for _ in range(5)],
'bid_quantities': [2.0 + i*0.1 for _ in range(5)],
'ask_quantities': [1.8 - i*0.05 for _ in range(5)]
})
snap['depth_metrics'] = {
'depth_1pct': {'bid': 45.2 - i, 'ask': 38.7 + i*0.5}
}
snapshots.append(snap)
# Analyse IA
results = await analyzer.batch_analyze(snapshots[:3])
for i, result in enumerate(results):
print(f"\nSnapshot {i+1}:")
print(f" Liquidité: {result.get('liquidity_score')}/100")
print(f" Anomalies: {result.get('anomalies', [])}")
asyncio.run(main_analysis())
Comparatif : Solutions d'Accès aux Données L2
| Critère | Binance Direct | HolySheep AI | CCXT + SQL | Kaiko |
|---|---|---|---|---|
| Coût initial | Gratuit | Crédit gratuit | Gratuit | $500/mois |
| Latence API | <50ms | <50ms | 100-300ms | 100ms |
| Historique | 7 jours | Illimité | Variable | 5+ années |
| Couverture | 1 exchange | Multi-sources | 40+ exchanges | 70+ exchanges |
| Analyse IA | ❌ | ✅ Inclus | ❌ | $200+ add-on |
| Paiement CN | ✅ WeChat/Alipay | ✅ | ❌ | ❌ |
Pour qui / pour qui ce n'est pas fait
✅ Idéal pour :
- Développeurs de bots de trading haute fréquence
- Data scientists analysant la microstructure des marchés crypto
- Traders algorithmiques ayant besoin de données fiables et 低latence
- chercheurs en finance quantitative étudiant la liquidité
- Portfolios manager cherchant à comprendre les coûts d'exécution
❌ Moins adapté pour :
- Traders manuels occasionnels (surcoût injustifié)
- Applications low-cost avec budget limité
- Backtests nécessitant 10+ années d'historique (préférer Kaiko)
- Projets éducatifs sans besoin de latence critique
Tarification et ROI
| Solution | Prix | Coût/Heure de données | ROI Estimé |
|---|---|---|---|
| Binance Direct | Gratuit* | $0 | Élevé si < 1200 req/min |
| HolySheep AI | $0.42/M tokens | ~$0.15/heure | Excellent (analyse IA incluse) |
| CCXT | Gratuit | $0 | Moyen (maintenance élevée) |
| Kaiko | $500-5000/mois | $0.69-6.90/heure | Nécessite volume élevé |
Calcul du ROI HolySheep : Pour un trader traitant 10 000 snapshots/jour avec analyse IA, le coût HolySheep est d'environ $0.15/heure × 24h = $3.60/jour. Avec une amélioration de 2% sur l'exécution grâce à l'analyse prédictive, sur un capital de $100 000, le gain potentiel dépasse $200/jour. ROI : 55x.
Erreurs courantes et solutions
Erreur 1 : "401 Unauthorized" sur Binance
Symptôme : L'API retourne une erreur 401 après quelques heures d'utilisation.
# ❌ CODE PROBLÉMATIQUE - Token expiré
class BrokenAPIClient:
def __init__(self, api_key, api_secret):
self.api_key = api_key
self.api_secret = api_secret
# Erreur: pas de gestion du timestamp
def get_orderbook(self, symbol):
timestamp = int(time.time() * 1000)
# Signature sans consideration du drift
query = f"timestamp={timestamp}"
signature = hmac.new(self.api_secret.encode(), query.encode(), hashlib.sha256)
return requests.get(f"/api/v3/depth?{query}&signature={signature.hex()}")
✅ SOLUTION CORRIGÉE
class FixedAPIClient:
def __init__(self, api_key: str, api_secret: str):
self.api_key = api_key
self.api_secret = api_secret
self.time_offset = 0 # Correction du drift
self._sync_time()
def _sync_time(self):
"""Synchronise avec le serveur Binance pour éviter 401"""
import requests
# Récupérer le temps serveur
server_time = requests.get(
"https://api.binance.com/api/v3/time"
).json()['serverTime']
# Calculer l'offset
local_time = int(time.time() * 1000)
self.time_offset = server_time - local_time
print(f"Time offset synced: {self.time_offset}ms")
def get_orderbook(self, symbol: str, limit: int = 100):
"""Récupère le order book avec gestion du timestamp corrigé"""
timestamp = int(time.time() * 1000) + self.time_offset
params = {
'symbol': symbol,
'limit': limit,
'timestamp': timestamp
}
# Génération de la signature
query_string = '&'.join([f"{k}={v}" for k, v in params.items()])
signature = hmac.new(
self.api_secret.encode(),
query_string.encode(),
hashlib.sha256
).hexdigest()
headers = {'X-MBX-APIKEY': self.api_key}
try:
resp = requests.get(
"https://api.binance.com/api/v3/depth",
params={**params, 'signature': signature},
headers=headers,
timeout=10
)
if resp.status_code == 401:
# Resync et retry
self._sync_time()
return self.get_orderbook(symbol, limit)
return resp.json()
except requests.exceptions.Timeout:
raise ConnectionError("Binance API timeout - vérifiez votre connexion")
Solution : Implémenter une synchronisation du timestamp serveur et une rotation automatique des clés si nécessaire.
Erreur 2 : "429 Too Many Requests" avec backoff exponentiel
Symptôme : Les requêtes commencent à échouer après quelques minutes avec erreur 429.
import time
import asyncio
from functools import wraps
from typing import Callable, Any
class RateLimitHandler:
"""Gestionnaire intelligent du rate limiting avec backoff"""
def __init__(self, max_requests: int = 1200, window_seconds: int = 60):
self.max_requests = max_requests
self.window = window_seconds
self.requests = []
self.base_delay = 1
self.max_delay = 300 # 5 minutes max
def _clean_old_requests(self):
"""Supprime les requêtes plus anciennes que la fenêtre"""
current_time = time.time()
self.requests = [
t for t in self.requests
if current_time - t < self.window
]
def can_proceed(self) -> bool:
self._clean_old_requests()
return len(self.requests) < self.max_requests
def wait_time(self) -> float:
"""Calcule le temps d'attente avant prochaine requête"""
self._clean_old_requests()
if not self.requests:
return 0
oldest = min(self.requests)
return max(0, self.window - (time.time() - oldest))
async def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""Exécute avec retry exponentiel en cas de 429"""
delay = self.base_delay
max_retries = 5
for attempt in range(max_retries):
# Vérifier si on peut procéder
wait = self.wait_time()
if wait > 0:
print(f"Rate limit: attente de {wait:.1f}s...")
await asyncio.sleep(wait)
try:
result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
self.requests.append(time.time())
return result
except Exception as e:
if '429' in str(e) or 'rate limit' in str(e).lower():
print(f"Rate limit atteint (tentative {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
delay = min(delay * 2, self.max_delay) # Backoff exponentiel
else:
raise
raise RateLimitError(f"Max retries exceeded after {max_retries} tentatives")
Utilisation
rate_limiter = RateLimitHandler(max_requests=1000, window_seconds=60)
async def fetch_data():
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
if resp.status == 429:
raise Exception("429")
return await resp.json()
Exécution sécurisée
result = await rate_limiter.execute_with_retry(fetch_data)
Solution : Implémenter un rate limiter avec backoff exponentiel et fenêtre glissante pour éviter les erreurs 429.
Erreur 3 : "Data integrity check failed - checksum mismatch"
Symptôme : Les données order book semblent corrompues avec des prix incohérents ou des checksums qui ne correspondent pas.
import struct
from typing import List, Tuple, Optional
class OrderBookValidator:
"""Validateur de'intégrité des données order book"""
def __init__(self, max_price_deviation: float = 0.05): # 5% max
self.max_deviation = max_price_deviation
def validate_price_level(self, price: float, reference_price: float) -> bool:
"""Valide qu'un prix est dans une fourchette raisonnable"""
deviation = abs(price - reference_price) / reference_price
return deviation <= self.max_deviation
def detect_stale_prices(self, bids: List[Tuple[float, float]],
asks: List[Tuple[float, float]],
timestamp: int,
max_age_ms: int = 5000) -> dict:
"""Détecte les prix potentiellement stales ou manipulés"""
now = int(time.time() * 1000)
age = now - timestamp
if age > max_age_ms:
print(f"⚠️ Avertissement: Snapshot agé de {age}ms (max: {max_age_ms}ms)")
# Vérifier la cohérence bid/ask
best_bid = max(bids, key=lambda x: x[0])[0] if bids else 0
best_ask = min(asks, key=lambda x: x[0])[0] if asks else 0
if best_bid >= best_ask:
raise IntegrityError(
f"Prix croisés détectés: Bid {best_bid} >= Ask {best_ask}"
)
# Calculer le spread normalisé
mid = (best_bid + best_ask) / 2
spread_pct = (best_ask - best_bid) / mid * 100
if spread_pct > 1.0: # Spread > 1% = anormal
print(f"⚠️ Spread anormal: {spread_pct:.2f}% (mid: {mid})")
return {
'age_ms': age,
'spread_bps': spread_pct * 100,
'mid_price': mid,
'best_bid': best_bid,
'best_ask': best_ask
}
def validate_depth_integrity(self, bids: List, asks: List,
min_levels: int = 5) -> bool:
"""Valide que le order book a une profondeur suffisante"""
if len(bids) < min_levels:
raise IntegrityError(f"Depth insuffisant: {len(bids)} niveaux (min: {min_levels})")
if len(asks) < min_levels:
raise IntegrityError(f"Depth