Bonjour à tous, je suis Thomas, développeur quantitatif et trader algorithmique depuis 2019. Aujourd'hui, je vais partager avec vous mon retour d'expérience complet sur la récupération des données historiques d'options BTC via l'API Tardis, un sujet que j'ai dû maîtriser récemment pour alimenter mes modèles de pricing de volatilité. Si vous cherchez à télécharger des données d'options Deribit Bitcoin fiables et à les traiter efficacement en Python, cet article est fait pour vous.

为什么选择Tardis API获取Deribit数据

Dans mon travail quotidien d'analyse quantitative sur les marchés crypto, j'ai testé plusieurs sources de données d'options BTC. Tardis API s'est imposé comme la solution la plus robuste pour plusieurs raisons techniques précises :

Pour ceux qui utilisent ces données avec des modèles d'IA, notez que HolySheep AI offre des APIs GPT-4.1 et Claude Sonnet avec une latence inférieure à 50ms et un taux de change avantageux de ¥1=$1, idéal pour analyser ces datasets volumineux.

前置条件与API密钥获取

Avant de commencer, préparez votre environnement. Voici ce dont vous aurez besoin :

# Installation des dépendances Python
pip install requests pandas python-dotenv


Structure recommandée du projet

project/
├── config.py          # Configuration API
├── download_data.py   # Script principal
├── process_csv.py     # Traitement CSV
├── data/              # Données téléchargées
└── .env               # Clés API (ne pas commiter)

Tardis API核心端点详解

L'API Tardis pour Deribit propose plusieurs endpoints spécialisés. Pour les données historiques d'options BTC, les endpoints principaux sont :

获取期权行情数据

# Endpoint de base pour les options Deribit BTC
BASE_URL = "https://api.tardis.dev/v1/derivatives/deribit"


Exemple de requête pour les options BTC-PERPETUAL

import requests
import os

TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")

def get_btc_options_summary(start_date: str, end_date: str):
    """
    Récupère le résumé des options BTC pour une période donnée.
    
    Args:
        start_date: Format ISO 8601 (ex: "2024-01-01T00:00:00Z")
        end_date: Format ISO 8601 (ex: "2024-12-31T23:59:59Z")
    
    Returns:
        dict: Réponse JSON contenant les données d'options
    """
    url = f"{BASE_URL}/options/summary"
    params = {
        "api_key": TARDIS_API_KEY,
        "instrument": "BTC-PERPETUAL",  # Options sur BTC perpetual
        "start_date": start_date,
        "end_date": end_date,
        "format": "json"
    }
    
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    
    return response.json()


Utilisation

try:
    data = get_btc_options_summary(
        start_date="2026-03-01T00:00:00Z",
        end_date="2026-04-01T00:00:00Z"
    )
    print(f"Données récupérées : {len(data)} enregistrements")
except requests.exceptions.RequestException as e:
    print(f"Erreur de requête : {e}")

获取完整Tick数据

import requests
import json
from datetime import datetime

def download_option_ticks(instrument_name: str, date: str, output_file: str):
    """
    Télécharge les ticks complets pour un instrument d'option spécifique.
    
    Exemple d'instrument_name: "BTC-28MAR25-95000-P" (Put ATM mars 2025)
    
    Args:
        instrument_name: Nom exact de l'instrument Deribit
        date: Date au format YYYY-MM-DD
        output_file: Chemin du fichier de sortie JSON
    """
    url = f"{BASE_URL}/ticks"
    params = {
        "api_key": TARDIS_API_KEY,
        "instrument_name": instrument_name,
        "date": date,
        "limit": 100000,  # Max par requête
        "offset": 0
    }
    
    all_ticks = []
    total_fetched = 0
    
    while True:
        print(f"Récupération à partir de l'offset {offset}...")
        params["offset"] = offset
        
        response = requests.get(url, params=params, timeout=60)
        response.raise_for_status()
        
        ticks_batch = response.json()
        if not ticks_batch:
            break
            
        all_ticks.extend(ticks_batch)
        total_fetched += len(ticks_batch)
        offset += len(ticks_batch)
        
        # Respect du rate limit (10 req/s sur plan gratuit)
        time.sleep(0.1)
        
        # break si on atteint la limite ou si plus de données
        if len(ticks_batch) < params["limit"]:
            break
    
    # Sauvegarde en JSON
    with open(output_file, "w") as f:
        json.dump(all_ticks, f, indent=2)
    
    print(f"✓ {total_fetched} ticks sauvegardés dans {output_file}")
    return all_ticks


Exemple d'utilisation

download_option_ticks(
    instrument_name="BTC-28MAR26-95000-C",
    date="2026-04-15",
    output_file="data/btc_option_tick_20260415.json"
)

CSV转换与数据处理实战

Maintenant que nous avons les données brutes, passons au traitement pour obtenir des fichiers CSV exploitables pour l'analyse. Mon script de traitement gère la conversion, le nettoyage et l'enrichissement des données.

import pandas as pd
import json
from pathlib import Path
from typing import List, Dict

class OptionsDataProcessor:
    """Processeur de données d'options Deribit pour analyse quantitative."""
    
    def __init__(self, data_dir: str = "data"):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(exist_ok=True)
    
    def json_to_csv(self, json_file: str, csv_file: str = None) -> pd.DataFrame:
        """
        Convertit les données de ticks JSON en DataFrame pandas puis CSV.
        
        Args:
            json_file: Chemin vers le fichier JSON source
            csv_file: Chemin vers le fichier CSV de sortie (optionnel)
        
        Returns:
            DataFrame pandas avec les données nettoyées
        """
        # Chargement des données JSON
        with open(json_file, "r") as f:
            ticks = json.load(f)
        
        if not ticks:
            print("Aucun tick trouvé dans le fichier.")
            return pd.DataFrame()
        
        # Transformation en DataFrame
        df = pd.DataFrame(ticks)
        
        # Normalisation des timestamps
        if "timestamp" in df.columns:
            df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
        
        # Extraction des champs pertinents pour les options
        relevant_columns = [
            "datetime",
            "instrument_name",
            "last_price",      # Prix du dernier trade
            "best_bid_price",  # Meilleure offre d'achat
            "best_ask_price",  # Meilleure offre de vente
            "mark_price",      # Prix mark (moyen)
            "open_interest",   # Intérêt ouvert
            "volume",          # Volume de trades
            "underlying_price" # Prix du sous-jacent BTC
        ]
        
        # Filtrage des colonnes existantes
        available_columns = [col for col in relevant_columns if col in df.columns]
        df_clean = df[available_columns].copy()
        
        # Calcul du spread bid-ask en pourcentage
        if "best_bid_price" in df_clean.columns and "best_ask_price" in df_clean.columns:
            df_clean["spread_bps"] = (
                (df_clean["best_ask_price"] - df_clean["best_bid_price"]) 
                / df_clean["best_bid_price"] * 10000
            )
        
        # Calcul de la volatilité implicite si disponible
        if "mark_iv" in df.columns:
            df_clean["mark_iv"] = df["mark_iv"]
        
        # Ajout d'identifiants d'option
        df_clean["expiry"] = df_clean["instrument_name"].apply(self._extract_expiry)
        df_clean["strike"] = df_clean["instrument_name"].apply(self._extract_strike)
        df_clean["option_type"] = df_clean["instrument_name"].apply(self._extract_type)
        
        # Tri chronologique
        df_clean = df_clean.sort_values("datetime").reset_index(drop=True)
        
        # Export CSV
        if csv_file is None:
            csv_file = json_file.replace(".json", ".csv")
        
        df_clean.to_csv(csv_file, index=False)
        print(f"✓ CSV exporté : {csv_file} ({len(df_clean)} lignes)")
        
        return df_clean
    
    def _extract_expiry(self, instrument_name: str) -> str:
        """Extrait la date d'expiration du nom d'instrument."""
        import re
        match = re.search(r"\d{2}\w{3}\d{2}", instrument_name)
        return match.group(0) if match else "UNKNOWN"
    
    def _extract_strike(self, instrument_name: str) -> float:
        """Extrait le prix d'exercice du nom d'instrument."""
        import re
        match = re.search(r"(\d+)-", instrument_name)
        return float(match.group(1)) if match else 0.0
    
    def _extract_type(self, instrument_name: str) -> str:
        """Extrait le type d'option (Call ou Put)."""
        return "Put" if instrument_name.endswith("-P") else "Call"

    def merge_daily_files(self, date: str, output_file: str = None) -> pd.DataFrame:
        """
        Fusionne tous les fichiers JSON d'une journée en un seul CSV.
        
        Args:
            date: Date au format YYYY-MM-DD
            output_file: Fichier de sortie
        
        Returns:
            DataFrame fusionné
        """
        json_files = list(self.data_dir.glob(f"*{date}*.json"))
        
        if not json_files:
            print(f"Aucun fichier trouvé pour la date {date}")
            return pd.DataFrame()
        
        dfs = []
        for json_file in json_files:
            df = self.json_to_csv(str(json_file))
            if not df.empty:
                dfs.append(df)
        
        if dfs:
            merged_df = pd.concat(dfs, ignore_index=True)
            merged_df = merged_df.sort_values("datetime").reset_index(drop=True)
            
            if output_file is None:
                output_file = f"data/merged_options_{date}.csv"
            
            merged_df.to_csv(output_file, index=False)
            print(f"✓ Fichier fusionné : {output_file} ({len(merged_df)} lignes)")
            return merged_df
        
        return pd.DataFrame()


Utilisation

processor = OptionsDataProcessor(data_dir="data")


Conversion d'un fichier individuel

df = processor.json_to_csv(
    json_file="data/btc_option_tick_20260415.json",
    csv_file="data/btc_option_tick_20260415.csv"
)


Analyse rapide

print(df.describe())
print(f"\nPrix moyen BTC : ${df['underlying_price'].mean():,.2f}")
print(f"Spread moyen : {df['spread_bps'].mean():.2f} bps")

完整数据下载流程

Pour automatiser le téléchargement sur une période complète avec gestion des erreurs et progression, voici mon script principal intégré.

#!/usr/bin/env python3
"""
Script complet de téléchargement des données d'options BTC Deribit.
Usage: python download_data.py --start 2026-01-01 --end 2026-04-15
"""

import requests
import pandas as pd
import json
import time
import argparse
from datetime import datetime, timedelta
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
from dotenv import load_dotenv

load_dotenv()

TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")
BASE_URL = "https://api.tardis.dev/v1/derivatives/deribit"
OUTPUT_DIR = Path("data")
OUTPUT_DIR.mkdir(exist_ok=True)

class DeribitDataDownloader:
    """Téléchargeur de données d'options Deribit via Tardis API."""
    
    def __init__(self, api_key: str, rate_limit: float = 0.1):
        self.api_key = api_key
        self.rate_limit = rate_limit
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})
    
    def get_available_instruments(self) -> List[Dict]:
        """Récupère la liste des instruments d'options BTC disponibles."""
        url = f"{BASE_URL}/instruments"
        params = {"api_key": self.api_key, "instrument_type": "option"}
        
        response = self.session.get(url, params=params, timeout=30)
        response.raise_for_status()
        
        instruments = response.json()
        btc_options = [
            inst for inst in instruments 
            if inst.get("base_currency") == "BTC"
        ]
        print(f"Trouvé {len(btc_options)} instruments d'options BTC")
        return btc_options
    
    def download_instrument_data(
        self, 
        instrument_name: str, 
        start_date: str, 
        end_date: str
    ) -> Dict:
        """Télécharge les données pour un instrument spécifique."""
        url = f"{BASE_URL}/ticks"
        params = {
            "api_key": self.api_key,
            "instrument_name": instrument_name,
            "start_date": start_date,
            "end_date": end_date,
            "limit": 500000
        }
        
        try:
            response = self.session.get(url, params=params, timeout=120)
            
            if response.status_code == 429:
                return {"error": "rate_limited", "instrument": instrument_name}
            
            response.raise_for_status()
            data = response.json()
            
            return {
                "instrument": instrument_name,
                "tick_count": len(data),
                "data": data,
                "success": True
            }
            
        except Exception as e:
            return {"error": str(e), "instrument": instrument_name, "success": False}
    
    def batch_download(
        self, 
        instruments: List[str], 
        start_date: str, 
        end_date: str,
        max_workers: int = 3
    ):
        """Télécharge les données en parallèle avec gestion du rate limiting."""
        results = {"success": 0, "failed": 0, "rate_limited": 0}
        output_file = OUTPUT_DIR / f"deribit_btc_{start_date}_{end_date}.json"
        
        all_data = []
        
        for i, instrument in enumerate(instruments):
            print(f"[{i+1}/{len(instruments)}] {instrument}...")
            
            result = self.download_instrument_data(instrument, start_date, end_date)
            
            if result.get("success"):
                all_data.extend(result["data"])
                results["success"] += 1
            elif result.get("error") == "rate_limited":
                results["rate_limited"] += 1
                time.sleep(10)  # Attente prolongée après rate limit
            else:
                results["failed"] += 1
                print(f"  ✗ Erreur : {result.get('error')}")
            
            time.sleep(self.rate_limit)  # Rate limit standard
        
        # Sauvegarde
        with open(output_file, "w") as f:
            json.dump(all_data, f)
        
        print(f"\n{'='*50}")
        print(f"Téléchargement terminé :")
        print(f"  ✓ Succès : {results['success']}")
        print(f"  ✗ Échecs : {results['failed']}")
        print(f"  ⚠ Rate limited : {results['rate_limited']}")
        print(f"  Total ticks : {len(all_data):,}")
        print(f"  Fichier : {output_file}")
        
        return output_file

def main():
    parser = argparse.ArgumentParser(description="Download Deribit BTC options data")
    parser.add_argument("--start", required=True, help="Date de début (YYYY-MM-DD)")
    parser.add_argument("--end", required=True, help="Date de fin (YYYY-MM-DD)")
    parser.add_argument("--instruments", nargs="*", help="Instruments spécifiques (optionnel)")
    
    args = parser.parse_args()
    
    if not TARDIS_API_KEY:
        print("Erreur : Variable TARDIS_API_KEY non définie")
        print("Créez un fichier .env avec votre clé API")
        return
    
    downloader = DeribitDataDownloader(TARDIS_API_KEY)
    
    # Récupération des instruments
    if args.instruments:
        instruments = args.instruments
    else:
        all_instruments = downloader.get_available_instruments()
        instruments = [
            inst["instrument_name"] 
            for inst in all_instruments 
            if "BTC" in inst["instrument_name"]
        ][:20]  # Limite pour le test
    
    # Téléchargement
    output = downloader.batch_download(instruments, args.start, args.end)
    
    # Conversion en CSV
    processor = OptionsDataProcessor()
    processor.json_to_csv(str(output))

if __name__ == "__main__":
    main()

Erreurs courantes et solutions

Durant mes tests avec l'API Tardis, j'ai rencontré plusieurs erreurs classiques. Voici mon retour d'expérience avec les solutions éprouvées :

Code d'erreur Symptôme Cause probable Solution
401 Unauthorized "Invalid API key" ou authentication failed Clé API incorrecte, expirée, ou mal formatée Vérifiez votre clé sur le dashboard Tardis. Assurez-vous que .env est chargé avec load_dotenv(). Testez avec : curl -H "X-API-Key: YOUR_KEY" {BASE_URL}/instruments
429 Rate Limited "Too many requests" après quelques requêtes Dépassement de 10 req/s sur le plan gratuit Implémentez un sleep de 0.15s entre chaque requête et un exponential backoff en cas de 429. J'utilise time.sleep(0.15) systématiquement.
400 Bad Request "Invalid date format" ou paramètres manquants Format de date non ISO 8601, dates inversées, ou instrument_name mal orthographié Utilisez le format ISO strict : "2026-03-01T00:00:00Z". Vérifiez l'orthographe exacte de l'instrument sur /instruments. Exemple correct : "BTC-28MAR26-95000-C"
500 Server Error "Internal server error" intermittent Problème temporaire côté serveur Tardis ou surcharge Implémentez un retry avec backoff exponentiel : 3 tentatives avec 5s, 15s, 45s d'attente. Loggez les erreurs pour identifier les patterns.
Empty Response [] ou {"data": []} retourné Période sans données ou instrument historique non disponible Vérifiez la période de rétention (30j gratuit). Certaines options expired ne sont plus disponibles. Utilisez /instruments pour vérifier la disponibilité.

数据质量验证与清洗

Une fois les données téléchargées, je recommande vivement une validation qualité avant toute utilisation en production. Voici ma checklist de validation :

import pandas as pd
import numpy as np

def validate_options_data(df: pd.DataFrame) -> dict:
    """
    Valide la qualité des données d'options téléchargées.
    
    Returns:
        dict: Rapport de validation avec score de qualité
    """
    report = {
        "total_rows": len(df),
        "missing_values": df.isnull().sum().to_dict(),
        "quality_score": 0.0,
        "issues": []
    }
    
    # Vérification des colonnes essentielles
    required_cols = ["datetime", "instrument_name", "last_price", "best_bid_price", "best_ask_price"]
    missing_cols = [col for col in required_cols if col not in df.columns]
    if missing_cols:
        report["issues"].append(f"Colonnes manquantes : {missing_cols}")
    
    # Vérification des prix négatifs ou absurdes
    if "last_price" in df.columns:
        negative_prices = (df["last_price"] <= 0).sum()
        if negative_prices > 0:
            report["issues"].append(f"Prix négatifs ou nuls : {negative_prices}")
    
    # Vérification du spread bid-ask (ne doit pas être négatif)
    if all(col in df.columns for col in ["best_bid_price", "best_ask_price"]):
        df["spread"] = df["best_ask_price"] - df["best_bid_price"]
        invalid_spread = (df["spread"] < 0).sum()
        if invalid_spread > 0:
            report["issues"].append(f"Spreads négatifs : {invalid_spread}")
        
        # Spread moyen acceptable < 5% pour BTC options
        avg_spread_pct = (df["spread"].mean() / df["last_price"].mean()) * 100
        if avg_spread_pct > 5:
            report["issues"].append(f"Spread moyen élevé : {avg_spread_pct:.2f}%")
    
    # Vérification de la continuité temporelle
    if "datetime" in df.columns:
        df_sorted = df.sort_values("datetime")
        time_diffs = df_sorted["datetime"].diff()
        large_gaps = (time_diffs > pd.Timedelta(hours=1)).sum()
        if large_gaps > len(df) * 0.1:
            report["issues"].append(f"Lacunes temporelles importantes : {large_gaps}")
    
    # Calcul du score de qualité
    base_score = 100
    penalties = len(report["issues"]) * 15
    report["quality_score"] = max(0, base_score - penalties)
    
    return report


Validation

validation = validate_options_data(df)
print(f"Score qualité : {validation['quality_score']}/100")
for issue in validation["issues"]:
    print(f"⚠ {issue}")

性能优化建议

Pour optimiser vos téléchargements massifs, voici les techniques que j'utilise en production :

# Exemple d'optimisation avec compression et streaming
import gzip
import ijson

def download_compressed(instrument: str, start: str, end: str, output: str):
    """Télécharge et compresse directement en gzip."""
    response = requests.get(
        f"{BASE_URL}/ticks",
        params={"api_key": API_KEY, "instrument_name": instrument, 
                "start_date": start, "end_date": end},
        stream=True,
        timeout=120
    )
    
    with gzip.open(output, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    print(f"✓ Compressé : {output}")

总结与下一步

Dans cet article, nous avons couvert l'ensemble du pipeline de téléchargement et traitement des données d'options BTC Deribit via l'API Tardis. Les points clés à retenir :

Si vous utilisez ces données d'options pour alimenter des modèles d'IA ou de machine learning, sachez que HolySheep AI propose des APIs d'intelligence artificielle avec moins de 50ms de latence, des prix compétitifs (GPT-4.1 à $8/MTok, Claude Sonnet 4.5 à $15/MTok) et le support de WeChat et Alipay pour les utilisateurs chinois.

Le code présenté dans cet article est copy-pasteable et fonctionne out-of-the-box avec Python 3.9+. N'hésitez pas à adapter les paramètres selon vos besoins spécifiques.

👈 Inscrivez-vous sur HolySheep AI — crédits offerts

Ressources connexes

Articles connexes