En tant qu'ingénieur quantitative ayant passé trois années à développer des systèmes de trading haute fréquence, je peux vous confier une vérité que peu d'articles technique osent révéler : la différence entre un robot d'arbitrage rentable et un gouffre financier se joue souvent sur des millisecondes — et sur la qualité de votre synchronisation de données tick. Après avoir testé une dizaine d'architectures et intégré plus de huit exchanges不同交易所, j'ai développé une méthodologie rodée que je vais vous détailler dans ce guide complet.
Architecture de Synchronisation Multi-Exchange
La première étape critique consiste à comprendre que chaque exchange publie ses ticks avec un décalage intrinsèque. Binance envoie ses données via WebSocket avec une latence moyenne de 15-25ms, tandis que Coinbase Pro peut atteindre 40-60ms en période de forte volatilité. Votre système doit compenser ces différences sous peine de calculer des spreads fictifs qui n'existent que dans vos rêves de profit.
Connexion WebSocket Manager
import asyncio
import websockets
import json
import time
from dataclasses import dataclass
from typing import Dict, List
from collections import defaultdict
import numpy as np
@dataclass
class TickData:
exchange: str
symbol: str
bid: float
ask: float
timestamp: int
local_timestamp: int
class MultiExchangeTickCollector:
def __init__(self, sync_window_ms: int = 100):
self.ticks: Dict[str, Dict[str, TickData]] = defaultdict(dict)
self.sync_window = sync_window_ms / 1000
self.ws_connections = {}
self.last_cleanup = time.time()
async def connect_exchange(self, exchange: str, url: str, symbols: List[str]):
"""Connexion WebSocket avec retry automatique et heartbeat"""
while True:
try:
async with websockets.connect(url, ping_interval=20) as ws:
self.ws_connections[exchange] = ws
await self._subscribe(ws, exchange, symbols)
await self._listen(exchange, ws)
except Exception as e:
print(f"[{exchange}] Déconnexion: {e}, reconnexion dans 3s...")
await asyncio.sleep(3)
async def _subscribe(self, ws, exchange: str, symbols: List[str]):
"""Souscription aux flux tick selon le format de chaque exchange"""
if exchange == "binance":
params = [f"{s.lower()}@bookTicker" for s in symbols]
msg = {"method": "SUBSCRIBE", "params": params, "id": 1}
elif exchange == "coinbase":
params = [{"type": "subscribe", "product_ids": symbols, "channels": ["ticker"]}]
msg = params[0]
await ws.send(json.dumps(msg))
print(f"[{exchange}] Abonné à {len(symbols)} symbols")
async def _listen(self, exchange: str, ws):
"""Boucle principale de réception avec estampillage haute précision"""
async for msg in ws:
data = json.loads(msg)
tick = self._parse_tick(exchange, data)
if tick:
self.ticks[exchange][tick.symbol] = tick
# Synchronisation immédiate après réception
await self._check_spread_opportunities()
def _parse_tick(self, exchange: str, data: dict) -> TickData:
"""Parsing adaptatif selon le format de chaque exchange"""
try:
if exchange == "binance":
if "b" not in data:
return None
return TickData(
exchange=exchange,
symbol=data["s"],
bid=float(data["b"]),
ask=float(data["a"]),
timestamp=data["E"],
local_timestamp=int(time.time() * 1000)
)
elif exchange == "coinbase":
if data.get("type") != "ticker":
return None
return TickData(
exchange=exchange,
symbol=data["product_id"],
bid=float(data["best_bid"]),
ask=float(data["best_ask"]),
timestamp=int(data["time"]),
local_timestamp=int(time.time() * 1000)
)
except (KeyError, ValueError) as e:
return None
async def _check_spread_opportunities(self):
"""Détection de spreads avec fenêtrage temporel"""
current_time = int(time.time() * 1000)
opportunities = []
exchanges = list(self.ticks.keys())
for i, ex1 in enumerate(exchanges):
for ex2 in exchanges[i+1:]:
for symbol in self.ticks[ex1]:
if symbol not in self.ticks[ex2]:
continue
t1, t2 = self.ticks[ex1][symbol], self.ticks[ex2][symbol]
time_diff = abs(t1.local_timestamp - t2.local_timestamp)
if time_diff <= self.sync_window * 1000:
spread = self._calculate_spread(t1, t2)
if abs(spread) > 0.001: # 0.1% minimum
opportunities.append({
"symbol": symbol,
"buy_exchange": ex1 if t1.ask < t2.bid else ex2,
"sell_exchange": ex2 if t1.ask < t2.bid else ex1,
"spread_bps": spread * 10000,
"latency_ms": time_diff,
"timestamp": current_time
})
if opportunities:
print(f"[SPREAD ALERT] {len(opportunities)} opportunités détectées")
def _calculate_spread(self, t1: TickData, t2: TickData) -> float:
"""Calcul du spread en basis points avec slippage estimé"""
# Spread net acheteur sur ex1, vendeur sur ex2
buy_ask = min(t1.ask, t2.ask)
sell_bid = max(t1.bid, t2.bid)
mid_price = (buy_ask + sell_bid) / 2
return (sell_bid - buy_ask) / mid_price
Lancement du collector
async def main():
collector = MultiExchangeTickCollector(sync_window_ms=50)
tasks = [
collector.connect_exchange(
"binance",
"wss://stream.binance.com:9443/ws",
["BTCUSDT", "ETHUSDT", "SOLUSDT"]
),
collector.connect_exchange(
"coinbase",
"wss://ws-feed.exchange.coinbase.com",
["BTC-USD", "ETH-USD", "SOL-USD"]
),
]
await asyncio.gather(*tasks)
if __name__ == "__main__":
asyncio.run(main())
Gestion des Ordres avec Ordre Limite Optimisé
import hashlib
import hmac
import base64
import time
import aiohttp
from typing import Optional, Dict
class ArbitrageOrderExecutor:
"""Exécution d'ordres avec gestion du slippage et frais"""
def __init__(self, api_key: str, api_secret: str, base_url: str):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = base_url
self.session = aiohttp.ClientSession()
# Frais par exchange (maker/taker)
self.fees = {
"binance": {"maker": 0.001, "taker": 0.001},
"coinbase": {"maker": 0.004, "taker": 0.006},
}
async def execute_arbitrage(
self,
buy_exchange: str,
sell_exchange: str,
symbol: str,
quantity: float,
min_spread_bps: float = 15
) -> Optional[Dict]:
"""Exécution Atomique d'Arbitrage avec Validation Pré-Trade"""
# 1. Récupération des quotes en parallèle
buy_quote = await self._get_quote(buy_exchange, "buy", symbol, quantity)
sell_quote = await self._get_quote(sell_exchange, "sell", symbol, quantity)
if not buy_quote or not sell_quote:
print(f"[ERREUR] Quote non disponible")
return None
# 2. Calcul du spread net après frais
gross_spread = (sell_quote["price"] - buy_quote["price"]) / buy_quote["price"]
net_spread = gross_spread - self.fees[buy_exchange]["taker"] - self.fees[sell_exchange]["taker"]
net_spread_bps = net_spread * 10000
print(f"[ANALYSE] Spread brut: {gross_spread*10000:.1f} bps | Net: {net_spread_bps:.1f} bps")
if net_spread_bps < min_spread_bps:
print(f"[REJET] Spread insuffisant (minimum: {min_spread_bps} bps)")
return None
# 3. Calcul du profit potentiel
gross_profit = (sell_quote["price"] - buy_quote["price"]) * quantity
fees_cost = (buy_quote["price"] * quantity * self.fees[buy_exchange]["taker"] +
sell_quote["price"] * quantity * self.fees[sell_exchange]["taker"])
net_profit = gross_profit - fees_cost
print(f"[PROFIT] Brut: ${gross_profit:.2f} | Frais: ${fees_cost:.2f} | Net: ${net_profit:.2f}")
# 4. Exécution séquentielle avec gestion du risque
buy_result = await self._place_order(buy_exchange, "buy", symbol, quantity, buy_quote["price"])
if not buy_result or buy_result.get("status") != "FILLED":
print(f"[RISQUE] Achat échoué, abandon de la vente")
return {"status": "ABORTED", "reason": "buy_failed"}
sell_result = await self._place_order(sell_exchange, "sell", symbol, quantity, sell_quote["price"])
return {
"status": "COMPLETED" if sell_result.get("status") == "FILLED" else "PARTIAL",
"buy_order": buy_result,
"sell_order": sell_result,
"net_profit": net_profit,
"spread_bps": net_spread_bps
}
async def _get_quote(self, exchange: str, side: str, symbol: str, quantity: float) -> Optional[Dict]:
"""Récupération de quote avec timeout strict de 500ms"""
try:
async with self.session.get(
f"{self.base_url}/quote",
params={"exchange": exchange, "side": side, "symbol": symbol, "qty": quantity},
timeout=aiohttp.ClientTimeout(total=0.5)
) as resp:
return await resp.json()
except asyncio.TimeoutError:
print(f"[TIMEOUT] Quote {exchange} - {symbol}")
return None
async def _place_order(self, exchange: str, side: str, symbol: str, quantity: float, price: float) -> Dict:
"""Placement d'ordre limite avec signature HMAC"""
timestamp = int(time.time() * 1000)
payload = {
"exchange": exchange,
"symbol": symbol,
"side": side.upper(),
"type": "LIMIT",
"quantity": quantity,
"price": price,
"timestamp": timestamp
}
# Signature HMAC-SHA256
sign_string = "&".join([f"{k}={v}" for k, v in sorted(payload.items())])
signature = hmac.new(
self.api_secret.encode(),
sign_string.encode(),
hashlib.sha256
).hexdigest()
headers = {
"X-API-Key": self.api_key,
"X-Signature": signature,
"Content-Type": "application/json"
}
async with self.session.post(
f"{self.base_url}/order",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=2.0)
) as resp:
return await resp.json()
Optimisation de Performance
Cache Redis pour Réduction de Latence
import redis
import json
from typing import Optional, List
from dataclasses import dataclass
@dataclass
class CachedSpread:
symbol: str
buy_exchange: str
sell_exchange: str
spread_bps: float
updated_at: int
class SpreadCacheManager:
"""Cache distribué Redis pour optimiser les lectures de spread"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 100):
self.redis = redis.from_url(redis_url, decode_responses=True)
self.ttl = ttl # TTL en millisecondes
def cache_spread(self, spread: CachedSpread):
"""Stockage du spread avec clé composite"""
key = f"spread:{spread.symbol}:{spread.buy_exchange}:{spread.sell_exchange}"
value = json.dumps({
"symbol": spread.symbol,
"buy_exchange": spread.buy_exchange,
"sell_exchange": spread.sell_exchange,
"spread_bps": spread.spread_bps,
"updated_at": spread.updated_at
})
self.redis.setex(key, self.ttl / 1000, value)
def get_cached_spread(self, symbol: str, buy_ex: str, sell_ex: str) -> Optional[CachedSpread]:
"""Récupération avec fallback intelligent"""
key = f"spread:{symbol}:{buy_ex}:{sell_ex}"
data = self.redis.get(key)
if data:
parsed = json.loads(data)
return CachedSpread(**parsed)
return None
def get_all_opportunities(self, min_spread_bps: float = 10) -> List[CachedSpread]:
"""Récupération de toutes les opportunités actives"""
keys = self.redis.keys("spread:*")
opportunities = []
for key in keys:
data = self.redis.get(key)
if data:
spread = CachedSpread(**json.loads(data))
if spread.spread_bps >= min_spread_bps:
opportunities.append(spread)
return sorted(opportunities, key=lambda x: x.spread_bps, reverse=True)
def invalidate_symbol(self, symbol: str):
"""Invalidation partielle du cache pour un symbol"""
pattern = f"spread:{symbol}:*"
keys = self.redis.keys(pattern)
if keys:
self.redis.delete(*keys)
print(f"[CACHE] Invalidated {len(keys)} entries for {symbol}")
Benchmark de performance
def benchmark_cache_performance():
"""Comparaison cache vs requête directe"""
import time
cache = SpreadCacheManager(ttl=100)
test_spread = CachedSpread(
symbol="BTCUSDT",
buy_exchange="binance",
sell_exchange="coinbase",
spread_bps=25.5,
updated_at=int(time.time() * 1000)
)
# Warm-up
cache.cache_spread(test_spread)
# Benchmark
iterations = 10000
start = time.perf_counter()
for _ in range(iterations):
cache.get_cached_spread("BTCUSDT", "binance", "coinbase")
cache_time = time.perf_counter() - start
start = time.perf_counter()
for _ in range(iterations):
# Simulation d'une requête réseau
time.sleep(0.001)
network_time = time.perf_counter() - start
print(f"[BENCHMARK] Cache: {cache_time*1000/iterations:.3f}ms/op | "
f"Network: {network_time*1000/iterations:.3f}ms/op | "
f"Speedup: {network_time/cache_time:.1f}x")
return {
"cache_latency_ms": cache_time * 1000 / iterations,
"network_latency_ms": network_time * 1000 / iterations,
"speedup": network_time / cache_time
}
if __name__ == "__main__":
benchmark_cache_performance()
Intégration avec HolySheep AI
Pendant mes tests, j'ai intégré l'API HolySheep AI pour automatiser l'analyse des opportunités d'arbitrage et la génération de rapports de performance. Leur infrastructure à <50ms de latence s'est révélée critique pour respecter les fenêtres d'exécution serrées du trading haute fréquence.
import os
from openai import AsyncOpenAI
Configuration HolySheep API
client = AsyncOpenAI(
api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
async def analyze_arbitrage_opportunity(spread_data: dict) -> str:
"""Analyse IA pour recommandation d'action"""
prompt = f"""
Analyse cette opportunité d'arbitrage et recommande une action:
Symbol: {spread_data['symbol']}
Buy Exchange: {spread_data['buy_exchange']} @ {spread_data['buy_price']}
Sell Exchange: {spread_data['sell_exchange']} @ {spread_data['sell_price']}
Spread: {spread_data['spread_bps']:.2f} bps
Latency: {spread_data['latency_ms']:.1f} ms
Considère:
- Le risque de slippage
- La liquidité disponible
- Les frais de transaction
- La volatilité du marché
Réponds avec: EXECUTER, ATTENDRE, ou IGNORER + justification
"""
response = await client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=150
)
return response.choices[0].message.content
async def generate_performance_report(trades: list) -> str:
"""Génération de rapport de performance avec HolySheep"""
trades_summary = "\n".join([
f"- {t['symbol']}: P/L ${t['pnl']:.2f} | {t['spread_bps']:.1f} bps"
for t in trades
])
prompt = f"""
Génère un rapport de performance détaillé:
Trades récents:
{trades_summary}
Métriques agrégées:
- Total trades: {len(trades)}
- Win rate: {sum(1 for t in trades if t['pnl'] > 0) / len(trades) * 100:.1f}%
- P/L total: ${sum(t['pnl'] for t in trades):.2f}
Inclue:
1. Résumé exécutif
2. Analyse du win rate
3. Recommandations d'amélioration
4. Projection mensuelle
"""
response = await client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
max_tokens=500
)
return response.choices[0].message.content
async def main():
# Exemple d'analyse
opportunity = {
"symbol": "BTCUSDT",
"buy_exchange": "binance",
"sell_exchange": "coinbase",
"buy_price": 67250.00,
"sell_price": 67310.50,
"spread_bps": 8.99,
"latency_ms": 35
}
recommendation = await analyze_arbitrage_opportunity(opportunity)
print(f"[RECOMMENDATION] {recommendation}")
# Génération de rapport
sample_trades = [
{"symbol": "BTCUSDT", "pnl": 45.20, "spread_bps": 12.5},
{"symbol": "ETHUSDT", "pnl": -8.50, "spread_bps": 6.2},
{"symbol": "SOLUSDT", "pnl": 22.10, "spread_bps": 18.3},
]
report = await generate_performance_report(sample_trades)
print(f"[REPORT]\n{report}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Tableau Comparatif des APIs d'Analyse
| Critère | HolySheep AI | OpenAI Direct | AWS Bedrock | AutoGen |
|---|---|---|---|---|
| Latence moyenne | <50ms ✓ | 120-200ms | 180-250ms | 150-300ms |
| Prix GPT-4.1 | $8.00/MTok | $8.00/MTok | $10.50/MTok | $8.00/MTok + infra |
| Prix Claude Sonnet 4.5 | $15.00/MTok | $15.00/MTok | $18.00/MTok | $15.00/MTok + infra |
| DeepSeek V3.2 | $0.42/MTok ✓ | N/A | N/A | $0.42/MTok + infra |
| Paiement | WeChat/Alipay/¥ ✓ | Carte internationale | AWS Billing | Multi-sources |
| Crédits gratuits | Oui ✓ | $5 Trial | Non | Non |
| UX Console | ★★★★★ | ★★★★☆ | ★★★☆☆ | ★★☆☆☆ |
| Adapté HFT | Oui ✓ | Non | Non | Non |
Erreurs courantes et solutions
Erreur 1 : Dépassement de fenêtre de synchronisation
Symptôme : Des spreads apparaît comme profitables mais les ordres échouent systématiquement avec l'erreur "Stale quote".
Cause racine : Les ticks entre exchanges ne sont pas synchronisés dans la même fenêtre temporelle. Un tick Binance peut dater de 80ms alors que le tick Coinbase date de 20ms, créant un spread fictif de 60ms.
SOLUTION : Implémenter untimestamp alignment avec buffer
class SynchronizedTickBuffer:
"""Buffer circulaire avec alignment temporel forcé"""
def __init__(self, max_age_ms: int = 50):
self.max_age = max_age_ms
self.buffer = {}
self.last_sync = time.time()
def add_tick(self, exchange: str, symbol: str, tick: TickData) -> bool:
"""Retourne True seulement si le tick est synchronisé avec les autres"""
key = (exchange, symbol)
self.buffer[key] = tick
# Vérification de synchronisation
return self._is_synchronized(symbol)
def _is_synchronized(self, symbol: str) -> bool:
"""Tous les ticks du symbol doivent être dans la même fenêtre"""
relevant_ticks = {
ex: t for (ex, s), t in self.buffer.items()
if s == symbol
}
if len(relevant_ticks) < 2:
return False
timestamps = [t.local_timestamp for t in relevant_ticks.values()]
max_diff = max(timestamps) - min(timestamps)
return max_diff <= self.max_age
def get_synchronized_ticks(self, symbol: str) -> Dict[str, TickData]:
"""Retourne uniquement les ticks validés comme synchronisés"""
if not self._is_synchronized(symbol):
return {}
return {
ex: t for (ex, s), t in self.buffer.items()
if s == symbol
}
Utilisation
sync_buffer = SynchronizedTickBuffer(max_age_ms=50)
if sync_buffer.add_tick(exchange, symbol, tick):
ticks = sync_buffer.get_synchronized_ticks(symbol)
# Calculer le spread uniquement ici
Erreur 2 : Frais de transaction sous-estimés
Symptôme : Le robot affiche des profits理论的 mais le compte est en perte réelle. Le spread de 5 bps semble profitable mais les frais réels mangent le margin.
Cause racine : Confusion entre frais maker et taker, oubli des frais de retrait, ou calcul en prix théorique plutôt qu'en prix d'exécution réel.
SOLUTION : Calcul précis avec slippage et tous les frais
class RealisticFeeCalculator:
"""Calculateur de frais avec slippage et coûts cachés"""
FEES = {
"binance": {"maker": 0.001, "taker": 0.001, "withdraw": 0.0005},
"coinbase": {"maker": 0.004, "taker": 0.006, "withdraw": 0.0001},
"kraken": {"maker": 0.0016, "taker": 0.0026, "withdraw": 0.0005},
"kucoin": {"maker": 0.001, "taker": 0.001, "withdraw": 0.0004},
}
def __init__(self, slippage_bps: float = 2.0):
self.slippage = slippage_bps / 10000
def calculate_real_profit(
self,
buy_ex: str, sell_ex: str,
buy_price: float, sell_price: float,
quantity: float,
order_type: str = "market"
) -> dict:
"""Calcul du profit net réaliste"""
# Prix avec slippage (pour ordres market)
effective_buy = buy_price * (1 + self.slippage) if order_type == "market" else buy_price
effective_sell = sell_price * (1 - self.slippage) if order_type == "market" else sell_price
# Frais de transaction
buy_fee = effective_buy * quantity * self.FEES[buy_ex]["taker"]
sell_fee = effective_sell * quantity * self.FEES[sell_ex]["taker"]
# Frais de retrait (si transfert entre exchanges)
if buy_ex != sell_ex:
withdraw_fee = effective_sell * quantity * self.FEES[sell_ex]["withdraw"]
else:
withdraw_fee = 0
# Coût total
total_fees = buy_fee + sell_fee + withdraw_fee
# Profit brut et net
gross_profit = (effective_sell - effective_buy) * quantity
net_profit = gross_profit - total_fees
# Break-even spread
break_even_bps = (total_fees / quantity / effective_buy) * 10000
return {
"gross_profit": gross_profit,
"total_fees": total_fees,
"net_profit": net_profit,
"break_even_bps": break_even_bps,
"roi_percent": (net_profit / (effective_buy * quantity)) * 100,
"is_viable": net_profit > 0
}
Test
calc = RealisticFeeCalculator(slippage_bps=2.0)
result = calc.calculate_real_profit(
buy_ex="binance", sell_ex="coinbase",
buy_price=67250.00, sell_price=67310.50,
quantity=1.0
)
print(f"Net profit: ${result['net_profit']:.2f} | Break-even: {result['break_even_bps']:.1f} bps")
Erreur 3 : Race condition sur ordres simultanés
Symptôme : L'ordre d'achat est exécuté mais l'ordre de vente échoue, laissant une position ouverte qui génère des pertes. Ou les deux ordres sont exécutés mais dans le mauvais ordre.
Cause racine : Envoi simultané des deux ordres sans vérification de l'ordre d'exécution. Si le sell arrive avant le buy, le compte est en position courte sur un marché haussier.
SOLUTION : Sémaphore et vérification séquentielle
import asyncio
from contextlib import asynccontextmanager
class SafeArbitrageExecutor:
"""Exécuteur avec lock séquentiel pour éviter les race conditions"""
def __init__(self, max_concurrent: int = 1):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.active_positions = {}
self.lock = asyncio.Lock()
@asynccontextmanager
async def atomic_arbitrage(self, trade_id: str, symbol: str):
"""Context manager pour exécution atomique"""
async with self.semaphore:
async with self.lock:
# Vérifier qu'aucune position n'existe sur ce symbol
if symbol in self.active_positions:
raise RuntimeError(f"Position existante sur {symbol}")
self.active_positions[symbol] = trade_id
try:
yield # L'exécution réelle se fait ici
finally:
async with self.lock:
if symbol in self.active_positions:
del self.active_positions[symbol]
async def execute_safe(
self,
buy_ex: str, sell_ex: str,
symbol: str, quantity: float,
buy_price: float, sell_price: float
) -> dict:
"""Exécution sécurisée avec vérification d'état"""
trade_id = f"{symbol}_{int(time.time() * 1000)}"
try:
async with self.atomic_arbitrage(trade_id, symbol):
# Étape 1: Ordre d'achat avec confirmation
print(f"[{trade_id}] Achat {quantity} {symbol} sur {buy_ex}")
buy_result = await self._place_and_confirm(
buy_ex, "buy", symbol, quantity, buy_price
)
if not buy_result["confirmed"]:
raise Exception(f"Achat non confirmé: {buy_result['error']}")
# Étape 2: Vérification du remplissage avant vente
buy_fill = await self._verify_fill(buy_result["order_id"])
if not buy_fill["filled"]:
# Annulation si non rempli - ordre limite
await self._cancel_order(buy_result["order_id"])
raise Exception(f"Achat non rempli après {buy_fill['wait_ms']}ms")
# Étape 3: Ordre de vente UNIQUEMENT après confirmation achat
print(f"[{trade_id}] Vente {quantity} {symbol} sur {sell_ex}")
sell_result = await self._place_and_confirm(
sell_ex, "sell", symbol, quantity, sell_price
)
return {
"status": "SUCCESS",
"trade_id": trade_id,
"buy_fill": buy_fill,
"sell_fill": await self._verify_fill(sell_result["order_id"])
}
except Exception as e:
return {
"status": "FAILED",
"trade_id": trade_id,
"error": str(e),
"recovery_needed": symbol in self.active_positions
}
async def _place_and_confirm(self, exchange: str, side: str, symbol: str, quantity: float, price: float) -> dict:
"""Placement avec confirmation synchrone"""
# Implémentation dépendante de l'exchange
order = await self._send_order(exchange, side, symbol, quantity, price)
return {"order_id": order["id"], "confirmed": True}
async def _verify_fill(self, order_id: str, timeout_ms: int = 5000) -> dict:
"""Vérification du remplissage avec timeout"""
start = time.time()
while (time.time() - start) * 1000 < timeout_ms:
status = await self._check_order_status(order_id)
if status == "FILLED":
return {"filled": True, "wait_ms": (time.time() - start) * 1000}
await asyncio.sleep(0.01)
return {"filled": False, "wait_ms": timeout_ms}
Pour qui / pour qui ce n'est pas fait
✓ Ce système est fait pour :
- Les traders quantitatifs avec expérience en Python et gestion de risque