En tant qu'ingénieur senior spécialisé dans les systèmes de trading haute fréquence, j'ai passé les cinq dernières années à optimiser des architectures d'order book pour des volumes dépassant les 100 000 transactions par seconde. Aujourd'hui, je partage mon retour d'expérience complet sur le calcul des indicateurs de liquidité en temps réel, avec du code production-ready et des benchmarks mesurés.
Comprendre l'Architecture d'un Order Book
Un order book est une structure de données ordonnée qui enregistre tous les ordres d'achat (bids) et de vente (asks) pour un actif donné. L'architecture classique utilise une structure en prix-densité où chaque niveau de prix contient le volume cumulé. La latence critique se situe entre la réception du marché et le calcul effectif des indicateurs.
Calcul du Spread
Le spread représente la différence entre le meilleur ask (prix de vente le plus bas) et le meilleur bid (prix d'achat le plus élevé). C'est l'indicateur fondamental de la liquidité immédiate.
import asyncio
import time
from dataclasses import dataclass
from typing import List, Tuple, Optional
from collections import defaultdict
import heapq
@dataclass
class Order:
price: float
quantity: float
side: str # 'bid' ou 'ask'
timestamp: int
class OrderBook:
def __init__(self, symbol: str):
self.symbol = symbol
self.bids: List[Tuple[float, float]] = [] # (prix, quantité)
self.asks: List[Tuple[float, float]] = []
self._lock = asyncio.Lock()
def add_order(self, order: Order):
"""Ajout synchrone d'un ordre - O(log n)"""
if order.side == 'bid':
heapq.heappush(self.bids, (-order.price, order.quantity))
else:
heapq.heappush(self.asks, (order.price, order.quantity))
async def calculate_spread(self) -> Optional[dict]:
"""Calcul du spread en millisecondes - latence mesurable"""
async with self._lock:
if not self.bids or not self.asks:
return None
start_ns = time.perf_counter_ns()
best_bid = -self.bids[0][0]
best_ask = self.asks[0][0]
absolute_spread = best_ask - best_bid
percentage_spread = (absolute_spread / best_ask) * 100
# Mid price pour référence
mid_price = (best_bid + best_ask) / 2
elapsed_ns = time.perf_counter_ns() - start_ns
return {
'symbol': self.symbol,
'best_bid': best_bid,
'best_ask': best_ask,
'absolute_spread': absolute_spread,
'percentage_spread_bps': percentage_spread * 100, # basis points
'mid_price': mid_price,
'calculation_latency_ns': elapsed_ns,
'calculation_latency_ms': elapsed_ns / 1_000_000
}
async def get_depth(self, levels: int = 10) -> dict:
"""Calcul du depth (profondeur) sur N niveaux"""
async with self._lock:
start_ns = time.perf_counter_ns()
# Bids en ordre décroissant
bid_depth = []
cumulative_bid = 0.0
for _, qty in sorted(self.bids, key=lambda x: -x[0])[:levels]:
cumulative_bid += qty
bid_depth.append({
'price': -self.bids[self.bids.index((_, qty))][0],
'quantity': qty,
'cumulative': cumulative_bid
})
# Asks en ordre croissant
ask_depth = []
cumulative_ask = 0.0
for price, qty in sorted(self.asks, key=lambda x: x[0])[:levels]:
cumulative_ask += qty
ask_depth.append({
'price': price,
'quantity': qty,
'cumulative': cumulative_ask
})
elapsed_ns = time.perf_counter_ns() - start_ns
return {
'bid_depth': bid_depth,
'ask_depth': ask_depth,
'total_bid_volume': cumulative_bid,
'total_ask_volume': cumulative_ask,
'imbalance_ratio': cumulative_bid / (cumulative_ask + 1e-10),
'depth_latency_ms': elapsed_ns / 1_000_000
}
Exemple d'utilisation
async def main():
book = OrderBook("BTC/USDT")
# Simulation d'ordre book avec données réalistes
orders = [
Order(42150.0, 2.5, 'bid'),
Order(42145.0, 1.8, 'bid'),
Order(42140.0, 3.2, 'bid'),
Order(42155.0, 2.0, 'ask'),
Order(42160.0, 1.5, 'ask'),
Order(42165.0, 4.0, 'ask'),
]
for order in orders:
book.add_order(order)
spread_data = await book.calculate_spread()
depth_data = await book.get_depth(levels=3)
print(f"Spread: {spread_data['absolute_spread']} USDT")
print(f"Latence calcul: {spread_data['calculation_latency_ms']:.3f} ms")
print(f"Imbalance: {depth_data['imbalance_ratio']:.4f}")
asyncio.run(main())Calcul du Depth (Profondeur de Marché)
Le depth mesure le volume cumulé disponible à différents niveaux de prix. Il est crucial pour évaluer la capacité d'absorption d'un ordre important sans impact excessif sur le prix.
import numpy as np
from typing import Dict, List
import pandas as pd
class MarketDepthAnalyzer:
"""Analyseur de profondeur avec métriques avancées"""
def __init__(self, tick_size: float = 0.01):
self.tick_size = tick_size
self.depth_cache = {}
self.cache_ttl_ms = 100
def compute_vwap_depth(
self,
bids: List[Tuple[float, float]],
asks: List[Tuple[float, float]],
levels: int = 50
) -> Dict:
"""
Volume Weighted Average Price sur les N premiers niveaux
Métrique clé pour l'exécution d'ordres volumineux
"""
# Construction des courbes de profondeur
bid_prices = [-p for p, q in bids[:levels]]
bid_volumes = [