En tant que développeur indépendant spécialisé dans les systèmes de trading algorithmique depuis maintenant quatre ans, j'ai traversé toutes les étapes possibles de frustration : des APIs qui changent sans préavis, des latences qui ruinent vos stratégies, des coûts qui explosent le budget de développement. Il y a six mois, j'ai reçu un mandat particulièrement stimulant : construire un système de backtesting haute fréquence capable de traiter les données orderbook L2 d'OKX avec une précision au millisecond près. Le défi ? Intégrer ces flux massifs tout en maintenant des coûts d'infrastructure raisonnables — et c'est là que HolySheep AI est devenu mon allié stratégique.
Cas d'Utilisation Concret : Backtesting d'une Stratégie Market-Making
Imaginons que vous développiez une stratégie de market-making sur la paire BTC/USDT d'OKX. Votre système doit :
- Ingérer les 400+ mises à jour d'ordre par seconde du flux L2
- Recalculer le carnet d'ordres en temps réel
- Simuler des transactions avec des frais de maker/taker réalistes
- Générer des rapports de performance avec drawdown et Sharpe ratio
Sans une architecture adaptée, votre système Python va s'effondrer sous le poids des données. Voici comment structurer cette intégration proprement.
Architecture de l'Intégration OKX L2
1. Configuration de l'Environnement
# Installation des dépendances
pip install okx-sdk pandas numpy asyncio aiohttp
Structure du projet
project/
├── config/
│ ├── __init__.py
│ ├── settings.py # Configuration OKX API
│ └── backtest_config.py # Paramètres de backtest
├── data/
│ ├── __init__.py
│ ├── okx_connector.py # Connecteur WebSocket OKX
│ └── orderbook_manager.py # Gestionnaire de carnet d'ordres
├── backtest/
│ ├── __init__.py
│ ├── engine.py # Moteur de backtest
│ └── strategy.py # Template de stratégie
├── utils/
│ ├── __init__.py
│ └── holysheep_client.py # Client API HolySheep
├── main.py
└── requirements.txt2. Connecteur WebSocket OKX L2
import asyncio
import json
import aiohttp
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class OrderBookEntry:
"""Représente une entrée du carnet d'ordres L2"""
price: float
size: float
side: str # 'bid' ou 'ask'
timestamp: int
orders_count: int = 1
@dataclass
class OrderBook:
"""Carnet d'ordres L2 complet"""
symbol: str
bids: List[OrderBookEntry] = field(default_factory=list)
asks: List[OrderBookEntry] = field(default_factory=list)
timestamp: int = 0
local_timestamp: float = 0.0
def get_mid_price(self) -> float:
if not self.bids or not self.asks:
return 0.0
return (self.bids[0].price + self.asks[0].price) / 2
def get_spread_bps(self) -> float:
"""Calcule le spread en basis points"""
if not self.bids or not self.asks or self.bids[0].price == 0:
return 0.0
return (self.asks[0].price - self.bids[0].price) / self.bids[0].price * 10000
class OKXL2Connector:
"""
Connecteur WebSocket pour les données orderbook L2 d'OKX.
Documentation: https://www.okx.com/docs-v5/
"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(self, symbol: str = "BTC-USDT-SWAP"):
self.symbol = symbol
self.orderbook = OrderBook(symbol=symbol)
self._running = False
self._ws = None
self._session = None
self._buffer = [] # Buffer pour le backtesting
self._message_count = 0
self._last_stats_time = datetime.now()
def _format_symbol(self) -> str:
"""Convertit le symbole au format OKX (ex: BTC-USDT-SWAP -> BTC-USDT-SWAP)"""
return self.symbol
async def connect(self) -> None:
"""Établit la connexion WebSocket"""
self._session = aiohttp.ClientSession()
self._ws = await self._session.ws_connect(self.WS_URL)
self._running = True
# Souscription au flux L2
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books",
"instId": self._format_symbol()
}]
}
await self._ws.send_json(subscribe_msg)
logger.info(f"Souscrit au flux L2 pour {self.symbol}")
async def _parse_orderbook_update(self, data: dict) -> Optional[OrderBook]:
"""Parse une mise à jour du carnet d'ordres"""
try:
if "data" not in data:
return None
for update in data["data"]:
# Extraction des bids et asks
bids_raw = update.get("bids", [])
asks_raw = update.get("asks", [])
timestamp = int(update.get("ts", 0))
bids = [
OrderBookEntry(
price=float(b[0]),
size=float(b[1]),
side="bid",
timestamp=timestamp,
orders_count=int(b[2]) if len(b) > 2 else 1
)
for b in bids_raw
]
asks = [
OrderBookEntry(
price=float(a[0]),
size=float(a[1]),
side="ask",
timestamp=timestamp,
orders_count=int(a[2]) if len(a[2]) > 2 else 1
)
for a in asks_raw
]
# Mise à jour incrémentale ou snapshot
if update.get("action") == "snapshot":
self.orderbook.bids = bids
self.orderbook.asks = asks
else:
# Mise à jour incrémentale
self._apply_incremental_update(bids, asks)
self.orderbook.timestamp = timestamp
self.orderbook.local_timestamp = datetime.now().timestamp()
# Stockage pour le backtest
self._buffer.append({
"timestamp": timestamp,
"orderbook": OrderBook(
symbol=self.symbol,
bids=list(self.orderbook.bids),
asks=list(self.orderbook.asks),
timestamp=timestamp
)
})
self._message_count += 1
except Exception as e:
logger.error(f"Erreur de parsing: {e}")
return self.orderbook
def _apply_incremental_update(self, bids: List[OrderBookEntry],
asks: List[OrderBookEntry]) -> None:
"""Applique les mises à jour incrémentales au carnet"""
for bid in bids:
if bid.size == 0:
# Suppression
self.orderbook.bids = [
b for b in self.orderbook.bids
if abs(b.price - bid.price) > 1e-8
]
else:
# Insertion ou mise à jour
found = False
for i, b in enumerate(self.orderbook.bids):
if abs(b.price - bid.price) < 1e-8:
self.orderbook.bids[i] = bid
found = True
break
if not found:
self.orderbook.bids.append(bid)
# Même logique pour les asks
for ask in asks:
if ask.size == 0:
self.orderbook.asks = [
a for a in self.orderbook.asks
if abs(a.price - ask.price) > 1e-8
]
else:
found = False
for i, a in enumerate(self.orderbook.asks):
if abs(a.price - ask.price) < 1e-8:
self.orderbook.asks[i] = ask
found = True
break
if not found:
self.orderbook.asks.append(ask)
# Tri et limitation
self.orderbook.bids.sort(key=lambda x: x.price, reverse=True)
self.orderbook.asks.sort(key=lambda x: x.price)
self.orderbook.bids = self.orderbook.bids[:25] # Top 25 bids
self.orderbook.asks = self.orderbook.asks[:25] # Top 25 asks
async def listen(self, callback=None) -> None:
"""Boucle principale d'écoute"""
await self.connect()
while self._running:
msg = await self._ws.receive()
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
orderbook = await self._parse_orderbook_update(data)
if callback and orderbook:
await callback(orderbook)
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"Erreur WebSocket: {msg.data}")
break
# Logging des statistiques
now = datetime.now()
if (now - self._last_stats_time).seconds >= 10:
elapsed = (now - self._last_stats_time).seconds
rate = self._message_count / elapsed if elapsed > 0 else 0
logger.info(f"Messages/seconde: {rate:.1f}, Buffer: {len(self._buffer)}")
self._message_count = 0
self._last_stats_time = now
def stop(self) -> None:
"""Arrête la connexion"""
self._running = False
def get_buffer(self) -> List[dict]:
"""Retourne le buffer de données pour le backtest"""
return self._buffer
async def close(self) -> None:
"""Ferme proprement la connexion"""
self.stop()
if self._ws:
await self._ws.close()
if self._session:
await self._session.close()3. Intégration HolySheep pour l'Analyse IA
Ici intervient la magie HolySheep. Pour analyser les patterns de liquidité et détecter les anomalies dans votre carnet d'ordres, j'utilise leur API avec DeepSeek V3.2 — le modèle le plus économique du marché à seulement 0,42 $ par million de tokens. Le taux de change avantageux (¥1 = $1) rend le service encore plus accessible.
aiohttp.ClientSession: """Lazy initialization de la session HTTP""" if self._session is None or self._session.closed: self._session = aiohttp.ClientSession( headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" }, timeout=aiohttp.ClientTimeout(total=10) ) return self._session async def analyze_liquidity_pattern( self, orderbook_data: Dict, context: Optional[str] = None ) -> Dict: """ Analyse les patterns de liquidité via DeepSeek V3.2. Args: orderbook_data: Données du carnet d'ordres context: Contexte additionnel (volatilité, actualité...) Returns: Analyse structurée avec recommandations """ prompt = self._build_liquidity_prompt(orderbook_data, context) session = await self._get_session() payload = { "model": "deepseek-v3.2", "messages": [ { "role": "system", "content": """Tu es un analyste quantitatif expert en carnet d'ordres. Réponds en JSON avec les champs: - liquidite_score (0-100) - imbalance_ratio (rapport bid/ask) - recommendations (liste de conseils) - risk_factors (liste des risques identifiés)""" }, { "role": "user", "content": prompt } ], "temperature": 0.3, "max_tokens": 500 } try: async with session.post( f"{self.BASE_URL}/chat/completions", json=payload ) as response: if response.status != 200: error_text = await response.text() raise Exception(f"API Error {response.status}: {error_text}") result = await response.json() content = result["choices"][0]["message"]["content"] # Parsing JSON de la réponse return json.loads(content) except Exception as e: return { "error": str(e), "liquidite_score": 50, "imbalance_ratio": 1.0, "recommendations": ["Vérifier la connexion API"], "risk_factors": ["Erreur de connexion"] } def _build_liquidity_prompt(self, orderbook_data: Dict, context: Optional[str]) -> str: """Construit le prompt pour l'analyse""" symbol = orderbook_data.get("symbol", "UNKNOWN") bids = orderbook_data.get("bids", [])[:10] asks = orderbook_data.get("asks", [])[:10] bids_str = "\n".join([ f" Prix {b['price']} | Taille {b['size']}" for b in bids ]) asks_str = "\n".join([ f" Prix {a['price']} | Taille {a['size']}" for a in asks ]) prompt = f"""Analyse du carnet d'ordres pour {symbol}: BIDS (Achats): {bids_str} ASKS (Ventes): {asks_str} """ if context: prompt += f"Contexte: {context}\n" prompt += """ Donne une analyse quantitative de la liquidité et des risques. """ return prompt async def batch_analyze( self, orderbook_samples: List[Dict], batch_size: int = 10 ) -> List[Dict]: """ Analyse un batch de snapshots orderbook. Optimisé pour le post-backtesting. Args: orderbook_samples: Liste de snapshots orderbook batch_size: Nombre de requêtes parallèles Returns: Liste des analyses """ results = [] for i in range(0, len(orderbook_samples), batch_size): batch = orderbook_samples[i:i+batch_size] # Requêtes parallèles tasks = [ self.analyze_liquidity_pattern(sample) for sample in batch ] batch_results = await asyncio.gather(*tasks, return_exceptions=True) results.extend(batch_results) # Rate limiting doux if i + batch_size < len(orderbook_samples): await asyncio.sleep(0.1) return results async def generate_backtest_report( self, backtest_results: Dict, holysheep_api_key: str ) -> str: """ Génère un rapport d'analyse post-backtest avec l'IA. Utilise DeepSeek V3.2 pour une analyse approfondie. Returns: Rapport HTML formaté """ session = await self._get_session() summary_prompt = f"""Génère un rapport de backtest détaillé: Performance: - Sharpe Ratio: {backtest_results.get('sharpe_ratio', 0):.2f} - Drawdown Max: {backtest_results.get('max_drawdown', 0)*100:.1f}% - Return Total: {backtest_results.get('total_return', 0)*100:.1f}% - Win Rate: {backtest_results.get('win_rate', 0)*100:.1f}% Métriques: - Nb Trades: {backtest_results.get('num_trades', 0)} - PnL Moyen: ${backtest_results.get('avg_pnl', 0):.2f} - Volatilité: {backtest_results.get('volatility', 0)*100:.2f}% Fournis une analyse approfondie et des recommandations. Réponds en français, format HTML. """ payload = { "model": "deepseek-v3.2", "messages": [ {"role": "system", "content": "Tu es un analyste quantitatif expert."}, {"role": "user", "content": summary_prompt} ], "temperature": 0.5, "max_tokens": 1000 } try: async with session.post( f"{self.BASE_URL}/chat/completions", json=payload ) as response: result = await response.json() return result["choices"][0]["message"]["content"] except Exception as e: return f" Erreur génération rapport: {e}
" async def close(self) -> None: """Ferme la session HTTP""" if self._session and not self._session.closed: await self._session.close()--- Exemple d'utilisation ---
async def main(): # Initialisation holysheep = HolySheepAnalysisClient( api_key="YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé ) # Analyse d'un snapshot sample_orderbook = { "symbol": "BTC-USDT-SWAP", "bids": [ {"price": 67450.50, "size": 2.5}, {"price": 67449.00, "size": 1.8}, {"price": 67448.50, "size": 3.2} ], "asks": [ {"price": 67451.00, "size": 1.2}, {"price": 67452.00, "size": 2.1}, {"price": 67453.50, "size": 0.9} ] } # Analyse avec IA analysis = await holysheep.analyze_liquidity_pattern( sample_orderbook, context="Volatilité élevée suite aux annonces Fed" ) print(f"Score liquidité: {analysis.get('liquidite_score', 'N/A')}") print(f"Recommandations: {analysis.get('recommendations', [])}") await holysheep.close() if __name__ == "__main__": asyncio.run(main())
4. Moteur de Backtesting
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
@dataclass
class Trade:
"""Représente une transaction exécutée"""
timestamp: int
side: OrderSide
price: float
size: float
fee: float
fee_tier: str = "taker"
@property
def value(self) -> float:
return self.price * self.size
@property
def total_cost(self) -> float:
return self.value + self.fee if self.side == OrderSide.BUY else self.value + self.fee
@dataclass
class Position:
"""Position ouverte"""
entry_price: float
size: float
entry_time: int
side: OrderSide
@dataclass
class BacktestStats:
"""Statistiques de performance"""
total_return: float = 0.0
sharpe_ratio: float = 0.0
max_drawdown: float = 0.0
win_rate: float = 0.0
num_trades: int = 0
avg_pnl: float = 0.0
volatility: float = 0.0
pnl_history: List[float] = field(default_factory=list)
class BacktestEngine:
"""
Moteur de backtesting haute performance pour données orderbook L2.
Supporte les frais maker/taker, le slippage et l'exécution réaliste.
"""
# Frais OKX (exemple pour tier VIP 0)
MAKER_FEE = 0.0002 # 0.02%
TAKER_FEE = 0.0005 # 0.05%
def __init__(
self,
initial_capital: float = 100000.0,
max_position_pct: float = 0.1,
slippage_bps: float = 1.0
):
self.initial_capital = initial_capital
self.capital = initial_capital
self.max_position_pct = max_position_pct
self.slippage_bps = slippage_bps
self.position: Optional[Position] = None
self.trades: List[Trade] = []
self.equity_curve: List[float] = []
self.daily_returns: List[float] = []
self.current_time: int = 0
self._last_equity = initial_capital
def execute_order(
self,
side: OrderSide,
price: float,
size: float,
order_type: OrderType = OrderType.MARKET,
limit_price: Optional[float] = None
) -> Optional[Trade]:
"""Exécute un ordre avec prise en compte du slippage et des frais"""
if size <= 0:
return None
# Calcul du prix d'exécution avec slippage
if order_type == OrderType.MARKET:
if side == OrderSide.BUY:
exec_price = price * (1 + self.slippage_bps / 10000)
else:
exec_price = price * (1 - self.slippage_bps / 10000)
else:
exec_price = limit_price if limit_price else price
# Calcul des frais
fee = exec_price * size * self.TAKER_FEE
fee_tier = "taker"
trade = Trade(
timestamp=self.current_time,
side=side,
price=exec_price,
size=size,
fee=fee,
fee_tier=fee_tier
)
# Mise à jour du capital et position
if side == OrderSide.BUY:
self.capital -= trade.total_cost
if self.position is None:
self.position = Position(
entry_price=exec_price,
size=size,
entry_time=self.current_time,
side=side
)
else:
# Averaging
total_size = self.position.size + size
self.position.entry_price = (
(self.position.entry_price * self.position.size + exec_price * size)
/ total_size
)
self.position.size = total_size
else: # SELL
self.capital += (trade.value - trade.fee)
self.position = None
self.trades.append(trade)
return trade
def update_equity(self, current_price: float) -> float:
"""Met à jour l'equity curve avec le prix mark-to-market"""
# Calcul de la valeur actuelle du portfolio
portfolio_value = self.capital
if self.position:
if self.position.side == OrderSide.BUY:
unrealized_pnl = (current_price - self.position.entry_price) * self.position.size
else:
unrealized_pnl = (self.position.entry_price - current_price) * self.position.size
portfolio_value += unrealized_pnl
self.equity_curve.append(portfolio_value)
# Calcul du return quotidien
if len(self.equity_curve) > 1:
daily_return = (portfolio_value - self._last_equity) / self._last_equity
self.daily_returns.append(daily_return)
self._last_equity = portfolio_value
return portfolio_value
def process_orderbook_snapshot(
self,
orderbook,
strategy_func: Callable,
**strategy_params
) -> None:
"""Traite un snapshot du carnet d'ordres via la stratégie"""
self.current_time = orderbook.timestamp
# Mise à jour de l'equity
mid_price = orderbook.get_mid_price()
if mid_price > 0:
self.update_equity(mid_price)
# Exécution de la stratégie
signal = strategy_func(
orderbook=orderbook,
position=self.position,
capital=self.capital,
**strategy_params
)
if signal:
action = signal.get("action")
size_pct = signal.get("size_pct", self.max_position_pct)
max_size = self.capital * size_pct / mid_price if mid_price > 0 else 0
if action == "buy" and self.position is None:
self.execute_order(OrderSide.BUY, mid_price, max_size)
elif action == "sell" and self.position is not None:
self.execute_order(OrderSide.SELL, mid_price, self.position.size)
def get_stats(self) -> BacktestStats:
"""Calcule les statistiques finales de backtest"""
if not self.equity_curve:
return BacktestStats()
equity = np.array(self.equity_curve)
returns = np.diff(equity) / equity[:-1]
# Métriques de base
total_return = (equity[-1] - self.initial_capital) / self.initial_capital
# Sharpe Ratio (annualisé, 假设252 jours de trading)
if len(returns) > 0 and np.std(returns) > 0:
sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(252)
else:
sharpe_ratio = 0.0
# Maximum Drawdown
cummax = np.maximum.accumulate(equity)
drawdowns = (cummax - equity) / cummax
max_drawdown = np.max(drawdowns)
# Win Rate
pnl_list = []
for trade in self.trades:
if trade.side == OrderSide.SELL and self.trades.index(trade) > 0:
# Trouver le trade d'achat correspondant
for i in range(len(pnl_list)-1, -1, -1):
if pnl_list[i] == 0:
# Calculer PnL approximatif
pnl_list[i] = trade.value
break
winning_trades = sum(1 for p in pnl_list if p > 0)
num_trades = len([t for t in self.trades if t.side == OrderSide.SELL])
win_rate = winning_trades / num_trades if num_trades > 0 else 0
# Volatilité annualisée
volatility = np.std(returns) * np.sqrt(252) if len(returns) > 0 else 0
# PnL moyen par trade
avg_pnl = (equity[-1] - self.initial_capital) / num_trades if num_trades > 0 else 0
return BacktestStats(
total_return=total_return,
sharpe_ratio=sharpe_ratio,
max_drawdown=max_drawdown,
win_rate=win_rate,
num_trades=num_trades,
avg_pnl=avg_pnl,
volatility=volatility,
pnl_history=pnl_list
)
def generate_report(self) -> Dict:
"""Génère un rapport complet de backtest"""
stats = self.get_stats()
return {
"configuration": {
"capital_initial": self.initial_capital,
"capital_final": self.equity_curve[-1] if self.equity_curve else self.initial_capital,
"max_position_pct": self.max_position_pct,
"slippage_bps": self.slippage_bps
},
"performance": {
"return_total": f"{stats.total_return*100:.2f}%",
"sharpe_ratio": f"{stats.sharpe_ratio:.2f}",
"max_drawdown": f"{stats.max_drawdown*100:.2f}%",
"win_rate": f"{stats.win_rate*100:.1f}%",
"volatilite": f"{stats.volatility*100:.2f}%",
"nb_trades": stats.num_trades,
"pnl_moyen": f"${stats.avg_pnl:.2f}"
},
"frais_totaux": {
"total": f"${sum(t.fee for t in self.trades):.2f}",
"avg_par_trade": f"${np.mean([t.fee for t in self.trades]):.3f}" if self.trades else "$0"
},
"equity_curve": self.equity_curve,
"trades": [
{
"time": t.timestamp,
"side": t.side.value,
"price": t.price,
"size": t.size,
"fee": t.fee
}
for t in self.trades
]
}
--- Exemple de stratégie simple ---
def simple_midprice_strategy(
orderbook,
position: Optional[Position],
capital: float,
imbalance_threshold: float = 0.1,
spread_threshold: float = 5.0
) -> Optional[Dict]:
"""
Stratégie simple basée sur le prix moyen et l'imbalance du livre.
Achète quand:
- Le spread est faible (< spread_threshold bps)
- L'imbalance est en faveur des bids (plus de volume acheteur)
Vends quand:
- Position existante et profit > 0.1%
- Ou l'imbalance se retourne
"""
if not orderbook.bids or not orderbook.asks:
return None
mid_price = orderbook.get_mid_price()
spread_bps = orderbook.get_spread_bps()
# Calcul de l'imbalance
bid_volume = sum(b.size for b in orderbook.bids[:10])
ask_volume = sum(a.size for a in orderbook.asks[:10])
total_volume = bid_volume + ask_volume
if total_volume > 0:
imbalance = (bid_volume - ask_volume) / total_volume
else:
imbalance = 0
# Logique de trading
if position is None:
# Chercher une opportunité d'achat
if spread_bps < spread_threshold and imbalance > imbalance_threshold:
return {
"action": "buy",
"size_pct": 0.05,
"reason": f"Imbalance={imbalance:.2f}, Spread={spread_bps:.1f}bps"
}
else:
# Vérifier si on doit vendre
current_pnl_pct = (mid_price - position.entry_price) / position.entry_price
if current_pnl_pct > 0.001: # Profit > 0.1%
return {
"action": "sell",
"size_pct": 1.0,
"reason": f"Profit={current_pnl_pct*100:.2f}%"
}
elif imbalance < -imbalance_threshold:
return {
"action": "sell",
"size_pct": 1.0,
"reason": f"Imbalance négative={imbalance:.2f}"
}
return None
--- Script principal ---
async def run_backtest():
from okx_connector import OKXL2Connector
# Collecte des données (30 minutes)
connector = OKXL2Connector(symbol="BTC-USDT-SWAP")
print("Collecte des données orderbook...")
await connector.listen()
# Simulation de durée de collecte
await asyncio.sleep(1800) # 30 minutes
connector.stop()
# Initialisation du moteur
engine = BacktestEngine(
initial_capital=50000.0,
max_position_pct=0.1,
slippage_bps=1.0
)
# Exécution du backtest
print("Exécution du backtest...")
buffer = connector.get_buffer()
for snapshot in buffer:
engine.process_orderbook_snapshot(
orderbook=snapshot["orderbook"],
strategy_func=simple_midprice_strategy,
imbalance_threshold=0.15,
spread_threshold=3.0
)
# Génération du rapport
report = engine.generate_report()
print("\n=== RAPPORT DE BACKTEST ===")
print(f"Return Total: {report['performance']['return_total']}")
print(f"Sharpe Ratio: {report['performance']['sharpe_ratio']}")
print(f"Max Drawdown: {report['performance']['max_drawdown']}")
print(f"Nombre de Trades: {report['performance']['nb_trades']}")
print(f"Frais Totaux: {report['frais_totaux']['total']}")
await connector.close()
return report
if __name__ == "__main__":
report = asyncio.run(run_backtest())