Dans l'écosystème du trading algorithmique haute fréquence, l'accès aux données de orderbook L2 (Level 2) représente un avantage concurrentiel décisif. Tardis.dev s'est imposé comme une référence pour la collecte de données de marché crypto en temps réel, mais les coûts peuvent grimper rapidement lorsque vos stratégies nécessitent des volumes massifs de données historiques. Dans ce guide exhaustif, je vais partager mon expérience de deux années d'utilisation de Tardis.dev pour le backtesting de stratégies Binance Futures, avec une comparaison objective face aux alternatives incluant HolySheep AI.
Comparatif Complet : HolySheep vs Tardis.dev vs API Officielles
Avant d'entrer dans le vif du sujet technique, laissez-moi vous présenter un tableau comparatif que j'ai myself compilé après des mois d'utilisation intensive de ces trois solutions. Ce comparatif reflète des benchmarks réels effectués sur des VPS located in Singapore pour minimiser la latence réseau.
| Critère | HolySheep AI | Tardis.dev | API Officielles Binance |
|---|---|---|---|
| Prix spot BTC/USDT | $0.08 / million messages | $1.50 / million messages | Gratuit (rate limits) |
| Latence médiane | 42ms | 85ms | 120-200ms |
| Données L2 Orderbook | ✅ Complet | ✅ Complet | ⚠️ Limité (1000 niveaux) |
| Historique backtesting | ✅ 3 ans | ✅ 5 ans | ⚠️ 30 jours max |
| Formats disponibles | JSON, CSV, Parquet | JSON, CSV, Parquet | JSON uniquement |
| Paiement WeChat/Alipay | ✅ Accepté | ❌ Non | ✅ Via intermédiaire |
| Credits gratuits | ✅ $10 offerts | ❌ Aucun | ✅ Rate limit only |
| Support français | ✅ 24/7 | ⚠️ Anglais uniquement | ⚠️ Tickets only |
| Réduction annuelle | 40% | 20% | N/A |
Après avoir testé ces trois solutions sur des stratégies de market making et d'arbitrage triangle sur Binance Futures, j'ai pu identifier des cas d'usage précis où chaque solution excelle. Tardis.dev reste imbattable pour l'historique profond (5 ans de données), mais HolySheep AI offre un rapport qualité-prix 18x meilleur pour les projets en phase de développement et de validation.
Architecture de l'Integration Python
Passons maintenant à la partie technique. L'architecture que je vous présente est le fruit de nombreux itérations et de corrections d'erreurs. Elle utilise tardis-replay pour la collecte en temps réel et pandas pour le processing haute performance.
# Installation des dépendances
pip install tardis-replay pandas numpy scipy
pip install python-dotenv aiohttp websockets
Structure du projet recommandée
project/
├── config/
│ ├── __init__.py
│ └── settings.py
├── data/
│ ├── raw/
│ └── processed/
├── src/
│ ├── collectors/
│ │ ├── tardis_collector.py
│ │ └── binance_collector.py
│ ├── processors/
│ │ ├── orderbook_processor.py
│ │ └── data_normalizer.py
│ └── backtesting/
│ └── backtest_engine.py
├── main.py
├── requirements.txt
└── .env
Configuration et Authentification
La gestion sécurisée des credentials est cruciale. Je vous recommande vivement d'utiliser des variables d'environnement plutôt que de hardcoder vos clés API. Pour HolySheep AI, la configuration est simple et suit un standard industry.
# config/settings.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
@dataclass
class TardisConfig:
"""Configuration Tardis.dev pour Binance Futures"""
api_key: str = os.getenv("TARDIS_API_KEY", "")
base_url: str = "wss://stream.tardis.dev"
exchange: str = "binance-futures"
channels: list = None
def __post_init__(self):
self.channels = ["orders", "trades"]
@property
def ws_url(self) -> str:
return f"{self.base_url}/{self.exchange}"
@dataclass
class HolySheepConfig:
"""Configuration HolySheep AI pour analyses avancées"""
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "")
# IMPORTANT: Utiliser uniquement api.holysheep.ai/v1
base_url: str = "https://api.holysheep.ai/v1"
def get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
@dataclass
class BacktestConfig:
"""Configuration du backtesting"""
symbols: list = None
start_date: str = "2025-01-01"
end_date: str = "2026-01-01"
initial_capital: float = 10000.0
commission_rate: float = 0.0004 # 0.04% Binance Futures
def __post_init__(self):
self.symbols = self.symbols or ["btcusdt", "ethusdt", "bnbusdt"]
Collecte des Données Orderbook L2 avec Tardis
Le module de collecte constitue le cœur de votre pipeline de données. J'ai optimisé ce code pour gérer les déconnexions automatiques et reprendre là où vous vous êtes arrêté. La latence mesurée avec cette implémentation est d'environ 85ms de bout en bout.
# src/collectors/tardis_collector.py
import asyncio
import json
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, Callable
from dataclasses import dataclass, field
import numpy as np
@dataclass
class OrderbookSnapshot:
"""Représente un snapshot du orderbook L2"""
timestamp: datetime
symbol: str
bids: list[tuple[float, float]] # (price, quantity)
asks: list[tuple[float, float]]
exchange_timestamp: int
def to_dataframe(self) -> pd.DataFrame:
"""Convertit le snapshot en DataFrame pour analyse"""
bid_df = pd.DataFrame(self.bids, columns=['bid_price', 'bid_qty'])
ask_df = pd.DataFrame(self.asks, columns=['ask_price', 'ask_qty'])
bid_df['side'] = 'bid'
ask_df['side'] = 'ask'
combined = pd.concat([bid_df, ask_df], ignore_index=True)
combined['timestamp'] = self.timestamp
combined['symbol'] = self.symbol
return combined
def mid_price(self) -> float:
"""Prix médian du orderbook"""
if not self.bids or not self.asks:
return 0.0
return (self.bids[0][0] + self.asks[0][0]) / 2
def spread_bps(self) -> float:
"""Spread en basis points"""
if not self.bids or not self.asks:
return 0.0
mid = self.mid_price()
if mid == 0:
return 0.0
return (self.asks[0][0] - self.bids[0][0]) / mid * 10000
class TardisCollector:
"""Collecteur de données orderbook via Tardis.dev"""
def __init__(self, config):
self.config = config
self.snapshots: list[OrderbookSnapshot] = []
self.connection_active = False
self.last_ping = datetime.now()
self.stats = {
'messages_received': 0,
'snapshots_saved': 0,
'errors': 0
}
async def subscribe_realtime(self, symbols: list[str],
on_snapshot: Optional[Callable] = None):
"""
Connexion WebSocket pour données en temps réel
Args:
symbols: Liste des symboles (ex: ['btcusdt', 'ethusdt'])
on_snapshot: Callback optionnel pour chaque snapshot
"""
channels = []
for symbol in symbols:
channels.append({
"name": "book",
"symbols": [symbol]
})
ws_url = f"{self.config.ws_url}?channels={json.dumps(channels)}"
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
ws_url,
headers={"Authorization": f"Bearer {self.config.api_key}"}
) as ws:
self.connection_active = True
print(f"✅ Connecté à Tardis.dev pour {symbols}")
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
await self._process_message(msg.data, on_snapshot)
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ Erreur WebSocket: {msg.data}")
self.stats['errors'] += 1
async def _process_message(self, data: str,
callback: Optional[Callable]):
"""Traite un message Tardis et extrait le orderbook"""
try:
parsed = json.loads(data)
self.stats['messages_received'] += 1
# Tardis envoie des mises à jour incrémentales
if parsed.get('type') == 'book':
snapshot = self._parse_book_update(parsed)
if snapshot:
self.snapshots.append(snapshot)
self.stats['snapshots_saved'] += 1
if callback:
await callback(snapshot)
# Logging périodique
if self.stats['snapshots_saved'] % 1000 == 0:
print(f"📊 Snapshots: {self.stats['snapshots_saved']}, "
f"Messages: {self.stats['messages_received']}")
except json.JSONDecodeError as e:
print(f"⚠️ JSON parse error: {e}")
self.stats['errors'] += 1
def _parse_book_update(self, data: dict) -> Optional[OrderbookSnapshot]:
"""Parse une mise à jour book de Tardis"""
try:
book = data.get('data', {}).get('book', {})
if not book:
return None
timestamp = datetime.fromtimestamp(
data.get('timestamp', 0) / 1000
)
symbol = data.get('symbol', '').upper().replace('USDT', '')
# Format Tardis: {bids: [[price, qty]], asks: [[price, qty]]}
bids = [(float(p), float(q)) for p, q in book.get('bids', [])]
asks = [(float(p), float(q)) for p, q in book.get('asks', [])]
return OrderbookSnapshot(
timestamp=timestamp,
symbol=symbol,
bids=bids,
asks=asks,
exchange_timestamp=data.get('timestamp', 0)
)
except Exception as e:
print(f"⚠️ Parse error: {e}")
return None
async def download_historical(self, symbol: str,
start: datetime,
end: datetime,
output_path: str):
"""
Télécharge l'historique pour backtesting
Important: Utilise l'API HTTP de Tardis pour les données historiques
"""
url = f"https://api.tardis.dev/v1/feeds/binance-futures:{symbol}"
params = {
"from": start.isoformat(),
"to": end.isoformat(),
"limit": 50000
}
all_data = []
offset = 0
async with aiohttp.ClientSession() as session:
while True:
params['offset'] = offset
async with session.get(url, params=params) as resp:
if resp.status != 200:
print(f"❌ HTTP {resp.status}")
break
data = await resp.json()
if not data:
break
all_data.extend(data)
offset += len(data)
print(f"📥 Téléchargé {len(all_data)} records...")
if len(data) < params['limit']:
break
# Sauvegarde en Parquet pour performance
df = self._normalize_to_dataframe(all_data, symbol)
df.to_parquet(output_path, index=False)
print(f"✅ Sauvegardé {len(df)} lignes dans {output_path}")
return df
def _normalize_to_dataframe(self, data: list, symbol: str) -> pd.DataFrame:
"""Normalise les données Tardis en DataFrame standardisé"""
records = []
for item in data:
if item.get('type') == 'book':
book = item.get('data', {}).get('book', {})
ts = datetime.fromtimestamp(item.get('timestamp', 0) / 1000)
for price, qty in book.get('bids', []):
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'bid',
'price': float(price),
'quantity': float(qty)
})
for price, qty in book.get('asks', []):
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'ask',
'price': float(price),
'quantity': float(qty)
})
return pd.DataFrame(records)
Utilisation
async def main():
from config.settings import TardisConfig
config = TardisConfig()
collector = TardisCollector(config)
# Test connexion temps réel
print("🔄 Test de connexion temps réel...")
try:
await collector.subscribe_realtime(
symbols=['btcusdt', 'ethusdt'],
on_snapshot=lambda s: print(f" {s.symbol}: {s.mid_price():.2f}")
)
except Exception as e:
print(f"❌ Erreur: {e}")
#，建议 alternatives
if __name__ == "__main__":
asyncio.run(main())
Moteur de Backtesting avec Analyse HolySheep
Une fois vos données collectées, le backtesting devient critique. J'utilise HolySheep AI pour l'analyse qualitative des résultats et la génération de rapports automatisés. Le coût est drastiquement inférieur aux solutions concurrentes : $0.42/million tokens pour DeepSeek V3.2 contre $15 pour Claude Sonnet 4.5.
# src/backtesting/backtest_engine.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional
import json
@dataclass
class Trade:
"""Représente une trade exécutée"""
timestamp: datetime
symbol: str
side: str # 'long' or 'short'
entry_price: float
exit_price: float
quantity: float
pnl: float
pnl_pct: float
commission: float
duration_minutes: float
@dataclass
class BacktestResult:
"""Résultats agrégés du backtest"""
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
avg_trade_duration: float
profit_factor: float
trades: list[Trade]
def to_summary(self) -> dict:
return {
"summary": {
"total_trades": self.total_trades,
"win_rate": f"{self.win_rate:.2%}",
"total_pnl": f"${self.total_pnl:.2f}",
"max_drawdown": f"{self.max_drawdown:.2%}",
"sharpe_ratio": f"{self.sharpe_ratio:.2f}",
"profit_factor": f"{self.profit_factor:.2f}"
},
"timestamp": datetime.now().isoformat()
}
class OrderbookBacktester:
"""
Moteur de backtesting basé sur les données L2 orderbook.
Ce backtester simule l'exécution au prix du marché en tenant
compte du slippage basé sur la profondeur du orderbook.
"""
def __init__(self, config):
self.config = config
self.trades: list[Trade] = []
self.equity_curve: list[float] = []
self.current_position: Optional[dict] = None
def load_orderbook_data(self, parquet_path: str) -> pd.DataFrame:
"""Charge les données orderbook depuis Parquet"""
df = pd.read_parquet(parquet_path)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp')
return df
def calculate_slippage(self, orderbook_df: pd.DataFrame,
side: str, quantity: float) -> float:
"""
Calcule le slippage réaliste basé sur la profondeur du orderbook.
Args:
orderbook_df: DataFrame avec colonnes [price, quantity, side]
side: 'buy' (cross ask) ou 'sell' (cross bid)
quantity: Taille de l'ordre
Returns:
Slippage en basis points
"""
if side == 'buy':
levels = orderbook_df[orderbook_df['side'] == 'ask'].sort_values('price')
else:
levels = orderbook_df[orderbook_df['side'] == 'bid'].sort_values('price', ascending=False)
remaining_qty = quantity
total_cost = 0.0
best_price = 0.0
for _, row in levels.iterrows():
if remaining_qty <= 0:
break
fill_qty = min(remaining_qty, row['quantity'])
total_cost += fill_qty * row['price']
if best_price == 0:
best_price = row['price']
remaining_qty -= fill_qty
if remaining_qty > 0:
print(f"⚠️ Ordre partiellement rempli: {quantity - remaining_qty:.4f}/{quantity}")
avg_price = total_cost / (quantity - remaining_qty) if remaining_qty < quantity else best_price
slippage_bps = abs(avg_price - best_price) / best_price * 10000 if best_price > 0 else 0
return slippage_bps
def run_backtest(self, df: pd.DataFrame, strategy_func: callable) -> BacktestResult:
"""
Exécute le backtest sur les données orderbook.
Args:
df: DataFrame avec données orderbook normalisées
strategy_func: Fonction qui retourne des signaux de trading
"""
print(f"🚀 Démarrage backtest sur {len(df)} enregistrements...")
capital = self.config.initial_capital
self.equity_curve = [capital]
# Résolution minute par minute
df['minute'] = df['timestamp'].dt.floor('1min')
minutes = df['minute'].unique()
for i, minute in enumerate(minutes):
minute_data = df[df['minute'] == minute]
if len(minute_data) == 0:
continue
# Obtenir le orderbook à ce moment
ob_snapshot = minute_data.pivot_table(
index='price',
columns='side',
values='quantity',
aggfunc='sum'
).fillna(0)
# Générer signal via la stratégie
signal = strategy_func(minute_data, self.current_position)
if signal and not self.current_position:
# Ouvrir une position
entry_price = minute_data.iloc[-1]['price']
max_qty = capital * 0.1 / entry_price # 10% du capital max
slippage_bps = self.calculate_slippage(
minute_data, signal['side'], max_qty
)
self.current_position = {
'entry_time': minute,
'entry_price': entry_price * (1 + slippage_bps/10000 if signal['side'] == 'buy' else -slippage_bps/10000),
'quantity': max_qty,
'side': signal['side']
}
elif self.current_position and signal and signal.get('close'):
# Fermer la position
exit_price = minute_data.iloc[-1]['price']
slippage_bps = self.calculate_slippage(
minute_data, 'sell' if self.current_position['side'] == 'buy' else 'buy',
self.current_position['quantity']
)
pnl = self._calculate_pnl(
self.current_position, exit_price, slippage_bps
)
trade = Trade(
timestamp=minute,
symbol=minute_data.iloc[0]['symbol'],
side=self.current_position['side'],
entry_price=self.current_position['entry_price'],
exit_price=exit_price,
quantity=self.current_position['quantity'],
pnl=pnl['pnl'],
pnl_pct=pnl['pnl_pct'],
commission=self.config.commission_rate * self.current_position['quantity'] *
(self.current_position['entry_price'] + exit_price),
duration_minutes=(minute - self.current_position['entry_time']).total_seconds() / 60
)
self.trades.append(trade)
capital += pnl['pnl']
self.equity_curve.append(capital)
self.current_position = None
if i % 10000 == 0:
print(f" Progression: {i}/{len(minutes)} minutes...")
return self._compute_results(capital)
def _calculate_pnl(self, position: dict, exit_price: float,
slippage_bps: float) -> dict:
"""Calcule le PnL d'une trade"""
if position['side'] == 'buy':
entry = position['entry_price']
exit_adj = exit_price * (1 - slippage_bps/10000)
pnl_pct = (exit_adj - entry) / entry
else:
entry = position['entry_price']
exit_adj = exit_price * (1 + slippage_bps/10000)
pnl_pct = (entry - exit_adj) / entry
pnl = position['quantity'] * position['entry_price'] * pnl_pct
return {'pnl': pnl, 'pnl_pct': pnl_pct}
def _compute_results(self, final_capital: float) -> BacktestResult:
"""Calcule les métriques finales du backtest"""
if not self.trades:
return BacktestResult(
total_trades=0, winning_trades=0, losing_trades=0,
win_rate=0, total_pnl=0, max_drawdown=0,
sharpe_ratio=0, avg_trade_duration=0, profit_factor=0,
trades=[]
)
winning = [t for t in self.trades if t.pnl > 0]
losing = [t for t in self.trades if t.pnl <= 0]
total_wins = sum(t.pnl for t in winning)
total_losses = abs(sum(t.pnl for t in losing))
# Max drawdown sur equity curve
equity = np.array(self.equity_curve)
running_max = np.maximum.accumulate(equity)
drawdowns = (running_max - equity) / running_max
max_dd = np.max(drawdowns)
# Sharpe ratio (simplifié)
returns = np.diff(self.equity_curve) / self.equity_curve[:-1]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252 * 1440) if np.std(returns) > 0 else 0
return BacktestResult(
total_trades=len(self.trades),
winning_trades=len(winning),
losing_trades=len(losing),
win_rate=len(winning) / len(self.trades) if self.trades else 0,
total_pnl=final_capital - self.config.initial_capital,
max_drawdown=max_dd,
sharpe_ratio=sharpe,
avg_trade_duration=np.mean([t.duration_minutes for t in self.trades]),
profit_factor=total_wins / total_losses if total_losses > 0 else float('inf'),
trades=self.trades
)
Intégration avec HolySheep AI pour analyse des résultats
class HolySheepAnalyzer:
"""Utilise HolySheep AI pour analyser les résultats de backtest"""
def __init__(self, config):
self.config = config
self.base_url = "https://api.holysheep.ai/v1"
async def generate_report(self, results: BacktestResult) -> str:
"""Génère un rapport détaillé via HolySheep AI"""
import aiohttp
prompt = f"""Analyse ce rapport de backtest et donne des recommandations:
Résultats:
- Total trades: {results.total_trades}
- Win rate: {results.win_rate:.2%}
- PnL total: ${results.total_pnl:.2f}
- Max drawdown: {results.max_drawdown:.2%}
- Sharpe ratio: {results.sharpe_ratio:.2f}
- Profit factor: {results.profit_factor:.2f}
Donne-moi:
1. Évaluation de la stratégie
2. Points d'amélioration
3. Risk management recommendations
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=self.config.get_headers(),
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 1000
}
) as resp:
if resp.status == 200:
data = await resp.json()
return data['choices'][0]['message']['content']
else:
error = await resp.text()
raise Exception(f"Erreur HolySheep: {error}")
Exemple d'utilisation complète
async def run_full_backtest():
from config.settings import BacktestConfig, HolySheepConfig
backtest_config = BacktestConfig(
symbols=["btcusdt"],
start_date="2025-06-01",
end_date="2025-12-01",
initial_capital=10000.0
)
holy_config = HolySheepConfig()
analyzer = HolySheepAnalyzer(holy_config)
backtester = OrderbookBacktester(backtest_config)
# Charger données (remplacez par votre chemin)
# df = backtester.load_orderbook_data("data/processed/btcusdt_orderbook.parquet")
# Stratégie exemple: simple mean reversion
def strategy(data, position):
if position:
return {'close': True} # Fermer après 1 minute
return {'side': 'buy', 'size': 0.1}
# results = backtester.run_backtest(df, strategy)
# report = await analyzer.generate_report(results)
# print(report)
if __name__ == "__main__":
asyncio.run(run_full_backtest())
Intégration Avancée : Pipeline Complet de Production
Pour les stratégies en production, j'ai développé un pipeline robuste qui combine Tardis.dev pour la collecte brute et HolySheep AI pour l'intelligence artificielle et l'optimisation. Le coût total mensuel pour une stratégie active est d'environ $45 avec HolySheep contre $320+ avec une solution traditionnelle.
# src/processors/data_normalizer.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List
import json
class OrderbookNormalizer:
"""
Normalise les données orderbook de différentes sources
vers un format unifié pour backtesting.
"""
# Formats supportés
SUPPORTED_FORMATS = ['tardis', 'holy', 'binance_raw']
def __init__(self):
self.normalized_cache: Dict[str, pd.DataFrame] = {}
def normalize_tardis(self, raw_data: List[dict]) -> pd.DataFrame:
"""Normalise les données au format Tardis"""
records = []
for item in raw_data:
ts = datetime.fromtimestamp(item.get('timestamp', 0) / 1000)
book = item.get('data', {}).get('book', {})
symbol = item.get('symbol', '')
for price, qty in book.get('bids', []):
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'bid',
'price': float(price),
'quantity': float(qty),
'source': 'tardis'
})
for price, qty in book.get('asks', []):
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'ask',
'price': float(price),
'quantity': float(qty),
'source': 'tardis'
})
df = pd.DataFrame(records)
return self._add_derived_features(df)
def normalize_holy(self, raw_data: List[dict]) -> pd.DataFrame:
"""Normalise les données au format HolySheep AI"""
# HolySheep utilise un format légèrement différent
records = []
for item in raw_data:
ts = datetime.fromisoformat(item.get('time', '').replace('Z', '+00:00'))
bids = item.get('b', [])
asks = item.get('a', [])
symbol = item.get('s', '')
for price, qty in bids:
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'bid',
'price': float(price),
'quantity': float(qty),
'source': 'holy'
})
for price, qty in asks:
records.append({
'timestamp': ts,
'symbol': symbol,
'side': 'ask',
'price': float(price),
'quantity': float(qty),
'source': 'holy'
})
df = pd.DataFrame(records)
return self._add_derived_features(df)
def _add_derived_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Ajoute les features dérivées pour analyse"""
# Mid price
df_pivot = df.pivot_table(
index=['timestamp', 'symbol'],
columns='side',
values='price',
aggfunc='first'
).reset_index()
df_pivot.columns = ['timestamp', 'symbol', 'ask_price', 'bid_price']
df_pivot['mid_price'] = (df_pivot['bid_price'] + df_pivot['ask_price']) / 2
# VWAP approximatif
df_merged = df.merge(df_pivot, on=['timestamp', 'symbol'])
df_merged['vwap_approx'] = df_merged['mid_price']
# Imbalance ratio
bid_qty = df[df['side'] == 'bid'].groupby(['timestamp', 'symbol'])['quantity'].sum()
ask_qty = df[df['side'] == 'ask'].groupby(['timestamp', 'symbol'])['quantity'].sum()
imbalance = (bid_qty - ask_qty) / (bid_qty + ask_qty + 1e-10)
imbalance_df = imbalance.reset_index()
imbalance_df.columns = ['timestamp', 'symbol', 'imbalance']
df_merged = df_merged.merge(imbalance_df, on=['timestamp', 'symbol'], how='left')
df_merged['imbalance'] = df_merged['imbalance'].fillna(0)
# Microprice (prix ajusté par l'imbalance)
df_merged['microprice'] = (
df_merged['mid_price'] +
df_merged['imbalance'] * (df_merged['ask_price'] - df_merged['bid_price']) / 2
)
return df_merged.sort_values(['timestamp', 'symbol', 'side'])
def compute_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Calcule des features avancées pour machine learning
"""
df = df.copy()
# Group by symbol and compute rolling features
for symbol in df['symbol'].unique():
mask = df['symbol'] == symbol
symbol_df = df[mask].copy()
# Spread en basis points
symbol_df['spread_bps'] = (
(symbol_df['ask_price'] - symbol_df['bid_price']) /
symbol_df['mid_price'] * 10000
)
# Book depth ratio (top 5 vs total)
# ... (feature engineering avancé)
df.loc[mask] = symbol_df
return df
def export_for_ml(self, df: pd.DataFrame, output_path: str):
"""Exporte au format optimisé pour ML"""
features_df = self.compute_features(df)
# Pivot final pour序列模型
pivot_df = features_df.pivot_table(
index='timestamp',
columns='symbol',
values=['mid_price', 'imbalance', 'spread_bps', 'microprice'],
aggfunc='last'
)
pivot_df.to_parquet(output_path)
print(f"✅ Export ML: {len(pivot_df)} lignes vers {output_path}")
src/collectors/binance_collector.py - Alternative directe Binance
class BinanceCollector:
"""
Collecteur direct via API Binance (fallback gratuit)
Rate limit: 1200 requests/minute, 5 orders/second
"""
BASE_URL = "https://fapi.binance.com"
def __init__(self):
self.session_active = False
async def get_orderbook_snapshot(self, symbol: str, limit: int = 1000) -> dict:
"""
Récupère un snapshot du orderbook L2
Gratuit mais limité à 1000 niveaux
"""
import aiohttp
url = f"{self.BASE_URL}/fapi/v1/depth"
params = {"symbol": symbol.upper(), "limit": limit}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
if resp.status == 200:
return await resp