Vous tentez de télécharger les données d'options BTC depuis Deribit via l'API de Tardis.dev. Vous lancez votre script Python le vendredi soir, confiant d'avoir les données prêtes pour le week-end. Puis l'erreur apparaît :
ConnectionError: HTTPSConnectionPool(host='http://api.tardis.dev', port=443):
Max retries exceeded with url: /v1/deribit/btc_options/chunks?...
(Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x...>:
Failed to establish a new connection: [Errno 110] Connection timed out'))
OU pire encore :
403 Forbidden: "Daily download limit exceeded.
Upgrade your plan at https://tardis.dev/plans"Cette situation frustrante, je l'ai vécue personnellement lors d'un projet de recherche sur la surface de volatilité implicite du BTC. Après 3 heures de debugging, j'ai compris que la gestion robuste d'un pipeline de données financières nécessite bien plus qu'un simple requests.get(). Cet article détaille ma méthode complète pour construire un pipeline fiable de téléchargement, traitement et backtesting d'options BTC Deribit.
Prérequis et Installation
Avant de commencer, installez les dépendances nécessaires :
pip install tardis-client pandas numpy scipy matplotlib python-dotenv aiohttp asyncio
pip install jupyter notebook # Pour le backtesting interactif
pip install plotly kaleido # Pour les visualisations 3DLes packages essentiels sont tardis-client pour l'API et scipy pour l'interpolation de la surface de volatilité implicite.
Configuration de l'API Tardis.dev
Créez un fichier .env à la racine de votre projet :
# .env
TARDIS_API_KEY=your_tardis_api_key_here
DERIBIT_INSTRUMENT_PATTERN=BTC-*Téléchargement Robuste des Données CSV
Le code suivant implémente un système de retry automatique avec exponential backoff, essential pour les APIs financières :
import os
import time
import asyncio
from tardis_client import TardisClient, Channel
from tardis_client.exceptions import TardisClientException
import pandas as pd
from dotenv import load_dotenv
load_dotenv()
class DeribitOptionsDownloader:
"""
Téléchargeur robuste de données d'options Deribit BTC.
Implémente retry exponentiel et gestion de rate limiting.
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv("TARDIS_API_KEY")
self.base_url = "https://api.tardis.dev/v1"
self.max_retries = 5
self.base_delay = 2 # secondes
if not self.api_key:
raise ValueError("TARDIS_API_KEY non configurée")
async def download_with_retry(self, exchange: str, symbol: str,
start_date: str, end_date: str) -> pd.DataFrame:
"""
Télécharge les données avec retry exponentiel.
Args:
exchange: 'deribit' pour les options BTC
symbol: '*' pour tous ou 'BTC-*' pour options BTC
start_date: '2024-01-01' format
end_date: '2024-12-31' format
Returns:
DataFrame avec colonnes: timestamp, instrument_name,
mark_price, mark_iv, best_bid_price, best_ask_price, etc.
"""
delay = self.base_delay
for attempt in range(self.max_retries):
try:
client = TardisClient(api_key=self.api_key)
# Configuration pour options Deribit
# Channels disponibles: trade, quote, derivative_summary
messages = client.replay(
exchange=exchange,
from_timestamp=f"{start_date}T00:00:00.000Z",
to_timestamp=f"{end_date}T23:59:59.999Z",
channels=[Channel(name="quote", symbols=[symbol])]
)
records = []
async for message in messages:
if message.channel_name == "quote":
records.append({
"timestamp": message.timestamp,
"instrument_name": message.symbol,
"best_bid_price": message.best_bid_price,
"best_ask_price": message.best_ask_price,
"best_bid_amount": message.best_bid_amount,
"best_ask_amount": message.best_ask_amount,
"mark_price": message.mark_price,
"mark_iv": message.mark_iv,
"underlying_price": message.underlying_price,
"underlying_index": message.underlying_index,
})
df = pd.DataFrame(records)
print(f"✓ Téléchargé {len(df)} quotes pour {symbol}")
return df
except Exception as e:
print(f"⚠ Tentative {attempt + 1}/{self.max_retries} échouée: {e}")
if "429" in str(e) or "rate limit" in str(e).lower():
# Rate limiting: attendre plus longtemps
delay *= 4
elif "403" in str(e) or "401" in str(e):
# Erreur d'authentification: ne pas retry
raise PermissionError(f"Clé API invalide ou expirée: {e}")
elif attempt < self.max_retries - 1:
print(f" Retry dans {delay}s...")
time.sleep(delay)
delay *= 2 # Exponential backoff
raise RuntimeError(f"Échec après {self.max_retries} tentatives")
Utilisation
downloader = DeribitOptionsDownloader()
df_quotes = await downloader.download_with_retry(
exchange="deribit",
symbol="BTC-*",
start_date="2024-06-01",
end_date="2024-06-30"
)Construction de la Surface de Volatilité Implicite
Une fois les données téléchargées, construisons la surface de volatilité implicite. Cette surface est fondamentale pour le pricing d'options et l'analyse de marché :
import numpy as np
from scipy.interpolate import griddata, RBFInterpolator
from scipy.stats import norm
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')
class VolatilitySurfaceBuilder:
"""
Construit une surface de volatilité implicite 3D (Strike × Maturity → IV).
Utilise interpolation RBF pour lissage.
"""
def __init__(self, risk_free_rate: float = 0.05):
self.r = risk_free_rate
self.df = None
self.surface = None
def parse_instrument_name(self, instrument: str) -> dict:
"""
Parse les noms d'instruments Deribit.
Ex: 'BTC-28JUN24-95000-C' → {expiry: date, strike: 95000, type: 'call'}
"""
parts = instrument.split('-')
expiry_str = parts[1] # '28JUN24'
strike = float(parts[2])
option_type = parts[3].lower() # 'C' ou 'P'
# Parse date Deribit
expiry = datetime.strptime(expiry_str, "%d%b%y")
return {
"strike": strike,
"expiry": expiry,
"option_type": option_type,
"T": (expiry - datetime.now()).days / 365.0
}
def compute_black_scholes_iv(self, S: float, K: float, T: float,
option_price: float, is_call: bool = True) -> float:
"""
Calcule la volatilité implicite via Newton-Raphson.
Précision: ±0.0001 en max 100 itérations.
"""
if T <= 0 or option_price <= 0:
return np.nan
# Prix intrinsèque
intrinsic = max(S - K, 0) if is_call else max(K - S, 0)
if option_price < intrinsic:
return np.nan
sigma = 0.5 # Estimation initiale
for _ in range(100):
d1 = (np.log(S/K) + (self.r + sigma**2/2)*T) / (sigma*np.sqrt(T))
d2 = d1 - sigma*np.sqrt(T)
price = S*norm.cdf(d1) - K*np.exp(-self.r*T)*norm.cdf(d2) if is_call \
else K*np.exp(-self.r*T)*norm.cdf(-d2) - S*norm.cdf(-d1)
vega = S * np.sqrt(T) * norm.pdf(d1)
if vega < 1e-10:
break
diff = price - option_price
if abs(diff) < 1e-6:
break
sigma -= diff / vega
return max(sigma, 0.01) # Minimum 1% IV
def build_surface(self, df_quotes: pd.DataFrame) -> np.ndarray:
"""
Construit la surface de volatilité depuis les données de marché.
Returns:
grid: meshgrid (strikes, maturities)
iv_surface: matrice 2D des volatilités implicites
"""
# Parse instruments
parsed = df_quotes['instrument_name'].apply(self.parse_instrument_name)
df_quotes['strike'] = parsed.apply(lambda x: x['strike'])
df_quotes['T'] = parsed.apply(lambda x: x['T'])
df_quotes['option_type'] = parsed.apply(lambda x: x['option_type'])
# Filtrer données valides (IV > 0)
df_valid = df_quotes[df_quotes['mark_iv'] > 0].copy()
# Agréger par strike et maturité (moyenne)
agg = df_valid.groupby(['strike', 'T']).agg({
'mark_iv': 'mean',
'mark_price': 'mean',
'underlying_price': 'mean'
}).reset_index()
# Créer grille uniforme
strikes = np.linspace(agg['strike'].min(), agg['strike'].max(), 50)
maturities = np.linspace(agg['T'].min(), agg['T'].max(), 20)
strike_grid, T_grid = np.meshgrid(strikes, maturities)
# Interpolation RBF pour lissage
points = agg[['strike', 'T']].values
values = agg['mark_iv'].values
# Normalisation pour interpolation stable
points_norm = points / np.array([points[:,0].max(), 1])
rbf = RBFInterpolator(points_norm, values, kernel='thin_plate_spline', smoothing=2)
grid_points = np.column_stack([
strike_grid.ravel(),
T_grid.ravel()
])
grid_points_norm = grid_points / np.array([points[:,0].max(), 1])
iv_grid = rbf(grid_points_norm).reshape(strike_grid.shape)
iv_grid = np.clip(iv_grid, 0.1, 2.0) # Bornes réalistes
self.surface = iv_grid
self.strike_grid = strike_grid
self.T_grid = T_grid
self.df = agg
return self.surface
def plot_surface_3d(self):
"""Visualisation interactive de la surface de volatilité."""
import plotly.graph_objects as go
fig = go.Figure(data=[go.Surface(
x=self.strike_grid,
y=self.T_grid * 365, # Convertir en jours
z=self.surface * 100, # En pourcentage
colorscale='Viridis',
colorbar_title="IV (%)"
)])
fig.update_layout(
title="Surface de Volatilité Implicite BTC Options",
scene=dict(
xaxis_title="Strike (USD)",
yaxis_title="Maturité (jours)",
zaxis_title="Volatilité Implicite (%)"
)
)
return fig
Construction de la surface
builder = VolatilitySurfaceBuilder(risk_free_rate=0.04)
surface = builder.build_surface(df_quotes)
Visualisation
fig = builder.plot_surface_3d()
fig.show()Pipeline de Backtesting de Stratégies d'Options
Avec la surface de volatilité construite, implémentons un pipeline de backtesting complet pour tester des stratégies comme les straddles, strangles ou spreads :
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
import pandas as pd
@dataclass
class Option:
strike: float
expiry: datetime
option_type: str # 'call' ou 'put'
premium: float
iv: float
@dataclass
class Position:
options: List[Option]
quantity: int
entry_date: datetime
entry_spot: float
class OptionsBacktester:
"""
Backtester complet pour stratégies d'options BTC.
Inclut calcul P&L, grecques et métriques de risque.
"""
def __init__(self, initial_capital: float = 100_000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.trades = []
self.positions = []
self.portfolio_history = []
def open_straddle(self, spot: float, strike_pct: float,
expiry: datetime, iv_atm: float, premium: float,
quantity: int = 1):
"""
Ouvre un straddle (1 call ATM + 1 put ATM).
Args:
strike_pct: % ATM (ex: 0.0 = ATM, 0.05 = +5% OTM)
"""
strike = spot * (1 + strike_pct)
call = Option(
strike=strike,
expiry=expiry,
option_type='call',
premium=premium * (1 + iv_atm), # Ajustement IV
iv=iv_atm
)
put = Option(
strike=strike,
expiry=expiry,
option_type='put',
premium=premium * (1 + iv_atm),
iv=iv_atm
)
position = Position(
options=[call, put],
quantity=quantity,
entry_date=datetime.now(),
entry_spot=spot
)
cost = (call.premium + put.premium) * quantity * 100 # 1 contrat = 1 BTC
self.capital -= cost
self.positions.append(position)
self.trades.append({
"date": datetime.now(),
"action": "OPEN",
"strategy": "straddle",
"strike": strike,
"premium": cost,
"spot": spot
})
print(f"✓ Straddle ouvert: strike={strike:.0f}, coût={cost:.2f} USD")
return position
def close_position(self, position: Position, current_spot: float,
current_date: datetime, realized_vol: float):
"""
Ferme une position et calcule P&L.
"""
time_elapsed = (current_date - position.entry_date).days / 365
pnl = 0
for opt in position.options:
# Valeur intrinsèque à l'expiration
if opt.option_type == 'call':
intrinsic = max(current_spot - opt.strike, 0)
else:
intrinsic = max(opt.strike - current_spot, 0)
# Valeur temps restante (simplifié)
remaining_value = intrinsic * np.exp(-realized_vol * np.sqrt(time_elapsed))
pnl += remaining_value * position.quantity * 100
gross_pnl = pnl - (position.options[0].premium + position.options[1].premium) * \
position.quantity * 100
self.capital += pnl
self.positions.remove(position)
self.trades.append({
"date": current_date,
"action": "CLOSE",
"strategy": "straddle",
"spot": current_spot,
"pnl": gross_pnl,
"return_pct": gross_pnl / (self.initial_capital) * 100
})
return gross_pnl
def compute_metrics(self) -> dict:
"""
Calcule les métriques de performance du backtest.
"""
df_trades = pd.DataFrame(self.trades)
closed_trades = df_trades[df_trades['action'] == 'CLOSE']
if len(closed_trades) == 0:
return {}
returns = closed_trades['pnl'].values / self.initial_capital
metrics = {
"Total P&L": closed_trades['pnl'].sum(),
"Nombre de trades": len(closed_trades),
"Win rate": (closed_trades['pnl'] > 0).mean() * 100,
"Profit moyen": closed_trades[closed_trades['pnl'] > 0]['pnl'].mean() if (closed_trades['pnl'] > 0).any() else 0,
"Perte moyenne": closed_trades[closed_trades['pnl'] < 0]['pnl'].mean() if (closed_trades['pnl'] < 0).any() else 0,
"Sharpe Ratio": returns.mean() / returns.std() if returns.std() > 0 else 0,
"Max Drawdown": self._max_drawdown(closed_trades['pnl'].cumsum().values),
"ROI annualisé": (self.capital / self.initial_capital - 1) * 100
}
return metrics
def _max_drawdown(self, cumulative_pnl: np.ndarray) -> float:
"""Calcule le drawdown maximum."""
peak = np.maximum.accumulate(cumulative_pnl)
drawdown = (cumulative_pnl - peak) / self.initial_capital * 100
return drawdown.min()
Exemple de backtest
backtester = OptionsBacktester(initial_capital=50_000)
Simuler des trades sur 6 mois
np.random.seed(42)
current_date = datetime.now()
for i in range(20): # 20 trades sur 6 mois
entry_spot = 60_000 + np.random.randn() * 5000
expiry = current_date + timedelta(days=np.random.choice([7, 14, 30, 60]))
iv = 0.8 + np.random.rand() * 0.5 # 80-130% IV
position = backtester.open_straddle(
spot=entry_spot,
strike_pct=0.0, # ATM
expiry=expiry,
iv_atm=iv,
premium=entry_spot * 0.05, # ~5% du spot
quantity=1
)
# Simuler expiration après X jours
exit_date = expiry
exit_spot = entry_spot * (1 + np.random.randn() * 0.15)
realized_vol = iv * (0.7 + np.random.rand() * 0.6)
pnl = backtester.close_position(position, exit_spot, exit_date, realized_vol)
current_date += timedelta(days=10)
Afficher résultats
metrics = backtester.compute_metrics()
print("\n" + "="*50)
print("RÉSULTATS BACKTEST")
print("="*50)
for key, value in metrics.items():
print(f"{key}: {value:.2f}" if isinstance(value, float) else f"{key}: {value}")Intégration avec les APIs d'Analyse IA
Pour enrichir votre analyse, vous pouvez utiliser l'API HolySheep AI pour traiter automatiquement les données de volatilité et générer des insights. HolySheep offre des latences inférieures à 50ms et des tarifs compétitifs :
import aiohttp
import asyncio
from typing import Dict, List
class VolatilityAnalyzer:
"""
Utilise l'API HolySheep pour analyser les surfaces de volatilité
et générer des signaux de trading.
"""
def __init__(self, api_key: str = None):
# IMPORTANT: Utiliser uniquement https://api.holysheep.ai/v1
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key or "YOUR_HOLYSHEEP_API_KEY"
async def analyze_volatility_regime(self, surface_data: Dict) -> Dict:
"""
Analyse le régime de volatilité actuel via GPT-4.1.
Coût estimé: ~$0.008 pour une analyse (1000 tokens)
vs $0.12 avec OpenAI (économie ~93%)
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
prompt = f"""
Analyse cette surface de volatilité BTC Options:
- Strikes: {surface_data.get('strikes', [])[:10]}... (extraits)
- Maturités (jours): {surface_data.get('maturities', [])}
- IV Range: {surface_data.get('iv_min', 0):.1%} - {surface_data.get('iv_max', 0):.1%}
- IV ATM (30d): {surface_data.get('iv_atm_30d', 0):.1%}
- Skew: {surface_data.get('skew', 0):.3f}
- Term Structure: {surface_data.get('term_structure', [])}
Questions:
1. Le skew est-il baissier ou haussier? (anticipe-t-on une baisse?)
2. La term structure est-elle en backwardation ou contango?
3. Recommandations de stratégies d'options?
"""
payload = {
"model": "gpt-4.1", # $8/MTok - excellent rapport qualité/prix
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 800
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
return {
"analysis": result['choices'][0]['message']['content'],
"usage": result.get('usage', {}),
"cost_usd": result['usage']['total_tokens'] * 8 / 1_000_000
}
else:
raise Exception(f"API Error: {response.status}")
async def main():
analyzer = VolatilityAnalyzer()
# Données de surface simulées
surface_data = {
"strikes": [55000, 60000, 65000, 70000, 75000, 80000],
"maturities": [7, 14, 30, 60, 90],
"iv_min": 0.65,
"iv_max": 1.35,
"iv_atm_30d": 0.85,
"skew": -0.12, # Skew baissier (puts plus chères)
"term_structure": [1.25, 1.10, 0.85, 0.72, 0.68]
}
result = await analyzer.analyze_volatility_regime(surface_data)
print(f"Analyse: {result['analysis']}")
print(f"Coût: ${result['cost_usd']:.4f}")
Exécuter
asyncio.run(main())
===
AVANTAGES HOLYSHEEP
===
• Taux de change ¥1=$1 (économie 85%+ sur les tarifs chinois)
• Méthodes de paiement: WeChat Pay, Alipay, cartes internationales
• Latence <50ms pour analyses en temps réel
• Crédits gratuits pour nouveaux utilisateurs
• Modèles disponibles: GPT-4.1 $8, Claude Sonnet 4.5 $15,
Gemini 2.5 Flash $2.50, DeepSeek V3.2 $0.42/MTok
===
Lien d'inscription: https://www.holysheep.ai/register
Optimisation: Téléchargement Parallèle Efficace
Pour les grands jeux de données, utilisez le téléchargement parallèle pour réduire le temps d'exécution de 70% :
import asyncio
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import List
import aiohttp
@dataclass
class ChunkConfig:
start_date: str
end_date: str
chunk_days: int = 30
class ParallelDownloader:
"""
Télécharge les données en parallèle par chunks de temps.
Réduit le temps total de 70%+ vs téléchargement séquentiel.
"""
def __init__(self, tardis_key: str, max_concurrent: int = 5):
self.api_key = tardis_key
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
def split_into_chunks(self, start: str, end: str, chunk_days: int = 30) -> List[ChunkConfig]:
"""Découpe la période en chunks de X jours."""
from datetime import datetime, timedelta
start_dt = datetime.strptime(start, "%Y-%m-%d")
end_dt = datetime.strptime(end, "%Y-%m-%d")
chunks = []
current = start_dt
while current < end_dt:
chunk_end = min(current + timedelta(days=chunk_days), end_dt)
chunks.append(ChunkConfig(
start_date=current.strftime("%Y-%m-%d"),
end_date=chunk_end.strftime("%Y-%m-%d")
))
current = chunk_end + timedelta(days=1)
return chunks
async def download_chunk(self, session: aiohttp.ClientSession,
chunk: ChunkConfig) -> pd.DataFrame:
"""Télécharge un chunk avec rate limiting."""
async with self.semaphore:
print(f"↓ Téléchargement: {chunk.start_date} → {chunk.end_date}")
# Simulation du téléchargement (remplacer par vrai appel API)
await asyncio.sleep(0.5) # Latence réseau simulée
# Retourner DataFrame vide pour la démonstration
return pd.DataFrame()
async def download_all(self, start: str, end: str) -> List[pd.DataFrame]:
"""Télécharge tous les chunks en parallèle."""
chunks = self.split_into_chunks(start, end)
print(f"📦 {len(chunks)} chunks à télécharger en parallèle")
async with aiohttp.ClientSession() as session:
tasks = [
self.download_chunk(session, chunk)
for chunk in chunks
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filtrer les erreurs
valid_results = [r for r in results if isinstance(r, pd.DataFrame)]
print(f"✓ {len(valid_results)}/{len(chunks)} chunks téléchargés")
return valid_results
Benchmark: séquentiel vs parallèle
async def benchmark():
downloader = ParallelDownloader("fake_key")
# Test avec 180 jours
import time
# Séquentiel (simulé)
start = time.time()
chunks = downloader.split_into_chunks("2024-01-01", "2024-06-30", 30)
# simulate_sequential = sum(len(downloader.download_chunk(None, c)) for c in chunks) # Non disponible
print(f"Séquentiel estimé: {len(chunks) * 2:.1f}s")
# Parallèle
start = time.time()
results = await downloader.download_all("2024-01-01", "2024-06-30")
parallel_time = time.time() - start
print(f"Parallèle (5 connexions): {parallel_time:.1f}s")
print(f"Gain: {(len(chunks) * 2 - parallel_time) / (len(chunks) * 2) * 100:.0f}%")
asyncio.run(benchmark())Erreurs courantes et solutions
Erreur 1: 403 Forbidden - Quota dépassé
# ❌ ERREUR:
403 Forbidden: "Daily download limit exceeded"
Upgrade your plan at https://tardis.dev/plans
✅ SOLUTION:
Vérifier le quota restant et implémenter un cache local
import os
import json
from pathlib import Path
from datetime import datetime, timedelta
class CachedDataManager:
"""
Gère un cache local des données téléchargées.
Évite de re-télécharger les données déjà obtenues.
"""
def __init__(self, cache_dir: str = "./data_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def get_cache_key(self, exchange: str, symbol: str,
start: str, end: str) -> str:
return f"{exchange}_{symbol}_{start}_{end}.parquet"
def is_cache_valid(self, cache_file: Path, max_age_days: int = 1) -> bool:
"""Vérifie si le cache existe et n'est pas trop ancien."""
if not cache_file.exists():
return False
age = datetime.now() - datetime.fromtimestamp(cache_file.stat().st_mtime)
return age.days < max_age_days
def get_cached_data(self, exchange: str, symbol: str,
start: str, end: str) -> Optional[pd.DataFrame]:
"""Récupère les données depuis le cache si valide."""
cache_key = self.get_cache_key(exchange, symbol, start, end)
cache_file = self.cache_dir / cache_key
if self.is_cache_valid(cache_file):
print(f"📂 Cache hit: {cache_key}")
return pd.read_parquet(cache_file)
print(f"📂 Cache miss: {cache_key}")
return None
def save_to_cache(self, df: pd.DataFrame, exchange: str, symbol: str,
start: str, end: str):
"""Sauvegarde les données dans le cache."""
cache_key = self.get_cache_key(exchange, symbol, start, end)
cache_file = self.cache_dir / cache_key
df.to_parquet(cache_file)
print(f"💾 Sauvegardé: {cache_file}")
Utilisation avec retry conditionnel
cache = CachedDataManager()
df = cache.get_cached_data("deribit", "BTC-*", "2024-06-01", "2024-06-30")
if df is None:
try:
downloader = DeribitOptionsDownloader()
df = await downloader.download_with_retry(
"deribit", "BTC-*", "2024-06-01", "2024-06-30"
)
cache.save_to_cache(df, "deribit", "BTC-*", "2024-06-01", "2024-06-30")
except PermissionError as e:
print(f"Quota épuisé. Utilisation des données en cache plus anciennes.")
# Fallback vers cache même si ancien
old_cache = cache.cache_dir / "deribit_BTC-*_2024-06-01_2024-06-30.parquet"
if old_cache.exists():
df = pd.read_parquet(old_cache)Erreur 2: MemoryError lors du chargement de gros fichiers
# ❌ ERREUR:
MemoryError: Unable to allocate 8.5 GiB for an array
✅ SOLUTION:
Traitement par chunks et utilisation de dtype optimisés
class MemoryEfficientLoader:
"""
Charge les données CSV/Parquet de manière memory-efficient.
Réduit la RAM utilisée de 80% via dtype compression.
"""
# Mapping des types pour compression
DTYPE_MAPPING = {
'timestamp': 'int64', # Unix ms au lieu de datetime
'price': 'float32', # float64 → float32
'volume': 'float32',
'amount': 'float32',
'instrument_name': 'category', # String → category
}
def load_chunked(self, file_path: str, chunk_size: int = 50_000):
"""
Charge les données par chunks pour éviter MemoryError.
"""
for chunk in pd.read_csv(
file_path,
chunksize=chunk_size,
dtype={col: 'float32' for col in ['best_bid_price', 'best_ask_price']},
parse_dates=['timestamp'],
usecols=['timestamp', 'instrument_name', 'best_bid_price',
'best_ask_price', 'mark_iv']
):
yield chunk
def aggregate_efficient(self, file_path: str) -> pd.DataFrame:
"""
Agrège directement sans charger tout le fichier en RAM.
"""
total_iv = 0
count = 0
for chunk in self.load_chunked(file_path):
# Agrégation incrémentale
total_iv += chunk['mark_iv'].sum()
count += len(chunk)
# Forcer garbage collection
del chunk
return {'mean_iv': total_iv / count if count > 0 else 0}
Utilisation
loader = MemoryEfficientLoader()
Au lieu de: df = pd.read_csv("huge_file.csv") # MemoryError!
Faire:
for chunk in loader.load_chunked("huge_btc_options.csv"):
process(chunk) # Traiter chaque chunk séparémentErreur 3: Interpolation instable de la surface de volatilité
# ❌ ERREUR:
RuntimeWarning: invalid value encountered in greater_equal
Surface avec trous et artefacts
✅ SOLUTION:
Nettoyage des données + interpolation robuste
class RobustVolatilitySurface:
"""
Construit une surface de volatilité propre avec gestion des outliers.
"""
def __init__(