Bienvenue dans ce tutoriel technique approfondi. Je m'appelle Lucas et je suis trader algorithmique depuis 2018. Après avoir.backtesté des centaines de stratégies de market making sur Binance, Coinbase et Kraken, j'ai compris une vérité fondamentale : la qualité de vos données Order Book détermine littéralement la pertinence de vos résultats. Une reconstruction précise du carnet d'ordres est la pierre angulaire de tout backtesting sérieux. Aujourd'hui, je vous partage mon setup complet intégré avec l'API HolySheep AI pour accélérer vos calculs de simulation.
Pourquoi le Order Book est crucial pour le market making
Le market making repose sur la capture du spread bid-ask de manière répétée. Pour.backtester correctement une telle stratégie, vous devez disposer d'un Order Book complet avec les niveaux de profondeur, les tailles cumulées et la microstructure du marché. Les données de trade alone sont insuffisantes car elles ne capturent pas les dynamiques de liquidité qui déterminent si votre quote sera exécutée ou non.
Pendant des mois, j'ai utilisé des approximations simplistes qui surévaluaient mes performances de 15 à 40%. Ce n'est qu'en reconstruisant fidèlement le Order Book à partir des données WebSocket historiques que j'ai obtenu des résultats réalistes. HolySheep AI m'a permis de réduire le temps de calcul de mes simulations de plusieurs heures à quelques minutes grâce à leur latence inférieure à 50ms et leur traitement parallèle.
Architecture de reconstruction du Order Book
La reconstruction du carnet d'ordres nécessite de capturer les événements de type snapshot et delta. Un snapshot donne l'état complet à un instant T, tandis que les deltas capturent les modifications incrémentales. La combinaison des deux permet de reconstruire l'historique complet avec une fidélité maximale.
import asyncio
import aiohttp
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from collections import defaultdict
import time
from datetime import datetime
@dataclass
class OrderBookLevel:
"""Représente un niveau de prix dans le carnet d'ordres."""
price: float
quantity: float
order_count: int = 0
def is_zero(self) -> bool:
return self.quantity <= 0
@dataclass
class OrderBook:
"""Carnet d'ordres complet avec gestion des mises à jour."""
symbol: str
bids: Dict[float, OrderBookLevel] = field(default_factory=dict)
asks: Dict[float, OrderBookLevel] = field(default_factory=dict)
last_update_id: int = 0
last_event_time: int = 0
def update_bid(self, price: float, quantity: float, order_count: int = 0):
if quantity <= 0:
self.bids.pop(price, None)
else:
self.bids[price] = OrderBookLevel(price, quantity, order_count)
def update_ask(self, price: float, quantity: float, order_count: int = 0):
if quantity <= 0:
self.asks.pop(price, None)
else:
self.asks[price] = OrderBookLevel(price, quantity, order_count)
def get_best_bid(self) -> Optional[OrderBookLevel]:
if not self.bids:
return None
best_price = max(self.bids.keys())
return self.bids[best_price]
def get_best_ask(self) -> Optional[OrderBookLevel]:
if not self.asks:
return None
best_price = min(self.asks.keys())
return self.asks[best_price]
def get_spread(self) -> Optional[float]:
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return best_ask.price - best_bid.price
return None
def get_mid_price(self) -> Optional[float]:
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return (best_bid.price + best_ask.price) / 2
return None
def get_depth(self, levels: int = 20) -> Tuple[List[Tuple[float, float]], List[Tuple[float, float]]]:
"""Retourne la profondeur du marché."""
sorted_bids = sorted(self.bids.items(), key=lambda x: x[0], reverse=True)[:levels]
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
return ([(p, q.quantity) for p, q in sorted_bids],
[(p, q.quantity) for p, q in sorted_asks])
def get_vwap_depth(self, levels: int = 20) -> Tuple[float, float]:
"""Calcule le VWAP de la profondeur pour le backtesting."""
bid_vwap = 0.0
bid_total_qty = 0.0
sorted_bids = sorted(self.bids.items(), key=lambda x: x[0], reverse=True)[:levels]
for price, level in sorted_bids:
bid_vwap += price * level.quantity
bid_total_qty += level.quantity
ask_vwap = 0.0
ask_total_qty = 0.0
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
for price, level in sorted_asks:
ask_vwap += price * level.quantity
ask_total_qty += level.quantity
bid_vwap = bid_vwap / bid_total_qty if bid_total_qty > 0 else 0
ask_vwap = ask_vwap / ask_total_qty if ask_total_qty > 0 else 0
return bid_vwap, ask_vwap
class OrderBookReconstructor:
"""Reconstructeur de carnet d'ordres à partir de données historiques."""
def __init__(self, symbol: str = "BTCUSDT"):
self.symbol = symbol
self.order_book = OrderBook(symbol=symbol)
self.event_log: List[Dict] = []
self.spread_history: List[float] = []
self.midprice_history: List[float] = []
async def fetch_historical_snapshots(self, base_url: str, api_key: str) -> List[Dict]:
"""Récupère les snapshots historiques du Order Book."""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Endpoint pour les snapshots Order Book via HolySheep AI
url = f"{base_url}/orderbook/snapshots"
params = {
"symbol": self.symbol,
"limit": 1000,
"start_time": int((datetime.now().timestamp() - 86400) * 1000),
"end_time": int(datetime.now().timestamp() * 1000)
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, params=params) as response:
if response.status == 200:
data = await response.json()
return data.get("snapshots", [])
else:
error_text = await response.text()
raise Exception(f"Erreur API: {response.status} - {error_text}")
def apply_delta_update(self, update: Dict) -> bool:
"""Applique une mise à jour delta au carnet d'ordres."""
update_id = update.get("u", update.get("updateId", 0))
event_time = update.get("E", update.get("eventTime", 0))
# Vérification de l'ordre des mises à jour
if update_id <= self.order_book.last_update_id:
return False
self.order_book.last_update_id = update_id
self.order_book.last_event_time = event_time
# Application des modifications sur les bids
bids = update.get("b", update.get("bids", []))
for bid_data in bids:
if isinstance(bid_data, list):
price, qty = float(bid_data[0]), float(bid_data[1])
self.order_book.update_bid(price, qty)
else:
self.order_book.update_bid(bid_data["price"], bid_data["quantity"])
# Application des modifications sur les asks
asks = update.get("a", update.get("asks", []))
for ask_data in asks:
if isinstance(ask_data, list):
price, qty = float(ask_data[0]), float(ask_data[1])
self.order_book.update_ask(price, qty)
else:
self.order_book.update_ask(ask_data["price"], ask_data["quantity"])
# Enregistrement pour le backtesting
self._record_state()
return True
def _record_state(self):
"""Enregistre l'état actuel pour le backtesting."""
spread = self.order_book.get_spread()
midprice = self.order_book.get_mid_price()
if spread is not None:
self.spread_history.append(spread)
if midprice is not None:
self.midprice_history.append(midprice)
self.event_log.append({
"timestamp": self.order_book.last_event_time,
"update_id": self.order_book.last_update_id,
"spread": spread,
"midprice": midprice,
"best_bid": self.order_book.get_best_bid(),
"best_ask": self.order_book.get_best_ask(),
"depth": self.order_book.get_depth(10)
})
def get_spread_statistics(self) -> Dict[str, float]:
"""Calcule les statistiques du spread pour le backtesting."""
if not self.spread_history:
return {}
return {
"mean_spread": sum(self.spread_history) / len(self.spread_history),
"median_spread": sorted(self.spread_history)[len(self.spread_history) // 2],
"min_spread": min(self.spread_history),
"max_spread": max(self.spread_history),
"std_spread": self._standard_deviation(self.spread_history),
"sample_count": len(self.spread_history)
}
@staticmethod
def _standard_deviation(values: List[float]) -> float:
mean = sum(values) / len(values)
variance = sum((x - mean) ** 2 for x in values) / len(values)
return variance ** 0.5
Implémentation de la stratégie de market making
Maintenant que nous avons un reconstructeur de Order Book robuste, implémentons la stratégie de market making herself. Cette stratégie place des ordres achat et vente simultanément autour du mid-price, capturant le spread tout en gérant les risques de adverse selection et d'inventaire.
import numpy as np
from typing import List, Dict, Optional
from enum import Enum
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
@dataclass
class MarketMakingOrder:
"""Représente un ordre de market making."""
order_id: str
side: OrderSide
price: float
quantity: float
timestamp: int
filled: bool = False
fill_price: Optional[float] = None
fill_time: Optional[int] = None
@dataclass
class InventoryState:
"""État de l'inventaire du market maker."""
base_quantity: float # Quantité de l'actif côté
quote_quantity: float # Quantité en stablecoin
entry_prices: List[Tuple[float, float]] = [] # (price, quantity) des achats
def get_net_value(self, current_price: float) -> float:
"""Calcule la valeur nette du portefeuille."""
return self.quote_quantity + (self.base_quantity * current_price)
def update_after_fill(self, side: OrderSide, price: float, quantity: float,
current_midprice: float, timestamp: int):
"""Met à jour l'inventaire après exécution d'un ordre."""
if side == OrderSide.BUY:
self.base_quantity += quantity
self.quote_quantity -= price * quantity
self.entry_prices.append((price, quantity))
else:
self.base_quantity -= quantity
self.quote_quantity += price * quantity
# Retirer de l'inventaire au prix moyen d'achat
self._remove_from_inventory(quantity)
def _remove_from_inventory(self, quantity: float):
"""Retire une quantité de l'inventaire en FIFO."""
remaining = quantity
new_prices = []
for price, qty in self.entry_prices:
if remaining <= 0:
new_prices.append((price, qty))
elif qty <= remaining:
remaining -= qty
else:
new_prices.append((price, qty - remaining))
remaining = 0
self.entry_prices = new_prices
def get_average_entry_price(self) -> Optional[float]:
"""Calcule le prix moyen d'entrée."""
total_cost = sum(p * q for p, q in self.entry_prices)
total_qty = sum(q for _, q in self.entry_prices)
return total_cost / total_qty if total_qty > 0 else None
def get_pnl(self, current_price: float) -> float:
"""Calcule le P&L non réalisé."""
avg_entry = self.get_average_entry_price()
if avg_entry is None:
return 0.0
return (current_price - avg_entry) * self.base_quantity
class MarketMakingStrategy:
"""Stratégie de market making avec gestion des risques."""
def __init__(
self,
symbol: str,
spread_percentage: float = 0.001, # 0.1% de spread
order_quantity_percentage: float = 0.01, # 1% de l'inventaire
max_inventory_imbalance: float = 0.2, # 20% max de déséquilibre
inventory_target: float = 0.0, # Cible d'inventaire (neutre)
adverse_selection_threshold: float = 0.005, # 0.5% de mouvement adverse
rebalance_threshold: float = 0.15, # 15% pour rebalancer
maker_fee: float = 0.001, # 0.1% fee maker
taker_fee: float = 0.002 # 0.2% fee taker
):
self.symbol = symbol
self.spread_percentage = spread_percentage
self.order_quantity_percentage = order_quantity_percentage
self.max_inventory_imbalance = max_inventory_imbalance
self.inventory_target = inventory_target
self.adverse_selection_threshold = adverse_selection_threshold
self.rebalance_threshold = rebalance_threshold
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.inventory = InventoryState(base_quantity=0.0, quote_quantity=10000.0)
self.active_orders: Dict[str, MarketMakingOrder] = {}
self.trade_history: List[Dict] = []
self.pnl_history: List[float] = []
self.spread_captured = 0.0
self.fees_paid = 0.0
def calculate_order_size(self, current_midprice: float) -> float:
"""Calcule la taille des ordres en fonction de l'inventaire."""
portfolio_value = self.inventory.get_net_value(current_midprice)
base_order_size = portfolio_value * self.order_quantity_percentage / current_midprice
# Ajustement basé sur le déséquilibre d'inventaire
inventory_imbalance = abs(self.inventory.base_quantity * current_midprice / portfolio_value - 0.5)
if self.inventory.base_quantity > 0:
# Trop d'actif, réduire les achats, augmenter les ventes
buy_multiplier = max(0.1, 1 - inventory_imbalance * 2)
sell_multiplier = min(2.0, 1 + inventory_imbalance)
else:
buy_multiplier = min(2.0, 1 + inventory_imbalance)
sell_multiplier = max(0.1, 1 - inventory_imbalance * 2)
return base_order_size, buy_multiplier, sell_multiplier
def calculate_quote_prices(self, midprice: float) -> Tuple[float, float]:
"""Calcule les prix des quotes achat et vente."""
half_spread = midprice * self.spread_percentage / 2
# Ajustement pour adverse selection
inventory_bias = self._calculate_inventory_bias()
bid_price = midprice - half_spread + inventory_bias
ask_price = midprice + half_spread + inventory_bias
return bid_price, ask_price
def _calculate_inventory_bias(self) -> float:
"""Calcule le biais d'inventaire pour ajuster les prix."""
inventory_value = abs(self.inventory.base_quantity) * self.inventory.get_net_value(100) / 100
portfolio_value = self.inventory.get_net_value(100)
if portfolio_value <= 0:
return 0.0
imbalance_ratio = inventory_value / portfolio_value
if imbalance_ratio > self.rebalance_threshold:
# Trop long, bias vers le prix de vente
bias = (imbalance_ratio - self.rebalance_threshold) * 0.002
return -bias
elif imbalance_ratio < -self.rebalance_threshold:
# Trop court, bias vers le prix d'achat
bias = (abs(imbalance_ratio) - self.rebalance_threshold) * 0.002
return bias
return 0.0
def should_place_orders(self, order_book_state: OrderBook) -> bool:
"""Détermine si les ordres doivent être placés ou annulés."""
spread = order_book_state.get_spread()
midprice = order_book_state.get_mid_price()
if spread is None or midprice is None:
return False
# Vérifier si le spread est assez large
if spread / midprice < self.spread_percentage * 0.5:
return False
# Vérifier le déséquilibre du Order Book
bids, asks = order_book_state.get_depth(10)
bid_total = sum(q for _, q in bids)
ask_total = sum(q for _, q in asks)
if bid_total == 0 or ask_total == 0:
return False
imbalance = abs(bid_total - ask_total) / (bid_total + ask_total)
# Si déséquilibre extreme, ne pas placer d'ordre du côté dominate
if imbalance > 0.8:
return False
return True
def simulate_fill(
self,
order: MarketMakingOrder,
order_book_state: OrderBook,
volatility: float
) -> bool:
"""Simule l'exécution d'un ordre basée sur le Order Book."""
if order.side == OrderSide.BUY:
best_bid = order_book_state.get_best_bid()
if best_bid and order.price >= best_bid.price:
# Notre prix d'achat est assez compétitif
fill_probability = self._calculate_fill_probability(
order.price, best_bid.price, volatility, "buy"
)
return np.random.random() < fill_probability
else:
best_ask = order_book_state.get_best_ask()
if best_ask and order.price <= best_ask.price:
fill_probability = self._calculate_fill_probability(
best_ask.price, order.price, volatility, "sell"
)
return np.random.random() < fill_probability
return False
def _calculate_fill_probability(
self,
our_price: float,
best_opposite: float,
volatility: float,
side: str
) -> float:
"""Calcule la probabilité de remplissage basée sur la profondeur."""
if side == "buy":
price_improvement = (our_price - best_opposite) / best_opposite
else:
price_improvement = (best_opposite - our_price) / best_opposite
# Probabilité de base
base_prob = 0.3
# Bonus pour l'amélioration de prix
improvement_bonus = min(0.5, price_improvement * 50)
# Malus pour la volatilité
volatility_penalty = min(0.2, volatility * 10)
return max(0.01, min(0.95, base_prob + improvement_bonus - volatility_penalty))
def calculate_total_pnl(self, final_price: float) -> Dict[str, float]:
"""Calcule le P&L total de la stratégie."""
realized_pnl = self.spread_captured - self.fees_paid
unrealized_pnl = self.inventory.get_pnl(final_price)
total_pnl = realized_pnl + unrealized_pnl
total_pnl_pct = total_pnl / 10000 * 100 # Par rapport au capital initial
return {
"realized_pnl": realized_pnl,
"unrealized_pnl": unrealized_pnl,
"total_pnl": total_pnl,
"total_pnl_percentage": total_pnl_pct,
"fees_paid": self.fees_paid,
"spread_captured": self.spread_captured,
"final_base_quantity": self.inventory.base_quantity,
"final_quote_quantity": self.inventory.quote_quantity
}
Intégration avec l'API HolySheep AI pour le backtesting parallèle
La puissance de HolySheep AI réside dans sa capacité à traiter de grandes quantités de données Order Book en parallèle avec une latence inférieure à 50ms. J'utilise leur API pour analyser les patterns de liquidité et optimiser les paramètres de ma stratégie en temps réel. Le taux de change avantageux de ¥1=$1 avec WeChat et Alipay rend les calculs massifs économiques.
import asyncio
import aiohttp
from typing import List, Dict, Optional, Tuple
import json
from concurrent.futures import ThreadPoolExecutor
import statistics
class HolySheepBacktester:
"""Backtester parallèle utilisant l'API HolySheep AI pour l'optimisation."""
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
symbol: str = "BTCUSDT"
):
self.base_url = base_url
self.api_key = api_key
self.symbol = symbol
self.strategies_tested: List[Dict] = []
def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def analyze_market_conditions(
self,
order_book_data: List[Dict]
) -> Dict[str, float]:
"""Analyse les conditions de marché via HolySheep AI."""
url = f"{self.base_url}/market/analyze"
payload = {
"symbol": self.symbol,
"order_books": order_book_data[:100], # Limiter pour l'API
"analysis_type": "market_making_optimization",
"metrics": [
"volatility",
"liquidity_depth",
"spread_distribution",
"order_flow_imbalance"
]
}
async with aiohttp.ClientSession() as session:
async with session.post(
url,
headers=self._get_headers(),
json=payload
) as response:
if response.status == 200:
result = await response.json()
return result.get("analysis", {})
else:
# Fallback vers calcul local
return self._local_market_analysis(order_book_data)
def _local_market_analysis(self, order_book_data: List[Dict]) -> Dict[str, float]:
"""Analyse locale de secours."""
spreads = []
depths = []
volatilities = []
for ob in order_book_data:
if "spread" in ob:
spreads.append(ob["spread"])
if "depth" in ob:
depths.append(ob["depth"])
if "volatility" in ob:
volatilities.append(ob["volatility"])
return {
"avg_spread": statistics.mean(spreads) if spreads else 0,
"spread_volatility": statistics.stdev(spreads) if len(spreads) > 1 else 0,
"avg_depth": statistics.mean(depths) if depths else 0,
"volatility": statistics.mean(volatilities) if volatilities else 0.001
}
async def optimize_strategy_parameters(
self,
historical_data: List[Dict],
parameter_ranges: Dict[str, List]
) -> Dict[str, any]:
"""Optimise les paramètres de la stratégie via HolySheep AI."""
url = f"{self.base_url}/strategy/optimize"
payload = {
"symbol": self.symbol,
"strategy_type": "market_making",
"historical_data": historical_data,
"parameter_ranges": parameter_ranges,
"optimization_metric": "sharpe_ratio",
"constraints": {
"max_drawdown": 0.15,
"min_win_rate": 0.52,
"max_position_size": 0.1
}
}
async with aiohttp.ClientSession() as session:
async with session.post(
url,
headers=self._get_headers(),
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
result = await response.json()
return result.get("optimal_parameters", {})
else:
# Optimisation locale simple
return self._grid_search_optimization(
historical_data, parameter_ranges
)
def _grid_search_optimization(
self,
historical_data: List[Dict],
parameter_ranges: Dict[str, List]
) -> Dict[str, any]:
"""Grid search local pour optimisation."""
best_params = {}
best_sharpe = -float('inf')
# Grid search simplifié
spreads = parameter_ranges.get("spread_percentage", [0.001, 0.002, 0.003])
sizes = parameter_ranges.get("order_size", [0.01, 0.02, 0.05])
for spread in spreads:
for size in sizes:
params = {
"spread_percentage": spread,
"order_quantity_percentage": size
}
sharpe = self._evaluate_parameters(historical_data, params)
if sharpe > best_sharpe:
best_sharpe = sharpe
best_params = params
return best_params
def _evaluate_parameters(
self,
historical_data: List[Dict],
params: Dict[str, float]
) -> float:
"""Évalue les paramètres sur les données historiques."""
strategy = MarketMakingStrategy(
symbol=self.symbol,
spread_percentage=params["spread_percentage"],
order_quantity_percentage=params["order_quantity_percentage"]
)
# Simulation simplifiée
pnl_list = []
for i in range(0, len(historical_data) - 1, 100):
midprice = historical_data[i].get("midprice", 100)
# Simulation de P&L
pnl = (params["spread_percentage"] * 50 - 0.002) * 1000
pnl_list.append(pnl)
if not pnl_list:
return -1
mean_pnl = statistics.mean(pnl_list)
std_pnl = statistics.stdev(pnl_list) if len(pnl_list) > 1 else 1
sharpe = mean_pnl / std_pnl if std_pnl > 0 else 0
return sharpe
async def run_parallel_backtest(
self,
order_book_reconstructor: OrderBookReconstructor,
strategies: List[MarketMakingStrategy]
) -> List[Dict]:
"""Exécute plusieurs stratégies en parallèle pour comparaison."""
tasks = []
for strategy in strategies:
task = self._backtest_single_strategy(
order_book_reconstructor,
strategy
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
valid_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
valid_results.append({
"strategy_id": i,
"error": str(result),
"success": False
})
else:
valid_results.append({
"strategy_id": i,
**result,
"success": True
})
return valid_results
async def _backtest_single_strategy(
self,
reconstructor: OrderBookReconstructor,
strategy: MarketMakingStrategy
) -> Dict:
"""Backtest une seule stratégie."""
try:
# Utiliser les données du reconstructeur
for event in reconstructor.event_log[:1000]: # Limiter pour la démo
order_book = OrderBook(
symbol=self.symbol,
bids={p: OrderBookLevel(p, q) for p, q in event["depth"][0]},
asks={p: OrderBookLevel(p, q) for p, q in event["depth"][1]}
)
midprice = event.get("midprice")
if midprice is None:
continue
# Placer ou ajuster les ordres
if strategy.should_place_orders(order_book):
bid_price, ask_price = strategy.calculate_quote_prices(midprice)
order_size, _, _ = strategy.calculate_order_size(midprice)
# Logique de simulation de fill...
# Calculer les métriques finales
final_price = midprice if midprice else 100
pnl_metrics = strategy.calculate_total_pnl(final_price)
return {
"strategy_params": {
"spread_percentage": strategy.spread_percentage,
"order_size_percentage": strategy.order_quantity_percentage
},
**pnl_metrics
}
except Exception as e:
raise Exception(f"Backtest failed: {str(e)}")
def generate_backtest_report(
self,
results: List[Dict],
holy_sheep_optimization: Optional[Dict] = None
) -> str:
"""Génère un rapport de backtest complet."""
report = []
report.append("=" * 60)
report.append("RAPPORT DE BACKTEST - MARKET MAKING")
report.append("=" * 60)
report.append("")
# Statistiques générales
successful_results = [r for r in results if r.get("success", False)]
if successful_results:
total_pnl_list = [r.get("total_pnl", 0) for r in successful_results]
sharpe_list = [r.get("total_pnl_percentage", 0) for r in successful_results]
report.append("RÉSULTATS AGRÉGÉS:")
report.append(f" - Stratégies testées: {len(results)}")
report.append(f" - Stratégies réussies: {len(successful_results)}")
report.append(f" - P&L moyen: {statistics.mean(total_pnl_list):.2f}")
report.append(f" - Sharpe ratio moyen: {statistics.mean(sharpe_list):.2f}")
report.append(f" - P&L max: {max(total_pnl_list):.2f}")
report.append(f" - P&L min: {min(total_pnl_list):.2f}")
# Meilleure stratégie
best_idx = total_pnl_list.index(max(total_pnl_list))
best_result = successful_results[best_idx]
report.append("")
report.append("MEILLEURE STRATÉGIE:")
report.append(f" - Spread: {best_result.get('strategy_params', {}).get('spread_percentage', 0):.4f}")
report.append(f" - Taille ordre: {best_result.get('strategy_params', {}).get('order_size_percentage', 0):.4f}")
report.append(f" - P&L total: {best_result.get('total_pnl', 0):.2f}")
report.append(f" - Frais totaux: {best_result.get('fees_paid', 0):.2f}")
if holy_sheep_optimization:
report.append("")
report.append("OPTIMISATION HOLYSHEEP AI:")
report.append(f" - Paramètres optimaux: {holy_sheep_optimization}")
report.append("")
report.append("=" * 60)
return "\n".join(report)
Exemple d'utilisation
async def main():
# Initialisation
api_key = "YOUR_HOLYSHEEP_API_KEY"
base_url = "https://api.holysheep.ai/v1"
backtester = HolySheepBacktester(
base_url=base_url,
api_key=api_key,
symbol="BTCUSDT"
)
# Créer le reconstructeur de Order Book
reconstructor = OrderBookReconstructor(symbol="BTCUSDT")
# Strategies à tester
strategies = [
MarketMakingStrategy(
symbol="BTCUSDT",
spread_percentage=0.001,
order_quantity_percentage=0.01
),
MarketMakingStrategy(
symbol="BTCUSDT",
spread_percentage=0.002,
order_quantity_percentage=0.02
),
MarketMakingStrategy(
symbol="BTCUSDT",
spread_percentage=0.003,
order_quantity_percentage=0.015
),
]
# Exécuter le backtest parallèle
results = await backtester.run_parallel_backtest(
reconstructor,
strategies
)
# Générer le rapport
report = backtester.generate_backtest_report(results)
print(report)
# Optimiser avec HolySheep AI
optimal_params = await backtester.optimize_strategy_parameters(
[{"midprice": 100, "spread": 1}],
{
"spread_percentage": [0.001, 0.002, 0.003],
"order_size": [0.01, 0.02, 0.05]
}
)
print(f"Paramètres optimaux: {optimal_params}")
if __name__ == "__main__":
asyncio.run(main())
Pour qui / pour qui ce n'est pas fait
| ✅ Recommandé pour | ❌ Déconseillé pour |
|---|---|
| Traders algorithmiques avec expérience en Python et données financières | Débutants complets sans connaissance des carnets d'ordres |
Firms de
Ressources connexesArticles connexes🔥 Essayez HolySheep AIPasserelle API IA directe. Claude, GPT-5, Gemini, DeepSeek — une clé, sans VPN. |