Bonjour et bienvenue ! Je m'appelle Marc, et je suis analyste quantitatif spécialisé dans les données de marché crypto depuis maintenant six ans. Aujourd'hui, je vais partager avec vous une problématique que j'ai rencontrée des dizaines de fois : la qualité des données d'options Deribit pour vos modèles de pricing et vos analyses de risque. Si vous travaillez avec des options Deribit et que vous ressentez que quelque chose ne colle pas dans vos données historiques — notamment les Greeks qui semblent incohérents ou des trous inexpliqués dans les carnets d'ordres — cet article est fait pour vous.
Dans ce tutoriel complet, je vais vous guider pas à pas dans l'utilisation de l'API Tardis Machine pour effectuer un contrôle qualité rigoureux de vos données Deribit. Nous aborderons trois vérifications essentielles : la recalculation des Greeks, la détection des成交缺口 (gaps de trading), et la correction du漂移 d'horodatage (timestamp drift). Aucune expérience préalable avec les API n'est nécessaire — je pars de zéro avec vous.
Pourquoi le QA des données Deribit est crucial
Deribit est la plateforme de référence pour les options BTC et ETH avec plus de 10 milliards de dollars de volume mensuel. Cependant, comme toute source de données, elle présente des problématiques spécifiques que tout analyste sérieux doit maîtriser :
- Incohérence des Greeks : Les delta, gamma, vega et theta calculés par Deribit peuvent présenter des écarts significatifs avec vos propres modèles de pricing.
- Trous de données : Des gaps peuvent survenir lors de maintenance serveur, de pics de volatilité extrêmes, ou de problèmes de connectivité.
- Drift temporel : Les timestamps peuvent dévier de plusieurs millisecondes à plusieurs secondes, compromettant la synchronisation avec d'autres sources de données.
J'ai moi-même vécu des ситуаations critiques où des données défectueuses ont conduit à des valuations erronées de portefeuille à hauteur de plusieurs millions de dollars. La leçon ? Un contrôle qualité rigoureux n'est pas une option — c'est une nécessité absolue.
Pour qui / pour qui ce n'est pas fait
| ✅ Ce tutoriel est pour vous si... | ❌ Ce tutoriel n'est PAS pour vous si... |
|---|---|
| Vous êtes analyste quantitatif ou data scientist et manipulez des données d'options crypto. | Vous cherchez des stratégies de trading ou des signaux d'achat/vente. |
| Vous avez besoin de données historiques fiables pour backtester des modèles de pricing. | Vous n'avez aucune familiarité avec la programmation Python ou les concepts de base des options. |
| Vous travaillez sur des systèmes de risk management nécessitant des données en temps réel et historiques. | Vous souhaitez simplement consulter les prix actuels sans analyse approfondie. |
| Vous migrez depuis une autre source de données (Binance, OKX) et devez valider la qualité. | Vous n'avez pas accès à un environnement Python fonctionnel ou un environnement cloud. |
Prérequis : Ce dont vous avez besoin
Avant de commencer, assurons-nous que vous avez tous les outils nécessaires. Je vais détailler chaque élément pour les débutants complets.
1. Un compte Tardis Machine
Tardis Machine est le service que nous utiliserons pour accéder aux données historiques de Deribit. C'est une solution spécialisée qui archive les données de marché depuis 2017 avec une précision milliseconde.
Pour créer votre compte :
- Rendez-vous sur https://tardis.dev
- Cliquez sur "Sign Up" en haut à droite
- Choisissez le plan adapté à vos besoins (le plan gratuit suffit pour des tests initiaux)
- Après validation, récupérez votre API Key dans Settings → API Keys
2. Python 3.9+ installé
Si vous n'avez jamais programmé en Python, pas de panique. Téléchargez Python depuis python.org et installez-le. Assurez-vous de cocher "Add Python to PATH" lors de l'installation.
3. Les bibliothèques Python nécessaires
Ouvrez votre terminal (ouInvite de commandes sur Windows) et exécutez :
pip install requests pandas numpy scipy matplotlib tardis-clientSi vous rencontrez des erreurs de permission, ajoutez --user à la fin de la commande. Félicitations, vous venez d'installer vos premiers outils de data scientist !
4. (Optionnel) HolySheep AI pour l'analyse avancée
Pour les analyses plus poussées nécessitant du machine learning ou des modèles GPT, je vous recommande vous inscrire ici sur HolySheep AI. Leur API offre des performances exceptionnelles avec une latence inférieure à 50ms et des tarifs compétitifs — DeepSeek V3.2 à seulement 0,42 $ par million de tokens, soit une économie de 85% par rapport aux solutions américaines.
Comprendre la structure des données Deribit
Avant de plonge dans le code, comprenons ce que nous allons récupérer. Les données Deribit se déclinent en plusieurs types, chacun servant un objectif différent dans notre QA.
Les types de données essentiels
| Type de données | Description | Utilité pour le QA |
|---|---|---|
| Trades | Historique des transactions exécutées | Détection des gaps, calcul des volumes |
| Orderbook | Profondeur du carnet d'ordres | Calcul des mid-prices, slippage |
| Quotes (Ticker) | Prix bid/ask en temps réel | Vérification des Greeks理论 |
| Deribit Greeks | Delta, Gamma, Vega, Theta calculés par Deribit | Recalculation et comparaison |
| Funding | Taux de financement périodique | Vérification des coûts de portage |
Structure d'une réponse API Deribit
{
"trade_seq": 145678,
"trade_id": "12345-67890",
"timestamp": 1704067200000,
"tick_direction": "plus_tick",
"price": "45250.5",
"index_price": "45230.0",
"instrument_name": "BTC-28MAR25-45000-C",
"direction": "buy",
"amount": 0.1
}Notez que les prix sont exprimés en strings pour éviter les problèmes de virgule flottante — un piège classique pour les débutants !
Configuration initiale du projet
Créons ensemble votre premier script de QA. Je vais vous guider pas à pas.
Étape 1 : Structure du projet
Créez un dossier nommé deribit_qa et créez-y un fichier config.py :
# config.py — Configuration centrale du projet QA Deribit
=============================================================================
PARAMÈTRES D'ACCÈS API
=============================================================================
TARDIS_API_KEY = "votre_cle_tardis_ici" # Remplacez par votre vraie clé
EXCHANGE = "deribit"
DATA_TYPES = ["trades", "quotes", "greeks"]
=============================================================================
PARAMÈTRES DE CALCUL DES GREEKS (Black-Scholes)
=============================================================================
Paramètres par défaut pour le recalcul des Greeks
BLACK_SCHOLES_PARAMS = {
"risk_free_rate": 0.05, # Taux sans risque annualisé (5%)
"dividend_yield": 0.0, # Pas de dividende sur BTC/ETH
"default_volatility": 0.80, # Volatilité implicite par défaut (80%)
}
=============================================================================
SEUILS DE VALIDATION QA
=============================================================================
QA_THRESHOLDS = {
# Tolérance pour la comparaison des Greeks (en pourcentage)
"greeks_tolerance": 0.05, # 5% d'écart maximum toléré
# Tolérance pour la détection de gaps (en millisecondes)
"gap_threshold_ms": 5000, # Considérer comme gap si > 5 secondes
# Tolérance pour le drift temporel (en millisecondes)
"timestamp_drift_ms": 100, # Alerte si drift > 100ms
# Volume minimum par trade pour éviter le bruit
"min_trade_volume": 0.001, # Minimum 0.001 BTC ou ETH
}
=============================================================================
PÉRIODE D'ANALYSE (exemple : 1 jour)
=============================================================================
from datetime import datetime, timedelta
ANALYSIS_START = datetime(2025, 3, 15, 0, 0, 0)
ANALYSIS_END = datetime(2025, 3, 16, 0, 0, 0)
=============================================================================
INSTRUMENTS À ANALYSER
=============================================================================
INSTRUMENTS = [
"BTC-28MAR25-45000-C", # Call ATM BTC Mars 2025
"BTC-28MAR25-45000-P", # Put ATM BTC Mars 2025
"BTC-28MAR25-50000-C", # Call OTM BTC Mars 2025
"ETH-28MAR25-2500-C", # Call ATM ETH Mars 2025
]
print("✅ Configuration chargée avec succès !")Étape 2 : Script principal de récupération des données
# main_qa.py — Script principal de contrôle qualité Deribit
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from config import TARDIS_API_KEY, EXCHANGE, ANALYSIS_START, ANALYSIS_END
=============================================================================
CLASSE PRINCIPALE : DERIBIT DATA FETCHER
=============================================================================
class DeribitDataFetcher:
"""
Classe pour récupérer et valider les données historiques Deribit
via l'API Tardis Machine.
Cette classe abstrait toute la complexité de l'API pour vous permettre
de vous concentrer sur l'analyse.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://tardis-micro.api.tardis.dev/v1"
def _make_request(self, endpoint: str, params: dict) -> dict:
"""
Effectue une requête HTTP vers l'API Tardis.
Gère automatiquement les erreurs et le rate limiting.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
url = f"{self.base_url}/{endpoint}"
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise Exception("❌ Erreur d'authentification. Vérifiez votre clé API.")
elif response.status_code == 429:
raise Exception("⚠️ Rate limit atteint. Attendez quelques secondes.")
else:
raise Exception(f"❌ Erreur HTTP {response.status_code}: {e}")
except requests.exceptions.RequestException as e:
raise Exception(f"❌ Erreur de connexion: {e}")
def get_trades(self, instrument: str, start: datetime, end: datetime) -> pd.DataFrame:
"""
Récupère l'historique des trades pour un instrument donné.
Args:
instrument: Nom de l'instrument Deribit (ex: "BTC-28MAR25-45000-C")
start: Date de début
end: Date de fin
Returns:
DataFrame pandas avec les données de trades
"""
print(f"📥 Récupération des trades pour {instrument}...")
params = {
"exchange": EXCHANGE,
"symbol": instrument,
"from": int(start.timestamp()),
"to": int(end.timestamp()),
"limit": 10000, # Maximum par requête
}
data = self._make_request("trades", params)
if not data or "trades" not in data:
print(f"⚠️ Aucun trade trouvé pour {instrument}")
return pd.DataFrame()
df = pd.DataFrame(data["trades"])
# Conversion des timestamps
if "timestamp" in df.columns:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("datetime").reset_index(drop=True)
print(f"✅ {len(df)} trades récupérés pour {instrument}")
return df
def get_quotes(self, instrument: str, start: datetime, end: datetime) -> pd.DataFrame:
"""
Récupère les quotes (bid/ask) pour un instrument donné.
"""
print(f"📥 Récupération des quotes pour {instrument}...")
params = {
"exchange": EXCHANGE,
"symbol": instrument,
"from": int(start.timestamp()),
"to": int(end.timestamp()),
"limit": 10000,
}
data = self._make_request("quotes", params)
if not data or "quotes" not in data:
print(f"⚠️ Aucune quote trouvée pour {instrument}")
return pd.DataFrame()
df = pd.DataFrame(data["quotes"])
if "timestamp" in df.columns:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("datetime").reset_index(drop=True)
print(f"✅ {len(df)} quotes récupérées pour {instrument}")
return df
def get_greeks(self, instrument: str, start: datetime, end: datetime) -> pd.DataFrame:
"""
Récupère les Greeks calculés par Deribit.
"""
print(f"📥 Récupération des Greeks pour {instrument}...")
params = {
"exchange": EXCHANGE,
"symbol": instrument,
"from": int(start.timestamp()),
"to": int(end.timestamp()),
"limit": 10000,
}
data = self._make_request("greeks", params)
if not data or "greeks" not in data:
print(f"⚠️ Aucun Greek trouvé pour {instrument}")
return pd.DataFrame()
df = pd.DataFrame(data["greeks"])
if "timestamp" in df.columns:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("datetime").reset_index(drop=True)
print(f"✅ {len(df)} enregistrements Greeks récupérés pour {instrument}")
return df
=============================================================================
EXÉCUTION PRINCIPALE
=============================================================================
if __name__ == "__main__":
print("🚀 Démarrage du contrôle qualité Deribit...")
# Initialisation du fetcher
fetcher = DeribitDataFetcher(api_key=TARDIS_API_KEY)
# Test avec un instrument
test_instrument = "BTC-28MAR25-45000-C"
# Récupération des données
trades_df = fetcher.get_trades(test_instrument, ANALYSIS_START, ANALYSIS_END)
quotes_df = fetcher.get_quotes(test_instrument, ANALYSIS_START, ANALYSIS_END)
greeks_df = fetcher.get_greeks(test_instrument, ANALYSIS_START, ANALYSIS_END)
print("\n📊 Résumé des données récupérées :")
print(f" - Trades: {len(trades_df)} lignes")
print(f" - Quotes: {len(quotes_df)} lignes")
print(f" - Greeks: {len(greeks_df)} lignes")Module 1 : Vérification et recalculation des Greeks
Les Greeks (Delta, Gamma, Vega, Theta) sont les sensibilités du prix d'une option aux différents facteurs de risque. Deribit les calcule en interne, mais il est crucial de vérifier leur exactitude pour votre propre gestion des risques.
Comprendre les Greeks avec Black-Scholes
La formule de Black-Scholes permet de calculer le prix théorique d'une option européenne, puis de dériver les Greeks analytiquement. Voici les formules clés :
- Delta (Δ) : Dérivée du prix de l'option par rapport au prix du sous-jacent. Varie entre -1 et 1.
- Gamma (Γ) : Dérivée seconde du prix par rapport au spot. Mesure la convexité.
- Vega (ν) : Dérivée par rapport à la volatilité implicite. Toujours positif pour les options.
- Theta (Θ) : Dérivée par rapport au temps. Généralement négatif (le temps érode la valeur).
Script de recalculation des Greeks
# greeks_recalculation.py — Recalculation des Greeks avec Black-Scholes
import pandas as pd
import numpy as np
from scipy.stats import norm
from config import BLACK_SCHOLES_PARAMS
=============================================================================
FONCTIONS BLACK-SCHOLES
=============================================================================
def black_scholes_price(S, K, T, r, sigma, option_type="call"):
"""
Calcule le prix théorique d'une option avec Black-Scholes.
Args:
S: Prix du sous-jacent (spot)
K: Prix d'exercice (strike)
T: Temps jusqu'à l'échéance (en années)
r: Taux sans risque annualisé
sigma: Volatilité implicite
option_type: "call" ou "put"
Returns:
Prix de l'option
"""
if T <= 0:
# Option expirée : valeur intrinsèque uniquement
if option_type == "call":
return max(S - K, 0)
else:
return max(K - S, 0)
# Protection contre division par zéro
if sigma <= 0 or T <= 0:
return 0
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == "call":
price = S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
else:
price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
return price
def calculate_greeks(S, K, T, r, sigma, option_type="call"):
"""
Calcule tous les Greeks d'une option.
Returns:
Dict contenant delta, gamma, vega, theta, rho
"""
if T <= 0 or sigma <= 0:
return {
"delta": 0 if T <= 0 else (1 if option_type == "call" else -1),
"gamma": 0,
"vega": 0,
"theta": 0,
"rho": 0
}
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
# Delta
if option_type == "call":
delta = norm.cdf(d1)
else:
delta = norm.cdf(d1) - 1
# Gamma (identique pour call et put)
gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T))
# Vega (identique pour call et put)
vega = S * norm.pdf(d1) * np.sqrt(T) / 100 # Par unité de volatilité
# Theta
term1 = -S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
if option_type == "call":
theta = (term1 - r * K * np.exp(-r * T) * norm.cdf(d2)) / 365
else:
theta = (term1 + r * K * np.exp(-r * T) * norm.cdf(-d2)) / 365
# Rho
if option_type == "call":
rho = K * T * np.exp(-r * T) * norm.cdf(d2) / 100
else:
rho = -K * T * np.exp(-r * T) * norm.cdf(-d2) / 100
return {
"delta": delta,
"gamma": gamma,
"vega": vega,
"theta": theta,
"rho": rho,
"d1": d1,
"d2": d2
}
def compare_greeks(deribit_greeks: pd.DataFrame,
recalculated_greeks: pd.DataFrame,
tolerance: float = 0.05) -> pd.DataFrame:
"""
Compare les Greeks Deribit avec nos recalculations.
Args:
deribit_greeks: DataFrame des Greeks Deribit originaux
recalculated_greeks: DataFrame de nos recalculations
tolerance: Tolérance en pourcentage (5% par défaut)
Returns:
DataFrame avec les écarts relatifs
"""
comparison = pd.DataFrame()
greeks_list = ["delta", "gamma", "vega", "theta"]
for greek in greeks_list:
if greek in deribit_greeks.columns and greek in recalculated_greeks.columns:
# Calcul de l'écart relatif
# Note: On utilise abs() pour éviter les problèmes de signe
diff = abs(recalculated_greeks[greek] - deribit_greeks[greek])
relative_diff = diff / (abs(deribit_greeks[greek]) + 1e-10)
comparison[f"{greek}_diff"] = relative_diff
comparison[f"{greek}_passed"] = relative_diff <= tolerance
return comparison
=============================================================================
CLASSE DE VÉRIFICATION DES GREEKS
=============================================================================
class GreeksVerifier:
"""
Classe pour vérifier la cohérence des Greeks Deribit.
"""
def __init__(self, risk_free_rate=None, dividend_yield=None):
self.r = risk_free_rate or BLACK_SCHOLES_PARAMS["risk_free_rate"]
self.q = dividend_yield or BLACK_SCHOLES_PARAMS["dividend_yield"]
def verify_instrument(self, quotes_df: pd.DataFrame,
greeks_df: pd.DataFrame,
option_type: str = "call") -> dict:
"""
Vérifie les Greeks pour un instrument.
Returns:
Dict avec les statistiques de vérification
"""
if quotes_df.empty or greeks_df.empty:
return {"status": "INCOMPLETE_DATA", "message": "Données insuffisantes"}
results = {
"status": "SUCCESS",
"total_records": len(greeks_df),
"passed_records": 0,
"failed_records": 0,
"max_delta_error": 0,
"max_gamma_error": 0,
"max_vega_error": 0,
"max_theta_error": 0,
"warnings": []
}
# Merge des données
merged = pd.merge_asof(
greeks_df.sort_values("datetime"),
quotes_df.sort_values("datetime"),
on="datetime",
direction="nearest",
tolerance=1000 # Tolérance de 1 seconde
)
# Pour chaque enregistrement, recalculer et comparer
for idx, row in merged.iterrows():
if pd.isna(row.get("bid_iv")) or pd.isna(row.get("underlying_price")):
continue
# Extraire les paramètres
S = float(row["underlying_price"])
K = float(row.get("strike_price", 0))
# Calculer T en années
expiry_timestamp = row.get("expiration_timestamp", 0)
if expiry_timestamp and row["timestamp"]:
T = (expiry_timestamp - row["timestamp"]) / (365 * 24 * 3600 * 1000)
else:
T = row.get("time_to_expiry", 0.1) / 365
# Volatilité implicite
sigma = float(row.get("bid_iv", row.get("mark_iv", 0.8)))
# Recalculation
our_greeks = calculate_greeks(S, K, T, self.r, sigma, option_type)
# Comparaison
deribit_delta = float(row.get("delta", 0))
delta_diff = abs(our_greeks["delta"] - deribit_delta)
# Mise à jour des stats
results["max_delta_error"] = max(results["max_delta_error"], delta_diff)
if delta_diff <= BLACK_SCHOLES_PARAMS.get("greeks_tolerance", 0.05):
results["passed_records"] += 1
else:
results["failed_records"] += 1
results["warnings"].append({
"timestamp": row["datetime"],
"type": "DELTA_MISMATCH",
"expected": our_greeks["delta"],
"actual": deribit_delta,
"error": delta_diff
})
results["pass_rate"] = results["passed_records"] / results["total_records"] if results["total_records"] > 0 else 0
return results
=============================================================================
TEST
=============================================================================
if __name__ == "__main__":
# Test simple
S, K, T, r, sigma = 45000, 45000, 30/365, 0.05, 0.8
print("📊 Test de recalculation Black-Scholes")
print("-" * 50)
print(f"Spot: ${S:,.2f}, Strike: ${K:,.2f}")
print(f"Temps: {T*365:.0f} jours, Taux: {r*100:.1f}%, Vol: {sigma*100:.0f}%")
print("-" * 50)
for option_type in ["call", "put"]:
price = black_scholes_price(S, K, T, r, sigma, option_type)
greeks = calculate_greeks(S, K, T, r, sigma, option_type)
print(f"\n📈 {option_type.upper()}")
print(f" Prix: ${price:,.2f}")
print(f" Delta: {greeks['delta']:.4f}")
print(f" Gamma: {greeks['gamma']:.6f}")
print(f" Vega: {greeks['vega']:.4f}")
print(f" Theta: {greeks['theta']:.4f}")Module 2 : Détection des成交缺口 (Trading Gaps)
Les gaps de trading sont des périodes pendant lesquelles aucun trade n'a eu lieu alors qu'il aurait dû y en avoir. Ils peuvent être causés par :
- Des pannes serveur chez Deribit
- Des problèmes de connectivité réseau
- Des périodes de maintenance programmée
- Des mouvements de marché extrêmes où la liquidité disparaît
Script de détection des gaps
# gap_detection.py — Détection des gaps de trading
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from config import QA_THRESHOLDS
class GapDetector:
"""
Détecte les gaps dans les données de trades Deribit.
"""
def __init__(self, threshold_ms: int = None):
"""
Args:
threshold_ms: Seuil en millisecondes au-delà duquel
une absence de trade est considérée comme un gap
"""
self.threshold_ms = threshold_ms or QA_THRESHOLDS["gap_threshold_ms"]
def detect_gaps(self, trades_df: pd.DataFrame) -> pd.DataFrame:
"""
Analyse un DataFrame de trades et détecte les gaps.
Args:
trades_df: DataFrame avec colonne 'datetime' ou 'timestamp'
Returns:
DataFrame avec les gaps détectés
"""
if trades_df.empty or len(trades_df) < 2:
return pd.DataFrame()
# S'assurer que les timestamps sont triés
df = trades_df.copy()
if "datetime" in df.columns:
df = df.sort_values("datetime")
elif "timestamp" in df.columns:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("datetime")
else:
raise ValueError("Colonne datetime ou timestamp requise")
# Calculer les intervalles entre trades consécutifs
df["time_diff_ms"] = df["datetime"].diff().dt.total_seconds() * 1000
# Identifier les gaps
gaps = df[df["time_diff_ms"] > self.threshold_ms].copy()
if gaps.empty:
return pd.DataFrame()
# Créer un rapport de gaps
gap_report = pd.DataFrame({
"gap_id": range(1, len(gaps) + 1),
"start_time": gaps["datetime"].shift(1).values,
"end_time": gaps["datetime"].values,
"gap_duration_ms": gaps["time_diff_ms"].values,
"gap_duration_sec": gaps["time_diff_ms"].values / 1000,
"instrument_before": gaps["instrument_name"].shift(1).values if "instrument_name" in gaps.columns else None,
"instrument_after": gaps["instrument_name"].values if "instrument_name" in gaps.columns else None,
})
# Supprimer la première ligne (pas de gap avant le premier trade)
gap_report = gap_report.dropna(subset=["start_time"])
return gap_report
def analyze_gap_patterns(self, gap_report: pd.DataFrame) -> dict:
"""
Analyse les patterns de gaps pour identifier des problèmes systémiques.
Returns:
Dict avec statistiques sur les gaps
"""
if gap_report.empty:
return {
"status": "NO_GAPS",
"total_gaps": 0,
"message": "Aucun gap détecté"
}
stats = {
"status": "GAPS_DETECTED",
"total_gaps": len(gap_report),
"avg_gap_duration_sec": gap_report["gap_duration_sec"].mean(),
"max_gap_duration_sec": gap_report["gap_duration_sec"].max(),
"min_gap_duration_sec": gap_report["gap_duration_sec"].min(),
"gaps_by_hour": {},
"suspicious_gaps": []
}
# Analyser par heure
if "start_time" in gap_report.columns:
gap_report["hour"] = pd.to_datetime(gap_report["start_time"]).dt.hour
stats["gaps_by_hour"] = gap_report.groupby("hour").size().to_dict()
# Identifier les gaps suspects (> 1 minute)
suspicious_threshold = 60 # 1 minute
stats["suspicious_gaps"] = gap_report[
gap_report["gap_duration_sec"] > suspicious_threshold
]["gap_duration_sec"].tolist()
stats["has_suspicious_gaps"] = len(stats["suspicious_gaps"]) > 0
return stats
def generate_gap_alerts(self, gap_report: pd.DataFrame,
gap_stats: dict) -> list:
"""
Génère des alertes formatées pour les gaps critiques.
Returns:
Liste de dictionnaires avec les alertes
"""
alerts = []
# Alerte pour gaps moyens > 30 secondes
if gap_stats.get("avg_gap_duration_sec", 0) > 30:
alerts.append({
"severity": "HIGH",
"type": "HIGH_AVERAGE_GAP",
"message": f"Durée moyenne des gaps élevée: {gap_stats['avg_gap_duration_sec']:.1f}s",
"recommendation": "Vérifier la stabilité de la connexion Deribit"
})
# Alerte pour gaps > 5 minutes
max_gap = gap_stats.get("max_gap_duration_sec", 0)
if max_gap > 300:
alerts.append({
"severity": "CRITICAL",
"type": "MAJOR_DATA_LOSS",
"message": f"Gap majeur détecté: {max_gap/60:.1f} minutes",
"recommendation": "Contacter le support Deribit pour récupérer les données"
})
# Alerte pour concentration de gaps
gaps_by_hour = gap_stats.get("gaps_by_hour", {})
if gaps_by_hour:
peak_hour = max(gaps_by_hour, key=gaps_by_hour.get)
if gaps_by_hour[peak_hour] > 5:
alerts.append({
"severity": "MEDIUM",
"type": "HOURLY_CONCENTRATION",
"message": f"Concentration de {gaps_by_hour[peak_hour]} gaps à {peak_hour}h",
"recomm