En tant qu'ingénieur en trading algorithmique ayant développé des systèmes d'arbitrage crypto pendant plus de trois ans, je témoigne : la difficulté principale ne réside pas dans la conception de la stratégie, mais dans l'obtention de données fiables et dans la construction d'un framework de backtesting suffisamment rapide pour valider vos hypothèses avant de risquer du capital réel. J'ai testé des dizaines de configurations, brûlé des heures de calcul sur des données mal synchronisées, et développé ce framework qui condense tout ce que j'aurais aimé connaître en starting-block de ma carrière quantitatif.
Cet article détaille une architecture complète en Python permettant de récupérer des données OHLCV (Open, High, Low, Close, Volume) sur plusieurs exchanges simultanément, de détecter des opportunités d'arbitrage triangulaire et inter-exchange, puis de simuler ces stratégies sur un historique de plusieurs années avec une précision de latence inférieure à la milliseconde.
Architecture du Système de Récupération de Données Multi-Exchange
Le cœur du système repose sur une architecture asynchrone utilisant asyncio et aiohttp pour maximiser le débit de récupération tout en respectant les limites de rate-limiting de chaque plateforme. Chaque exchange (Binance, OKX, Bybit) possède sa propre structure d'API REST que notre framework unifie derrière une interface commune.
Installation et Configuration de l'Environnement
# Installation des dépendances requises
pip install aiohttp asyncio pandas numpy python-dotenv
pip install ccxt # Couche d'abstraction multi-exchange
pip install pandas-ta # Indicateurs techniques pour le backtesting
pip install scipy # Optimisation des paramètres de stratégie
Structure du projet recommandée
project/
├── config/
│ ├── __init__.py
│ ├── api_credentials.py # Clés API exchange
│ └── holy_config.py # Configuration HolySheep AI
├── data/
│ ├── raw/ # Données brutes K-line
│ ├── processed/ # Données nettoyées et alignées
│ └── backtest_results/ # Résultats de simulation
├── src/
│ ├── collectors/
│ │ ├── __init__.py
│ │ ├── base_collector.py # Classe de base abstraite
│ │ ├── binance_collector.py # Collecteur Binance
│ │ ├── okx_collector.py # Collecteur OKX
│ │ └── bybit_collector.py # Collecteur Bybit
│ ├── arbitrage/
│ │ ├── __init__.py
│ │ ├── triangular.py # Arbitrage triangulaire intra-exchange
│ │ └── exchange.py # Arbitrage inter-exchange
│ └── backtesting/
│ ├── __init__.py
│ ├── engine.py # Moteur de simulation
│ └── performance.py # Métriques de performance
├── notebooks/
│ └── analysis.ipynb # Analyse exploratoire
├── requirements.txt
└── .env # Variables d'environnementCollecteur de Données K-Line avec Gestion du Rate-Limiting
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import ccxt
import time
@dataclass
class KLineData:
"""Structure standardisée pour les données K-line"""
timestamp: int
open: float
high: float
low: float
close: float
volume: float
quote_volume: float
exchange: str
symbol: str
class BaseCollector:
"""Classe de base pour tous les collecteurs d exchange"""
def __init__(self, api_key: str, api_secret: str, rate_limit: int = 1200):
self.api_key = api_key
self.api_secret = api_secret
self.rate_limit = rate_limit # Requêtes par minute
self.request_count = 0
self.last_reset = time.time()
self.session: Optional[aiohttp.ClientSession] = None
async def _acquire_rate_limit(self):
"""Gestion intelligente du rate-limiting"""
current_time = time.time()
if current_time - self.last_reset >= 60:
self.request_count = 0
self.last_reset = current_time
if self.request_count >= self.rate_limit:
wait_time = 60 - (current_time - self.last_reset)
await asyncio.sleep(max(0, wait_time))
self.request_count = 0
self.last_reset = time.time()
self.request_count += 1
async def fetch_klines(
self,
symbol: str,
timeframe: str = "1m",
since: Optional[int] = None,
limit: int = 1000
) -> List[KLineData]:
"""Méthode abstraite à implémenter par chaque exchange"""
raise NotImplementedError
async def fetch_historical(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe: str = "1m"
) -> pd.DataFrame:
"""Récupère l historique complet avec pagination automatique"""
all_klines = []
current_time = start_date
timeframe_minutes = {
"1m": 1, "5m": 5, "15m": 15, "1h": 60,
"4h": 240, "1d": 1440, "1w": 10080
}
interval_ms = timeframe_minutes.get(timeframe, 1) * 60 * 1000
batch_size = 1000
while current_time < end_date:
since = int(current_time.timestamp() * 1000)
end_timestamp = int(end_date.timestamp() * 1000)
max_since = since + (batch_size * interval_ms)
klines = await self.fetch_klines(
symbol=symbol,
timeframe=timeframe,
since=since,
limit=min(batch_size, 1000)
)
if not klines:
break
all_klines.extend(klines)
last_timestamp = klines[-1].timestamp
if last_timestamp >= max_since - interval_ms:
current_time = datetime.fromtimestamp(
(last_timestamp + interval_ms) / 1000
)
else:
break
await asyncio.sleep(0.1) # Éviter la surcharge
df = pd.DataFrame([
{
"timestamp": k.timestamp,
"datetime": pd.to_datetime(k.timestamp, unit="ms"),
"open": k.open,
"high": k.high,
"low": k.low,
"close": k.close,
"volume": k.volume,
"quote_volume": k.quote_volume,
"exchange": k.exchange,
"symbol": k.symbol
}
for k in all_klines
])
return df.drop_duplicates(subset=["timestamp", "exchange"])
class BinanceCollector(BaseCollector):
"""Collecteur spécifique pour Binance Spot et Futures"""
BASE_URL = "https://api.binance.com"
async def fetch_klines(
self,
symbol: str,
timeframe: str = "1m",
since: Optional[int] = None,
limit: int = 1000
) -> List[KLineData]:
await self._acquire_rate_limit()
# Mapping timeframe Binance
binance_tf = {
"1m": "1m", "5m": "5m", "15m": "15m",
"1h": "1h", "4h": "4h", "1d": "1d"
}.get(timeframe, "1m")
url = f"{self.BASE_URL}/api/v3/klines"
params = {
"symbol": symbol.upper(),
"interval": binance_tf,
"limit": limit
}
if since:
params["startTime"] = since
async with self.session.get(url, params=params) as response:
if response.status == 429:
await asyncio.sleep(60)
return await self.fetch_klines(symbol, timeframe, since, limit)
data = await response.json()
return [
KLineData(
timestamp=int(k[0]),
open=float(k[1]),
high=float(k[2]),
low=float(k[3]),
close=float(k[4]),
volume=float(k[5]),
quote_volume=float(k[7]),
exchange="binance",
symbol=symbol
)
for k in data
]
class OKXCollector(BaseCollector):
"""Collecteur spécifique pour OKX"""
BASE_URL = "https://www.okx.com"
async def fetch_klines(
self,
symbol: str,
timeframe: str = "1m",
since: Optional[int] = None,
limit: int = 1000
) -> List[KLineData]:
await self._acquire_rate_limit()
okx_tf = {
"1m": "1m", "5m": "5m", "15m": "15m",
"1h": "1H", "4h": "4H", "1d": "1D"
}.get(timeframe, "1m")
# OKX utilise des paires au format BTC-USDT
okx_symbol = symbol.replace("/", "-")
url = f"{self.BASE_URL}/api/v5/market/history-candles"
params = {
"instId": okx_symbol.upper(),
"bar": okx_tf,
"limit": limit
}
if since:
params["after"] = since
async with self.session.get(url, params=params) as response:
data = await response.json()
if data.get("code") != "0":
raise Exception(f"OKX API Error: {data.get('msg')}")
candles = data.get("data", [])
return [
KLineData(
timestamp=int(c[0]),
open=float(c[1]),
high=float(c[2]),
low=float(c[3]),
close=float(c[4]),
volume=float(c[5]),
quote_volume=float(c[7]),
exchange="okx",
symbol=symbol
)
for c in candles
]
class BybitCollector(BaseCollector):
"""Collecteur spécifique pour Bybit"""
BASE_URL = "https://api.bybit.com"
async def fetch_klines(
self,
symbol: str,
timeframe: str = "1m",
since: Optional[int] = None,
limit: int = 1000
) -> List[KLineData]:
await self._acquire_rate_limit()
bybit_tf = {
"1m": "1", "5m": "5", "15m": "15",
"1h": "60", "4h": "240", "1d": "D"
}.get(timeframe, "1")
url = f"{self.BASE_URL}/v5/market/kline"
params = {
"category": "spot",
"symbol": symbol.upper(),
"interval": bybit_tf,
"limit": limit
}
if since:
params["start"] = since // 1000
async with self.session.get(url, params=params) as response:
data = await response.json()
if data.get("retCode") != 0:
raise Exception(f"Bybit API Error: {data.get('retMsg')}")
klines = data.get("result", {}).get("list", [])
return [
KLineData(
timestamp=int(k[0]),
open=float(k[1]),
high=float(k[2]),
low=float(k[3]),
close=float(k[4]),
volume=float(k[5]),
quote_volume=float(k[7]),
exchange="bybit",
symbol=symbol
)
for k in reversed(klines) # Bybit retourne les données en ordre décroissant
]Intégration avec l'Analyse IA via HolySheep AI
Pour optimiser l'analyse des patterns de prix et la détection automatisée des opportunités d'arbitrage, j'intègre désormais les capacités de modèle de langue de HolySheep AI directement dans mon pipeline de données. L'API HolySheep offre une latence inférieure à 50ms et des tarifs significativement inférieurs aux alternatives mainstream : DeepSeek V3.2 à 0,42 dollar par million de tokens contre 8 dollars pour GPT-4.1, soit une économie de plus de 85%.
L'intégration se fait via l'endpoint https://api.holysheep.ai/v1 avec votre clé API personnelle. Vous pouvez vous S'inscrire ici pour obtenir des crédits gratuits de démarrage et accéder à des méthodes de paiement locales comme WeChat Pay et Alipay, idéales pour les traders basés en Asie ou travaillant avec des volumes importants de données chinoises.
import aiohttp
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class ArbitrageOpportunity:
"""Opportunité d arbitrage détectée"""
timestamp: int
pair1: str # BTC/USDT
pair2: str # ETH/USDT
pair3: str # ETH/BTC
route: str # "BTC→ETH→USDT→BTC"
profit_pct: float
volume_usd: float
confidence: float
exchanges: List[str]
raw_prices: Dict[str, float]
class HolySheepAnalyzer:
"""Analyseur IA pour la détection de patterns et l optimisation de stratégies"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.model = "deepseek-v3.2" # Modèle économique, excellent pour l'analyse
self.session: Optional[aiohttp.ClientSession] = None
async def _request(self, messages: List[Dict]) -> str:
"""Requête générique à l API HolySheep"""
if not self.session:
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
payload = {
"model": self.model,
"messages": messages,
"temperature": 0.3, # Température basse pour des analyses déterministes
"max_tokens": 2000
}
async with self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as response:
if response.status != 200:
error = await response.text()
raise Exception(f"HolySheep API Error: {error}")
result = await response.json()
return result["choices"][0]["message"]["content"]
async def analyze_arbitrage_window(
self,
opportunities: List[ArbitrageOpportunity],
market_context: Dict
) -> Dict:
"""Demande à l IA d analyser une fenêtre d opportunités d arbitrage"""
prompt = f"""Analyse les opportunités d arbitrage suivantes pour la période
{market_context.get('period', 'recent')}:
Opportunités détectées:
{json.dumps([{
'route': o.route,
'profit': f"{o.profit_pct:.4f}%",
'volume': f"${o.volume_usd:,.2f}",
'confiance': f"{o.confidence*100:.1f}%"
} for o in opportunities[:20]], indent=2)}
Contexte marché:
- Volatilité: {market_context.get('volatility', 'N/A')}
- Volume 24h: {market_context.get('volume_24h', 'N/A')}
- Spread moyen: {market_context.get('avg_spread', 'N/A')}
Fournis:
1. Évaluation du risque globale (faible/moyen/élevé)
2. Meilleure opportunité avec justification
3. Paramètres de position recommandés (size, stop-loss)
4. Moments à éviter"""
messages = [
{"role": "system", "content": "Tu es un analyste quantitatif expert en trading crypto."},
{"role": "user", "content": prompt}
]
return await self._request(messages)
async def optimize_strategy_params(
self,
historical_results: Dict,
target_metrics: Dict
) -> Dict:
"""Optimise les paramètres de stratégie basés sur l historique"""
prompt = f"""Optimise les paramètres pour cette stratégie d arbitrage:
Résultats historiques:
{json.dumps(historical_results, indent=2)}
Métriques cibles:
- Sharpe ratio minimum: {target_metrics.get('min_sharpe', 1.5)}
- Drawdown maximum: {target_metrics.get('max_drawdown', 0.1)}%
- Profit mensuel minimum: ${target_metrics.get('min_monthly_profit', 1000)}
Propose une optimisation des hyperparamètres avec justification mathématique."""
messages = [
{"role": "system", "content": "Tu es un expert en optimisation de stratégies de trading."},
{"role": "user", "content": prompt}
]
return await self._request(messages)
async def detect_anomalies(self, price_data: pd.DataFrame) -> List[Dict]:
"""Détecte les anomalies dans les données de prix"""
# Préparation des statistiques
stats = {
"paires_analysées": price_data["symbol"].nunique(),
"période": f"{price_data['timestamp'].min()} - {price_data['timestamp'].max()}",
"nb_obs": len(price_data),
"volatilité_moyenne": price_data.groupby("symbol")["close"].pct_change().std().mean(),
"spike_prix_detectes": len(price_data[price_data["close"].pct_change().abs() > 0.05])
}
prompt = f"""Analyse ces statistiques de données de marché:
{json.dumps(stats, indent=2)}
Identifie:
1. Anomalies potentielles (spikes, gaps, données manquantes)
2. Périodes de liquidité anormale
3. Recommandations de nettoyage de données
4. Risques de survivorship bias"""
messages = [
{"role": "system", "content": "Tu es un data scientist expert en finance quantitative."},
{"role": "user", "content": prompt}
]
return await self._request(messages)
async def example_pipeline():
"""Exemple complet d utilisation du pipeline"""
# Initialisation
holy_sheep = HolySheepAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# Contexte marché à analyser
market_context = {
"period": "2024-01-15 14:00 - 15:00 UTC",
"volatility": "Élevée (VIX crypto > 45)",
"volume_24h": "$45.2B",
"avg_spread": "0.08%"
}
# Opportunités fictives pour l'exemple
opportunities = [
ArbitrageOpportunity(
timestamp=1705324800000,
pair1="BTC/USDT",
pair2="ETH/USDT",
pair3="ETH/BTC",
route="BTC→ETH→USDT→BTC",
profit_pct=0.12,
volume_usd=50000,
confidence=0.85,
exchanges=["binance", "okx"],
raw_prices={"binance": {"BTC": 42000, "ETH": 2650},
"okx": {"BTC": 42005, "ETH": 2648}}
),
ArbitrageOpportunity(
timestamp=1705324860000,
pair1="ETH/USDT",
pair2="BTC/USDT",
pair3="BTC/ETH",
route="ETH→BTC→USDT→ETH",
profit_pct=0.08,
volume_usd=35000,
confidence=0.72,
exchanges=["bybit", "binance"],
raw_prices={"bybit": {"ETH": 2652, "BTC": 42010},
"binance": {"ETH": 2650, "BTC": 42000}}
)
]
# Analyse IA
analysis = await holy_sheep.analyze_arbitrage_window(
opportunities, market_context
)
print("=== ANALYSE HOLYSHEEP AI ===")
print(analysis)
# Fermeture de session
await holy_sheep.session.close()
Exécution
if __name__ == "__main__":
asyncio.run(example_pipeline())Détection d'Arbitrage Triangulaire et Inter-Exchange
Le framework implémente deux types principaux de stratégies d'arbitrage : l'arbitrage triangulaire (exploitant les inefficiencies entre trois paires sur un même exchange) et l'arbitrage inter-exchange (profitant des différences de prix pour un même actif entre plateformes).
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from itertools import permutations
import asyncio
@dataclass
class TriangularOpportunity:
"""Opportunité d arbitrage triangulaire"""
exchange: str
pair1: str # Base → Quote
pair2: str
pair3: str
route: str
initial_amount: float
fees: float
gross_profit_pct: float
net_profit_pct: float
timestamp: int
data: Dict[str, Dict[str, float]]
class ArbitrageDetector:
"""Détecteur d opportunités d arbitrage multi-stratégie"""
# Frais de transaction par exchange (Maker/Taker combinés)
FEES = {
"binance": {"maker": 0.001, "taker": 0.001}, # 0.1%
"okx": {"maker": 0.0015, "taker": 0.002}, # 0.15-0.2%
"bybit": {"maker": 0.001, "taker": 0.002} # 0.1-0.2%
}
def __init__(self, min_profit_threshold: float = 0.001):
self.min_profit_threshold = min_profit_threshold
self.opportunities: List[TriangularOpportunity] = []
def calculate_triangular_arbitrage(
self,
prices: Dict[str, Tuple[float, float]],
symbol: str = "BTC"
) -> List[TriangularOpportunity]:
"""
Calcule les opportunités d arbitrage triangulaire
prices: Dict avec format {"PAIR": (bid, ask)}
Exemple: {"BTC/USDT": (42000, 42001), "ETH/BTC": (0.063, 0.0631), "ETH/USDT": (2646, 2647)}
"""
opportunities = []
# Extraction des prix (bid = meilleur acheteur, ask = meilleur vendeur)
btc_usdt = prices.get("BTC/USDT", prices.get("BTCUSDT"))
eth_btc = prices.get("ETH/BTC", prices.get("ETHBTC"))
eth_usdt = prices.get("ETH/USDT", prices.get("ETHUSDT"))
if not all([btc_usdt, eth_btc, eth_usdt]):
return opportunities
btc_usdt_bid, btc_usdt_ask = btc_usdt
eth_btc_bid, eth_btc_ask = eth_btc
eth_usdt_bid, eth_usdt_ask = eth_usdt
# Route 1: USDT → BTC → ETH → USDT
# Acheter BTC (payer USDT), Acheter ETH (payer BTC), Vendre ETH (recevoir USDT)
route1_gross = (1 / btc_usdt_ask) * (1 / eth_btc_ask) * eth_usdt_bid
# Route 2: USDT → ETH → BTC → USDT
# Acheter ETH (payer USDT), Vendre ETH (recevoir BTC), Vendre BTC (recevoir USDT)
route2_gross = (1 / eth_usdt_ask) * eth_btc_bid * btc_usdt_bid
# Route 3: BTC → USDT → ETH → BTC
route3_gross = btc_usdt_bid * eth_usdt_ask * eth_btc_bid
# Route 4: ETH → BTC → USDT → ETH
route4_gross = (1 / eth_btc_ask) * btc_usdt_bid * eth_usdt_ask
routes = [
("USDT→BTC→ETH→USDT", route1_gross, "BTC/USDT", "ETH/BTC", "ETH/USDT"),
("USDT→ETH→BTC→USDT", route2_gross, "ETH/USDT", "ETH/BTC", "BTC/USDT"),
("BTC→USDT→ETH→BTC", route3_gross, "BTC/USDT", "ETH/USDT", "ETH/BTC"),
("ETH→BTC→USDT→ETH", route4_gross, "ETH/BTC", "BTC/USDT", "ETH/USDT"),
]
for route_name, gross, p1, p2, p3 in routes:
# Calcul des frais (3 transactions)
total_fees = 0.006 # ~0.2% par transaction * 3
net_profit = (gross - 1) - total_fees
if net_profit > self.min_profit_threshold:
opportunities.append(TriangularOpportunity(
exchange="binance",
pair1=p1,
pair2=p2,
pair3=p3,
route=route_name,
initial_amount=10000,
fees=total_fees,
gross_profit_pct=(gross - 1) * 100,
net_profit_pct=net_profit * 100,
timestamp=int(pd.Timestamp.now().timestamp() * 1000),
data={
"BTC/USDT": {"bid": btc_usdt_bid, "ask": btc_usdt_ask},
"ETH/BTC": {"bid": eth_btc_bid, "ask": eth_btc_ask},
"ETH/USDT": {"bid": eth_usdt_bid, "ask": eth_usdt_ask}
}
))
return opportunities
async def detect_inter_exchange_arbitrage(
self,
all_prices: Dict[str, Dict[str, Tuple[float, float]]],
min_profit: float = 0.001
) -> List[Dict]:
"""
Détecte les opportunités d arbitrage inter-exchange
all_prices: {
"binance": {"BTC/USDT": (41950, 41952), ...},
"okx": {"BTC/USDT": (41955, 41957), ...},
"bybit": {"BTC/USDT": (41948, 41950), ...}
}
"""
opportunities = []
exchanges = list(all_prices.keys())
# Comparaison des prix pour chaque paire
for pair in ["BTC/USDT", "ETH/USDT", "SOL/USDT", "BNB/USDT"]:
best_buy = None # Prix le plus bas (où acheter)
best_sell = None # Prix le plus haut (où vendre)
for exchange, prices in all_prices.items():
if pair not in prices:
continue
bid, ask = prices[pair]
if best_buy is None or ask < best_buy[1]:
best_buy = (exchange, ask)
if best_sell is None or bid > best_sell[1]:
best_sell = (exchange, bid)
if best_buy and best_sell and best_buy[0] != best_sell[0]:
profit_pct = ((best_sell[1] / best_buy[1]) - 1) * 100
fees = 0.004 # Frais achat + vente
net_profit = profit_pct - (fees * 100)
if net_profit > min_profit * 100:
opportunities.append({
"pair": pair,
"buy_exchange": best_buy[0],
"sell_exchange": best_sell[0],
"buy_price": best_buy[1],
"sell_price": best_sell[1],
"gross_profit_pct": profit_pct,
"net_profit_pct": net_profit,
"estimated_volume": 10000,
"potential_profit": (10000 / best_buy[1]) * (1 - fees) * best_sell[1] * (1 - fees) - 10000,
"timestamp": int(pd.Timestamp.now().timestamp() * 1000)
})
return opportunities
def scan_all_pairs(self, prices_df: pd.DataFrame, exchanges: List[str]) -> pd.DataFrame:
"""
Scan toutes les combinaisons de paires pour des opportunités
"""
results = []
for exchange in exchanges:
exchange_data = prices_df[prices_df["exchange"] == exchange]
# Construction du dictionnaire de prix
price_dict = {}
for _, row in exchange_data.iterrows():
key = row["symbol"]
price_dict[key] = (row["close"] * 0.9999, row["close"] * 1.0001) # Simuler bid/ask
# Calcul des opportunités triangulaires
opps = self.calculate_triangular_arbitrage(price_dict)
results.extend(opps)
if results:
return pd.DataFrame([
{
"exchange": o.exchange,
"route": o.route,
"gross_profit_pct": o.gross_profit_pct,
"net_profit_pct": o.net_profit_pct,
"fees": o.fees,
"timestamp": o.timestamp
}
for o in results
])
return pd.DataFrame()
Exemple d utilisation
async def main():
detector = ArbitrageDetector(min_profit_threshold=0.001)
# Données de prix simulées (en réalité, récupérées via les collecteurs)
sample_prices = {
"BTC/USDT": (41950.00, 41952.50),
"ETH/BTC": (0.06310, 0.06312),
"ETH/USDT": (2648.00, 2649.00)
}
# Calcul des opportunités triangulaires
opps = detector.calculate_triangular_arbitrage(sample_prices)
print("=== OPPORTUNITÉS TRIANGULAIRES ===")
for opp in opps:
print(f"Route: {opp.route}")
print(f" Profit brut: {opp.gross_profit_pct:.4f}%")
print(f" Frais: {opp.fees * 100:.2f}%")
print(f" Profit net: {opp.net_profit_pct:.4f}%")
print()
# Arbitrage inter-exchange
inter_exchange_prices = {
"binance": {"BTC/USDT": (41950, 41952)},
"okx": {"BTC/USDT": (41955, 41958)},
"bybit": {"BTC/USDT": (41948, 41950)}
}
inter_opps = await detector.detect_inter_exchange_arbitrage(inter_exchange_prices)
print("=== OPPORTUNITÉS INTER-EXCHANGE ===")
for opp in inter_opps:
print(f"{opp['pair']}: Acheter sur {opp['buy_exchange']} à {opp['buy_price']}, "
f"Vendre sur {opp['sell_exchange']} à {opp['sell_price']}")
print(f" Profit net: {opp['net_profit_pct']:.4f}%")
if __name__ == "__main__":
asyncio.run(main())Moteur de Backtesting avec Gestion du Slippage
Le backtesting constitue la phase critique de validation. Un moteur réaliste doit reproduire fidèlement les conditions réelles de marché : slippage, latence d'exécution, frais de transaction variables, et impact sur le prix en fonction de la taille de l'ordre.
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime
from enum import Enum
import json
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP_LOSS = "stop_loss"
TAKE_PROFIT = "take_profit"
@dataclass
class Order:
timestamp: int
symbol: str
side: str # "buy" ou "sell"
order_type: OrderType
amount: float
price: Optional[float] = None
executed_price: Optional[float] = None
fee: float = 0.0
status: str = "pending"
@dataclass
class Position:
symbol: str
side: str
entry_price: float
amount: float
entry_time: int
pnl: float = 0.0
realized_pnl: float = 0.0
@dataclass
class BacktestConfig:
initial_capital: float = 100000
commission_rate: float = 0.001 # 0.1%
slippage_pct: float = 0.0005 # 0.05%
max_position_size_pct: float = 0.1 # Max 10% du capital par position
leverage: float = 1.0
risk_free_rate: float = 0.02 # Taux sans risque annuel
@dataclass
class BacktestResult:
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
profit_factor: float
total_trades: int
avg_trade_duration: float
equity_curve: pd.DataFrame
trades: List[Dict]
metrics: Dict
class BacktestingEngine:
"""Moteur de backtesting haute fidélité"""
def __init__(self, config: BacktestConfig):
self.config = config
self.positions: List[Position] = []
self.orders: List[Order] = []
self.equity_history: List[Dict] = []