Introduction aux données Tick OKX et au backtesting quantitatif

Le marché des cryptomonnaies génère des millions d'événements par seconde. Pour un trader quantitatif, la reconstruction précise d'un carnet d'ordres de niveau 2 (L2 Order Book) à partir de données Tick historiques représente un avantage compétitif majeur. Dans cet article, je vous présente une méthodologie complète utilisant Tardis.dev comme source de données historiques, combinée à une infrastructure de calcul haute performance via HolySheep AI pour vos besoins en intelligence artificielle.

Tableau comparatif : HolySheep vs API officielle vs services relais

Critère HolySheep AI API OpenAI / Anthropic officielle Services relais tiers
Latence moyenne <50ms 150-300ms 80-200ms
Coût par 1M tokens DeepSeek V3.2: $0.42 GPT-4.1: $8 Variable, +20-40%
Économie vs officiel 85%+ Référence 5-30%
Paiements acceptés WeChat Pay, Alipay, USDT Carte internationale uniquement Limité
Crédits gratuits Oui, dès l'inscription $5 promotionnel Rare
Support Python natif ✅ Complet ✅ Complet ⚠️ Variable

Pour qui / pour qui ce n'est pas fait

✓ Ce tutoriel est fait pour :

✗ Ce tutoriel n'est pas fait pour :

Tarification et ROI

Analysons le retour sur investissement de cette architecture pour un projet de backtesting typique :

Composant Coût mensuel estimé HolySheep alternatif
API LLM (analyse de données) $200-500 (OpenAI) $10-25 (HolySheep)
Stockage données Tick $50-100 $50-100
Compute (backtesting) $100-300 $100-300
Total mensuel $350-900 $160-425
Économie annuelle - $2,280-5,700

Récupération des données Tick OKX via Tardis.dev

Tardis.dev offre un accès historique aux données d'échange de cryptomonnaies avec une qualité de données exceptionnelle. Leur API permet de récupérer les trades, order books et Funding Rates pour OKX.


#!/usr/bin/env python3
"""
Récupération des données Tick OKX via Tardis.dev
Documentation: https://tardis.dev
"""

import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class TardisDataFetcher:
    """Classe pour récupérer les données Tick depuis Tardis.dev"""
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def get_trades(
        self,
        exchange: str = "okx",
        symbol: str = "BTC-USDT-SWAP",
        from_date: str = "2024-01-01",
        to_date: str = "2024-01-02",
        limit: int = 100000
    ) -> List[Dict]:
        """
        Récupère les trades historiques pour un symbole donné.
        
        Args:
            exchange: Exchange cible (okx, binance, etc.)
            symbol: Symbole de trading
            from_date: Date de début (YYYY-MM-DD)
            to_date: Date de fin (YYYY-MM-DD)
            limit: Nombre maximum de records par requête
            
        Returns:
            Liste de dictionnaires contenant les données de trade
        """
        url = f"{self.BASE_URL}/historical/trades"
        
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "from": from_date,
            "to": to_date,
            "limit": limit,
            "format": "json"
        }
        
        print(f"📥 Récupération des trades {symbol} du {from_date} au {to_date}...")
        
        all_trades = []
        has_more = True
        offset = 0
        
        while has_more:
            params["offset"] = offset
            response = self.session.get(url, params=params, timeout=60)
            response.raise_for_status()
            
            data = response.json()
            trades = data.get("data", [])
            all_trades.extend(trades)
            
            has_more = data.get("hasMore", False)
            offset += len(trades)
            
            print(f"   → {len(trades)} trades récupérés (total: {offset})")
            
            if not has_more:
                break
        
        print(f"✅ {len(all_trades)} trades récupérés au total")
        return all_trades
    
    def get_orderbook_snapshots(
        self,
        exchange: str = "okx",
        symbol: str = "BTC-USDT-SWAP",
        from_date: str = "2024-01-01",
        to_date: str = "2024-01-02",
        limit: int = 50000
    ) -> List[Dict]:
        """
        Récupère les snapshots du carnet d'ordres (order book snapshots).
        
        Ces données sont essentielles pour reconstruire le L2 Order Book.
        """
        url = f"{self.BASE_URL}/historical/orderbook-snapshots"
        
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "from": from_date,
            "to": to_date,
            "limit": limit,
            "format": "json"
        }
        
        print(f"📋 Récupération des snapshots order book {symbol}...")
        
        response = self.session.get(url, params=params, timeout=60)
        response.raise_for_status()
        
        data = response.json()
        snapshots = data.get("data", [])
        
        print(f"✅ {len(snapshots)} snapshots récupérés")
        return snapshots


=== Utilisation ===

if __name__ == "__main__": # Remplacez par votre clé API Tardis.dev TARDIS_API_KEY = "your_tardis_api_key" fetcher = TardisDataFetcher(api_key=TARDIS_API_KEY) # Exemple: récupérer 1 heure de données BTC-USDT trades = fetcher.get_trades( symbol="BTC-USDT-SWAP", from_date="2024-06-01T00:00:00Z", to_date="2024-06-01T01:00:00Z" ) # Afficher un échantillon if trades: print(f"\n📊 Échantillon du premier trade:") print(json.dumps(trades[0], indent=2))

Reconstruction du L2 Order Book à partir des données Tick

La reconstruction précise d'un order book L2 nécessite de traiter les mises à jour incrémentales (deltas) en les appliquant aux snapshots. Voici une implémentation robuste :


#!/usr/bin/env python3
"""
Reconstruction du L2 Order Book OKX à partir des données Tick
Intégration possible avec HolySheep AI pour l'analyse de patterns
"""

import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from collections import defaultdict
import heapq
from datetime import datetime

@dataclass
class OrderBookLevel:
    """Représente un niveau de prix dans le carnet d'ordres"""
    price: float
    size: float
    order_count: int = 0
    
    def is_empty(self) -> bool:
        return self.size <= 0

class L2OrderBookReconstructor:
    """
    Reconstructeur de carnet d'ordres L2 à partir de données Tick.
    
    Utilise une structure de données optimisée pour les opérations
    d'ajout/suppression de prix avec O(log n) de complexité.
    """
    
    def __init__(self, symbol: str = "BTC-USDT-SWAP"):
        self.symbol = symbol
        # Heaps pour bids (max-heap simulé avec valeurs négatives)
        self.bids: Dict[float, OrderBookLevel] = {}
        # Heaps pour asks (min-heap)
        self.asks: Dict[float, OrderBookLevel] = {}
        
        # Historique pour le backtesting
        self.snapshots_history: List[Dict] = []
        self.spread_history: List[float] = []
        self.mid_price_history: List[float] = []
        
        # Statistiques
        self.total_updates = 0
        self.last_update_time = None
    
    def apply_snapshot(self, snapshot: Dict) -> None:
        """
        Applique un snapshot complet du carnet d'ordres.
        Utilisé lors de l'initialisation ou de la resynchronisation.
        """
        timestamp = snapshot.get("timestamp", 0)
        self.last_update_time = timestamp
        
        # Extraire les asks et bids du snapshot OKX
        asks_data = snapshot.get("asks", [])
        bids_data = snapshot.get("bids", [])
        
        # Réinitialiser et reconstruire
        self.asks.clear()
        self.bids.clear()
        
        for price, size in asks_data:
            self.asks[float(price)] = OrderBookLevel(
                price=float(price),
                size=float(size)
            )
        
        for price, size in bids_data:
            self.bids[float(price)] = OrderBookLevel(
                price=float(price),
                size=float(size)
            )
        
        self._record_snapshot()
    
    def apply_update(self, update: Dict) -> None:
        """
        Applique une mise à jour delta au carnet d'ordres.
        
        Format OKX pour les mises à jour:
        {
            "timestamp": 1234567890000,
            "asks": [[price, size, orders_count], ...],
            "bids": [[price, size, orders_count], ...]
        }
        """
        timestamp = update.get("timestamp", 0)
        self.last_update_time = timestamp
        self.total_updates += 1
        
        # Traiter les mises à jour des asks
        asks_data = update.get("asks", [])
        for item in asks_data:
            price = float(item[0])
            size = float(item[1])
            
            if size == 0:
                # Supprimer le niveau de prix
                if price in self.asks:
                    del self.asks[price]
            else:
                # Ajouter ou mettre à jour
                self.asks[price] = OrderBookLevel(
                    price=price,
                    size=size,
                    order_count=int(item[2]) if len(item) > 2 else 0
                )
        
        # Traiter les mises à jour des bids
        bids_data = update.get("bids", [])
        for item in bids_data:
            price = float(item[0])
            size = float(item[1])
            
            if size == 0:
                if price in self.bids:
                    del self.bids[price]
            else:
                self.bids[price] = OrderBookLevel(
                    price=price,
                    size=size,
                    order_count=int(item[2]) if len(item) > 2 else 0
                )
        
        self._record_snapshot()
    
    def _record_snapshot(self) -> None:
        """Enregistre un snapshot pour le backtesting"""
        best_bid = self.get_best_bid()
        best_ask = self.get_best_ask()
        
        if best_bid and best_ask:
            spread = best_ask.price - best_bid.price
            mid_price = (best_ask.price + best_bid.price) / 2
            
            self.spread_history.append(spread)
            self.mid_price_history.append(mid_price)
            
            self.snapshots_history.append({
                "timestamp": self.last_update_time,
                "best_bid": best_bid.price,
                "best_ask": best_ask.price,
                "spread": spread,
                "mid_price": mid_price,
                "bid_depth": sum(o.size for o in self.bids.values()),
                "ask_depth": sum(o.size for o in self.asks.values()),
                "total_bids": len(self.bids),
                "total_asks": len(self.asks)
            })
    
    def get_best_bid(self) -> Optional[OrderBookLevel]:
        """Retourne le meilleur bid (prix le plus élevé)"""
        if not self.bids:
            return None
        return max(self.bids.values(), key=lambda x: x.price)
    
    def get_best_ask(self) -> Optional[OrderBookLevel]:
        """Retourne le meilleur ask (prix le plus bas)"""
        if not self.asks:
            return None
        return min(self.asks.values(), key=lambda x: x.price)
    
    def get_spread(self) -> Optional[float]:
        """Calcule le spread bid-ask actuel"""
        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]:
        """Calcule le prix médian"""
        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 = 10) -> Dict:
        """
        Retourne la profondeur du marché pour les N meilleurs niveaux.
        """
        sorted_bids = sorted(self.bids.values(), key=lambda x: -x.price)[:levels]
        sorted_asks = sorted(self.asks.values(), key=lambda x: x.price)[:levels]
        
        bid_cumulative = 0
        bid_levels = []
        for level in sorted_bids:
            bid_cumulative += level.size
            bid_levels.append({
                "price": level.price,
                "size": level.size,
                "cumulative": bid_cumulative
            })
        
        ask_cumulative = 0
        ask_levels = []
        for level in sorted_asks:
            ask_cumulative += level.size
            ask_levels.append({
                "price": level.price,
                "size": level.size,
                "cumulative": ask_cumulative
            })
        
        return {
            "bids": bid_levels,
            "asks": ask_levels,
            "bid_total": sum(o.size for o in self.bids.values()),
            "ask_total": sum(o.size for o in self.asks.values())
        }
    
    def get_imbalance(self, levels: int = 5) -> float:
        """
        Calcule le déséquilibre du carnet d'ordres.
        Valeur entre -1 (tout côté bid) et 1 (tout côté ask).
        """
        bid_volume = 0
        ask_volume = 0
        
        sorted_bids = sorted(self.bids.values(), key=lambda x: -x.price)[:levels]
        sorted_asks = sorted(self.asks.values(), key=lambda x: x.price)[:levels]
        
        for level in sorted_bids:
            bid_volume += level.size * (levels - sorted_bids.index(level))
        
        for level in sorted_asks:
            ask_volume += level.size * (levels - sorted_asks.index(level))
        
        total = bid_volume + ask_volume
        if total == 0:
            return 0
        
        return (bid_volume - ask_volume) / total
    
    def to_dataframe(self) -> pd.DataFrame:
        """Convertit l'historique des snapshots en DataFrame pandas"""
        return pd.DataFrame(self.snapshots_history)
    
    def print_state(self, levels: int = 5) -> None:
        """Affiche les N meilleurs niveaux du carnet d'ordres"""
        print(f"\n📊 Order Book State - {self.symbol}")
        print("=" * 60)
        
        print("\n🔴 ASKS (Sell Orders):")
        print(f"{'Prix':>12} | {'Taille':>12} | {'Cumul':>12}")
        print("-" * 40)
        
        sorted_asks = sorted(self.asks.values(), key=lambda x: x.price)[:levels]
        cumulative = 0
        for level in sorted_asks:
            cumulative += level.size
            print(f"{level.price:>12.2f} | {level.size:>12.4f} | {cumulative:>12.4f}")
        
        print("\n🟢 BIDS (Buy Orders):")
        print(f"{'Prix':>12} | {'Taille':>12} | {'Cumul':>12}")
        print("-" * 40)
        
        sorted_bids = sorted(self.bids.values(), key=lambda x: -x.price)[:levels]
        cumulative = 0
        for level in sorted_bids:
            cumulative += level.size
            print(f"{level.price:>12.2f} | {level.size:>12.4f} | {cumulative:>12.4f}")
        
        spread = self.get_spread()
        mid = self.get_mid_price()
        print(f"\n📈 Spread: {spread:.2f} | Mid: {mid:.2f} | Updates: {self.total_updates}")


=== Intégration avec HolySheep AI pour analyse ===

class OrderBookAnalyzer: """ Analyseur de carnet d'ordres utilisant l'IA HolySheep. Permet de détecter des patterns et anomalies dans le order book. """ def __init__(self, holysheep_api_key: str): self.api_key = holysheep_api_key self.base_url = "https://api.holysheep.ai/v1" def analyze_pattern(self, orderbook_df: pd.DataFrame) -> Dict: """ Utilise l'IA pour analyser les patterns du order book. Inclut automatiquement les avantages HolySheep: - Latence < 50ms - Coût DeepSeek V3.2: $0.42/1M tokens - Taux de change ¥1=$1 """ # Préparer les données pour l'analyse summary = { "total_snapshots": len(orderbook_df), "avg_spread": orderbook_df["spread"].mean(), "spread_volatility": orderbook_df["spread"].std(), "price_range": orderbook_df["mid_price"].max() - orderbook_df["mid_price"].min(), "imbalance_events": (orderbook_df["bid_depth"] - orderbook_df["ask_depth"]).abs().mean() } prompt = f""" Analyse ce carnet d'ordres BTC-USDT et identifie: 1. Les périodes de haute volatilité du spread 2. Les moments de déséquilibre significatif 3. Les patterns de liquidité anormaux Données résumées: - Snapshots: {summary['total_snapshots']} - Spread moyen: {summary['avg_spread']:.2f} - Volatilité du spread: {summary['spread_volatility']:.2f} - Range de prix: {summary['price_range']:.2f} """ # Appel API HolySheep avec DeepSeek V3.2 (économique) response = self._call_holysheep(prompt) summary["ai_analysis"] = response return summary def _call_holysheep(self, prompt: str) -> str: """Appel interne à l'API HolySheep""" import requests response = requests.post( f"{self.base_url}/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" }, json={ "model": "deepseek-v3.2", "messages": [ {"role": "system", "content": "Tu es un analyste quantitatif expert en crypto."}, {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 500 }, timeout=30 ) if response.status_code == 200: return response.json()["choices"][0]["message"]["content"] else: raise Exception(f"Erreur API HolySheep: {response.status_code}")

=== Backtesting Framework ===

class SimpleBacktester: """ Framework de backtesting basique pour stratégies basées sur le order book. """ def __init__(self, initial_balance: float = 10000.0): self.initial_balance = initial_balance self.balance = initial_balance self.position = 0.0 self.trades = [] self.equity_curve = [] def run_mid_price_crossover( self, df: pd.DataFrame, fast_ma: int = 5, slow_ma: int = 20 ) -> Dict: """ Stratégie de croisement de moyennes mobiles du prix médian. Achat quand MA rapide > MA lente Vente quand MA rapide < MA lente """ df = df.copy() df["fast_ma"] = df["mid_price"].rolling(fast_ma).mean() df["slow_ma"] = df["mid_price"].rolling(slow_ma).mean() position = 0 for idx, row in df.iterrows(): if pd.isna(row["fast_ma"]) or pd.isna(row["slow_ma"]): continue # Signal d'achat if row["fast_ma"] > row["slow_ma"] and position <= 0: # Acheter pour 50% du balance buy_amount = self.balance * 0.5 / row["mid_price"] self.balance -= buy_amount * row["mid_price"] self.position += buy_amount position = 1 self.trades.append({ "timestamp": row["timestamp"], "type": "BUY", "price": row["mid_price"], "size": buy_amount }) # Signal de vente elif row["fast_ma"] < row["slow_ma"] and position > 0: # Vendre toute la position sell_value = self.position * row["mid_price"] self.balance += sell_value self.trades.append({ "timestamp": row["timestamp"], "type": "SELL", "price": row["mid_price"], "size": self.position }) self.position = 0 position = -1 # Enregistrer l'equity total_equity = self.balance + self.position * row["mid_price"] self.equity_curve.append(total_equity) return self.get_performance_metrics() def get_performance_metrics(self) -> Dict: """Calcule les métriques de performance du backtest""" final_equity = self.balance + self.position * (self.equity_curve[-1] if self.equity_curve else 0) total_return = (final_equity - self.initial_balance) / self.initial_balance * 100 # Calcul du max drawdown equity_series = pd.Series(self.equity_curve) rolling_max = equity_series.expanding().max() drawdowns = (equity_series - rolling_max) / rolling_max * 100 max_drawdown = drawdowns.min() # Nombre de trades buy_trades = [t for t in self.trades if t["type"] == "BUY"] sell_trades = [t for t in self.trades if t["type"] == "SELL"] return { "initial_balance": self.initial_balance, "final_equity": final_equity, "total_return_pct": total_return, "max_drawdown_pct": max_drawdown, "total_trades": len(self.trades), "buy_trades": len(buy_trades), "sell_trades": len(sell_trades), "win_rate": len(sell_trades) / max(len(buy_trades), 1) * 100 }

=== Exemple d'utilisation complète ===

if __name__ == "__main__": # Initialiser le reconstructeur ob_reconstructor = L2OrderBookReconstructor(symbol="BTC-USDT-SWAP") # Simuler avec des données d'exemple print("🔧 Test du reconstructeur de Order Book L2") print("=" * 60) # Ajouter des niveaux initiaux test_snapshot = { "timestamp": 1717200000000, "asks": [ [67450.0, 2.5], [67455.0, 1.8], [67460.0, 3.2], [67470.0, 5.0], [67480.0, 8.0] ], "bids": [ [67445.0, 3.0], [67440.0, 2.0], [67435.0, 4.5], [67430.0, 6.0], [67425.0, 10.0] ] } ob_reconstructor.apply_snapshot(test_snapshot) ob_reconstructor.print_state(levels=5) # Tester l'analyse d'imbalance imbalance = ob_reconstructor.get_imbalance(levels=3) print(f"\n📊 Déséquilibre du book (niveaux 3): {imbalance:.4f}") # Tester le profondeur depth = ob_reconstructor.get_depth(levels=3) print(f"\n📈 Profondeur bids: {depth['bid_total']:.4f} BTC") print(f"📉 Profondeur asks: {depth['ask_total']:.4f} BTC") # Analyser avec HolySheep AI (optionnel) HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" try: analyzer = OrderBookAnalyzer(holysheep_api_key=HOLYSHEEP_API_KEY) print("\n🤖 Analyse HolySheep disponible (nécessite crédit)") except Exception as e: print(f"\n⚠️ HolySheep non configuré: {e}") print("\n✅ Test terminé avec succès!")

Exécution du backtest avec données réelles

Maintenant que nous avons construit notre framework de reconstruction et d'analyse, voyons comment l'intégrer avec des données réelles de Tardis.dev :


#!/usr/bin/env python3
"""
Backtest complet avec données OKX réelles via Tardis.dev
"""

import pandas as pd
import numpy as np
import json
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

Imports locaux (des scripts précédents)

from tardis_fetcher import TardisDataFetcher from orderbook_reconstructor import L2OrderBookReconstructor, SimpleBacktester class OKXBacktester: """ Classe de backtesting complète pour OKX utilisant les données Tick. """ def __init__(self, tardis_api_key: str, holysheep_api_key: str = None): self.tardis = TardisDataFetcher(api_key=tardis_api_key) self.holysheep_api_key = holysheep_api_key self.orderbook = L2OrderBookReconstructor() self.backtester = SimpleBacktester(initial_balance=10000.0) def load_and_reconstruct(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame: """ Charge les données depuis Tardis.dev et reconstruit le L2 Order Book. Args: symbol: Symbole de trading (ex: "BTC-USDT-SWAP") start_date: Date de début (YYYY-MM-DD) end_date: Date de fin (YYYY-MM-DD) Returns: DataFrame avec les snapshots du order book """ print(f"📥 Chargement des données pour {symbol}") print(f" Période: {start_date} → {end_date}") # Récupérer les snapshots du order book snapshots = self.tardis.get_orderbook_snapshots( exchange="okx", symbol=symbol, from_date=start_date, to_date=end_date ) # Reconstruire le order book print(f"🔧 Reconstruction du L2 Order Book...") for snapshot in snapshots: self.orderbook.apply_snapshot(snapshot) # Convertir en DataFrame df = self.orderbook.to_dataframe() # Ajouter des features techniques df = self._add_features(df) print(f"✅ {len(df)} snapshots traités") print(f" Range de prix: {df['mid_price'].min():.2f} - {df['mid_price'].max():.2f}") print(f" Spread moyen: {df['spread'].mean():.2f}") return df def _add_features(self, df: pd.DataFrame) -> pd.DataFrame: """Ajoute des features techniques pour le backtesting""" # Moyennes mobiles df["sma_20"] = df["mid_price"].rolling(20).mean() df["sma_50"] = df["mid_price"].rolling(50).mean() df["sma_200"] = df["mid_price"].rolling(200).mean() # Volatilité df["volatility_20"] = df["mid_price"].rolling(20).std() df["volatility_50"] = df["mid_price"].rolling(50).std() # Ratios de volume df["bid_ask_ratio"] = df["bid_depth"] / df["ask_depth"].replace(0, np.nan) df["volume_imbalance"] = (df["bid_depth"] - df["ask_depth"]) / (df["bid_depth"] + df["ask_depth"]).replace(0, np.nan) # Momentum df["returns"] = df["mid_price"].pct_change() df["momentum_20"] = df["mid_price"].pct_change(20) # Support et résistance (simplifié) df["rolling_high"] = df["mid_price"].rolling(50).max() df["rolling_low"] = df["mid_price"].rolling(50).min() df["price_position"] = (df["mid_price"] - df["rolling_low"]) / (df["rolling_high"] - df["rolling_low"]).replace(0, 0.5) return df def run_strategy(self, df: pd.DataFrame, strategy: str = "ma_crossover") -> Dict: """ Exécute une stratégie de trading sur les données. Stratégies disponibles: - "ma_crossover": Croisement de moyennes mobiles - "imbalance": Stratégie basée sur le déséquilibre du book - "breakout": Stratégie de breakout """ print(f"\n🎯 Exécution de la stratégie: {strategy}") if strategy == "ma_crossover": metrics = self.backtester.run_mid_price_crossover(df, fast_ma=20, slow_ma=50) elif strategy == "imbalance": metrics = self._run_imbalance_strategy(df) elif strategy == "breakout":