Tutoriel complet : importer les données tick des options Bybit dans DuckDB pour construire une surface de volatilité implicite

Dans un contexte de marchés dérivés crypto de plus en plus matures, j'ai personnellement passé plusieurs semaines à optimiser un pipeline complet permettant de récupérer les ticks d'options Bybit, de les charger dans DuckDB, puis de reconstruire une surface de volatilité implicite propre, lissée et exploitable pour le trading systématique. Ce tutoriel condense tout ce que j'aurais aimé trouver en un seul endroit — y compris l'étape d'analyse augmentée par IA via HolySheep AI, qui m'a fait gagner environ 85 % sur ma facture mensuelle par rapport à l'API OpenAI directe.

Avant d'entrer dans le code, comparons rapidement les coûts 2026 d'output pour 10 millions de tokens traités par mois, scénario réaliste pour un quant indépendant analysant chaque soir la surface de volatilité :

1. Prérequis techniques

• Python 3.11+ avec pip
• DuckDB ≥ 0.10.2 (moteur OLAP columnar in-process)
• httpx ou aiohttp pour les appels REST Bybit
• pandas, numpy, scipy pour le calcul de la surface
• Une clé API Bybit (lecture seule suffit pour les données publiques)
• Une clé HolySheep AI : S'inscrire ici pour obtenir des crédits gratuits

2. Récupérer les ticks d'options Bybit

Bybit expose plusieurs endpoints pour les options. Le plus utile pour un snapshot historique est /v5/option/tickers, complété par /v5/option/orderbook pour la profondeur et /v5/option/trades pour les transactions. Voici un script asynchrone robuste :

import asyncio
import httpx
from datetime import datetime, timezone

BASE = "https://api.bybit.com"

async def fetch_option_tickers(client: httpx.AsyncClient, symbol: str):
    url = f"{BASE}/v5/option/tickers"
    params = {"category": "option", "baseCoin": symbol, "limit": 500}
    r = await client.get(url, params=params, timeout=20.0)
    r.raise_for_status()
    data = r.json()
    if data.get("retCode") != 0:
        raise RuntimeError(data.get("retMsg", "Erreur Bybit"))
    return data["result"]["list"]

async def fetch_all(base_coin: str = "BTC"):
    async with httpx.AsyncClient() as client:
        rows = await fetch_option_tickers(client, base_coin)
        print(f"{len(rows)} options recuperees a {datetime.now(timezone.utc).isoformat()}")
        return rows

if __name__ == "__main__":
    asyncio.run(fetch_all("BTC"))

3. Charger dans DuckDB et nettoyer

DuckDB brille ici : ingestion directe depuis JSON, typage strict et SQL analytique ultra-rapide. On crée une table partitionnée par date :

import duckdb
import json
from datetime import datetime, timezone

con = duckdb.connect("bybit_options.duckdb")
con.execute("""
CREATE TABLE IF NOT EXISTS option_ticks (
    symbol        VARCHAR,
    underlying    VARCHAR,
    strike        DOUBLE,
    expiry        DATE,
    option_type   VARCHAR,
    mark_price    DOUBLE,
    index_price   DOUBLE,
    bid_iv        DOUBLE,
    ask_iv        DOUBLE,
    mark_iv       DOUBLE,
    delta         DOUBLE,
    gamma         DOUBLE,
    vega          DOUBLE,
    theta         DOUBLE,
    volume_24h    DOUBLE,
    open_interest DOUBLE,
    ts            TIMESTAMP
);
""")

def ingest(rows, underlying):
    records = []
    for r in rows:
        sym = r["symbol"]  # ex: BTC-30JUN24-70000-C
        parts = sym.split("-")
        strike = float(parts[2])
        opt_type = parts[3]
        expiry = datetime.strptime(parts[1], "%d%b%y").date()
        records.append((
            sym, underlying, strike, expiry, opt_type,
            float(r.get("markPrice", 0)),
            float(r.get("indexPrice", 0)),
            float(r.get("bidIv", 0)) * 100,
            float(r.get("askIv", 0)) * 100,
            float(r.get("markIv", 0)) * 100,
            float(r.get("delta", 0)),
            float(r.get("gamma", 0)),
            float(r.get("vega", 0)),
            float(r.get("theta", 0)),
            float(r.get("volume24h", 0)),
            float(r.get("openInterest", 0)),
            datetime.now(timezone.utc),
        ))
    con.executemany(
        "INSERT INTO option_ticks VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
        records,
    )

A chainer avec : ingest(rows, "BTC")
print("DuckDB pret :", con.execute("SELECT COUNT(*) FROM option_ticks").fetchone())

4. Construire la surface de volatilité implicite

Une surface propre se construit en trois passes : filtrage des arbitrages, interpolation sur une grille moneyness × maturité, puis lissage par splines cubiques. Voici un canevas reproductible :

import numpy as np
import duckdb
from scipy.interpolate import RectBivariateSpline
from datetime import date

def dte(expiry):
    return (expiry - date.today()).days

con = duckdb.connect("bybit_options.duckdb")
df = con.execute("""
    SELECT strike, expiry, option_type, mark_iv, delta, open_interest
    FROM option_ticks
    WHERE mark_iv > 5 AND mark_iv < 250
      AND open_interest > 0
      AND option_type = 'C'
""").df()

df["dte"] = df["expiry"].apply(dte)
df["log_moneyness"] = np.log(df["strike"] / 60000.0)  # spot approx

pivot = df.pivot_table(
    index="dte", columns="log_moneyness",
    values="mark_iv", aggfunc="mean"
).sort_index
Ressources connexes
📚 Tutoriels API IA
💰 Voir les tarifs
📖 Documentation
🚀 Inscription gratuite
Articles connexes
langgraph agent dongtailuyou gpt 55 yu deepseek v4
claude opus 47 yu gpt 55 bianmachangjing api zongh