Dernière mise à jour : 6 mai 2026 | Difficulté : Intermédiaire-avancé | Temps de lecture : 18 minutes
Introduction et Contexte
En tant qu'analyste quantitatif specialize dans les produits derives de crypto-actifs, j'ai passe les six derniers mois a construire des pipelines de donnees pour alimenter mes modeles de tarification d'options. Le defi ? Obtenir des donnees historiques fiables d'IV surface (Implicit Volatility Surface) pour les options sur BTC et ETH, puis les ingerer efficacement dans un data warehouse Parquet optimise pour les requetes analytiques.
Dans ce tutoriel terrain, je vais vous presenter ma stack complete : Tardis API pour la collecte brute des donnees de marche, et HolySheep AI pour le traitement intelligent et la transformation. Nous couvrons tout, du raw data au Parquet optimise, en passant par la validation et la gestion des erreurs.
Avertissement transparence : J'utilise HolySheep AI quotidiennement dans ma pipeline. S'inscrire ici si vous souhaitez tester vous-meme leur infrastructure.
Architecture du Pipeline
┌─────────────────────────────────────────────────────────────────────────────┐
│ PIPELINE IV SURFACE COMPLET │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌────────────┐│
│ │ TARDIS │───▶│ HOLYSHEEP │───▶│ PARQUET │───▶│ ANALYTICS││
│ │ API Client │ │ AI Process │ │ INGESTION │ │ DASHBOARD││
│ └──────────────┘ └──────────────┘ └──────────────┘ └────────────┘│
│ │ │ │ │
│ ▼ ▼ ▼ │
│ Raw JSON/CSV LLM Enrichment Schema Evolution │
│ Historical Data Data Validation Compression Zstd │
│ 50+ exchanges <50ms latency Columnar format │
│ │
└─────────────────────────────────────────────────────────────────────────────┘Prérequis et Configuration Initiale
- Python 3.11+ avec support asyncio
- Clef API Tardis (plan Historical Data)
- Compte HolySheep AI avec credits (inscription ici)
- Environnement avec pandas, pyarrow, aiohttp pre-installes
- Depot S3-compatible ou stockage local pour les fichiers Parquet
Installation des Dépendances
# Installation complete des dependances Python
pip install --upgrade \
tardis-client==2.0.18 \
pandas==2.2.2 \
pyarrow==17.0.0 \
fastparquet==2024.2.0 \
aiohttp==3.9.5 \
asyncio-dgram==3.0.2 \
python-dotenv==1.0.1
Variables d'environnement a configurer
cat >> ~/.bashrc << 'EOF'
export TARDIS_API_KEY="your_tardis_key_here"
export HOLYSHEEP_API_KEY="your_holysheep_key_here"
export DATA_OUTPUT_DIR="/data/iv_surface_parquet"
EOF
source ~/.bashrc1. Téléchargement des Données IV Surface via Tardis API
Tardis fournit des donnees de marche crypto de niveau institutionnel avec une latence mediane de 45ms pour les requetes historiques. Leur couverture inclut plus de 50 exchanges, ce qui est critique pour reconstruire une IV surface fiable sur Deribit, Binance Options et OKX.
#!/usr/bin/env python3
"""
Tardis IV Surface Data Downloader
==================================
Télécharge les données d'options BTC/ETH avec strikes, expirations et IV implicite.
"""
import os
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class IVSurfaceData:
"""Structure normalisee pour les donnees IV Surface."""
timestamp: datetime
exchange: str
underlying: str # BTC, ETH
strike: float
expiry: datetime
option_type: str # call, put
iv: float # Implicit volatility (decimal, ex: 0.65 = 65%)
spot: float
bid: float
ask: float
volume: float
open_interest: float
raw_instrument_name: str
class TardisIVDownloader:
"""Client asynchrone pour l'API Tardis - donnees d'options crypto."""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self.rate_limit_ms = 100 # Limite Tardis: 10 req/sec
async def __aenter__(self):
connector = aiohttp.TCPConnector(limit=20, limit_per_host=10)
self.session = aiohttp.ClientSession(
connector=connector,
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_iv_surface_snapshot(
self,
exchange: str,
underlying: str,
date: datetime
) -> List[Dict]:
"""
Recupere un snapshot de l'IV surface pour une date donnee.
Args:
exchange: Nom de l'exchange (deribit, binance, okx)
underlying: Sous-jacent (BTC, ETH)
date: Date du snapshot (UTC)
Returns:
Liste de dictionnaires contenant les donnees IV surface
"""
# Construction de la requete selon format Tardis
date_str = date.strftime("%Y-%m-%d")
url = f"{self.BASE_URL}/historical/{exchange}/option_summary"
params = {
"date": date_str,
"symbol": f"{underlying}-*", # Tous les strikes
"format": "json",
"apikey": self.api_key
}
try:
async with self.session.get(url, params=params) as response:
if response.status == 429:
logger.warning(f"Rate limit atteint, attente 1s...")
await asyncio.sleep(1)
return await self.fetch_iv_surface_snapshot(
exchange, underlying, date
)
response.raise_for_status()
data = await response.json()
# Filtrage des instruments pertinents
filtered = [
item for item in data
if item.get("underlying") == underlying.lower()
and item.get("iv") is not None
]
logger.info(
f"Snapshot {exchange}/{underlying} {date_str}: "
f"{len(filtered)} instruments recuperes"
)
return filtered
except aiohttp.ClientError as e:
logger.error(f"Erreur requete {exchange}: {e}")
return []
async def download_range(
self,
exchanges: List[str],
underlyings: List[str],
start_date: datetime,
end_date: datetime
) -> List[IVSurfaceData]:
"""Telecharge les snapshots sur une periode donnee."""
all_data = []
current = start_date
while current <= end_date:
for exchange in exchanges:
for underlying in underlyings:
snapshots = await self.fetch_iv_surface_snapshot(
exchange, underlying, current
)
for snap in snapshots:
all_data.append(IVSurfaceData(
timestamp=current,
exchange=exchange,
underlying=snap.get("underlying", underlying).upper(),
strike=snap.get("strike", 0),
expiry=datetime.fromisoformat(
snap.get("expiry", current.isoformat())
),
option_type=snap.get("option_type", "call"),
iv=snap.get("iv", 0),
spot=snap.get("spot", 0),
bid=snap.get("bid", 0),
ask=snap.get("ask", 0),
volume=snap.get("volume", 0),
open_interest=snap.get("open_interest", 0),
raw_instrument_name=snap.get("instrument_name", "")
))
await asyncio.sleep(self.rate_limit_ms / 1000)
current += timedelta(days=1)
return all_data
=============================================================================
UTILISATION PRATIQUE
=============================================================================
async def main():
tardis_key = os.getenv("TARDIS_API_KEY")
async with TardisIVDownloader(tardis_key) as downloader:
# Telechargement 30 jours de donnees BTC/ETH sur 3 exchanges
data = await downloader.download_range(
exchanges=["deribit", "binance", "okx"],
underlyings=["BTC", "ETH"],
start_date=datetime(2026, 4, 1),
end_date=datetime(2026, 4, 30)
)
logger.info(f"Total records recuperes: {len(data)}")
# Export JSON interim
import pandas as pd
df = pd.DataFrame([vars(d) for d in data])
df.to_json("/tmp/iv_raw.json", orient="records", date_format="iso")
logger.info("Export JSON termine: /tmp/iv_raw.json")
if __name__ == "__main__":
asyncio.run(main())2. Enrichissement et Validation avec HolySheep AI
A ce stade, nous avons des donnees brutes de 30 jours. Le probleme ? Les lacunes : donnees manquantes pour certains strikes, IV aberrantes, incoherences entre exchanges. C'est la que HolySheep AI intervient avec son infrastructure <50ms de latence et ses modeles de traitement.
Dans mon usage quotidien, HolySheep me fait economiser 85% sur les couts API grace a leur taux de change favorable (¥1 = $1). Pour un traitement comme celui-ci qui necessite 150 000 tokens, le cout est infinitesimal compare a l'utilisation directe d'OpenAI ou Anthropic.
#!/usr/bin/env python3
"""
HolySheep AI - Enrichissement et Validation des Donnees IV Surface
=====================================================================
Utilise les modeles LLM de HolySheep pour:
1. Detecter et corriger les anomalies IV
2. Completer les strikes manquants via interpolation
3. Valider la coherence cross-exchange
4. Generer des metadonnees enrichies
"""
import os
import json
import pandas as pd
from typing import List, Dict, Optional
from datetime import datetime
from openai import OpenAI, AsyncOpenAI
import asyncio
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
=============================================================================
CONFIGURATION HOLYSHEEP AI - BASE URL OFFICIELLE
=============================================================================
class HolySheepAIClient:
"""
Client HolySheep AI pour l'enrichissement des donnees financieres.
BASE URL: https://api.holysheep.ai/v1
AVANTAGES:
- Latence mediane: <50ms
- Taux: ¥1 = $1 (economies 85%+ vs OpenAI)
- Support: WeChat, Alipay
- Credits gratuits pour nouveaux inscrits
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = OpenAI(
api_key=api_key,
base_url=self.BASE_URL,
timeout=30.0,
max_retries=3
)
self.async_client = AsyncOpenAI(
api_key=api_key,
base_url=self.BASE_URL,
timeout=30.0
)
# Modeles disponibles (tarifs 2026)
self.models = {
"gpt4_1": "gpt-4.1", # $8/MTok - Haute precision
"claude_sonnet": "claude-sonnet-4.5", # $15/MTok - Analyse fine
"gemini_flash": "gemini-2.5-flash", # $2.50/MTok - Rapidite
"deepseek_v3": "deepseek-v3.2" # $0.42/MTok - Cout minimal
}
def validate_iv_anomalies(
self,
df: pd.DataFrame,
model: str = "deepseek_v3"
) -> pd.DataFrame:
"""
Detecte et corrige les anomalies IV via analyse LLM.
Le modele DeepSeek V3.2 a $0.42/MTok est ideal pour cette tache
de validation structurale - pas besoin de GPT-4.1 pour de la
detection de seuils.
"""
# Preparation du prompt avec echantillon representatif
sample = df.head(50).to_dict(orient="records")
sample_json = json.dumps(sample, indent=2, default=str)
prompt = f"""
Tu es un analyste quantitatif specialise en options crypto.
Analyse le JSON suivant contenant des donnees IV Surface et identifie:
1. **IV aberrantes**: IV < 0.20 (trop faible) ou IV > 2.50 (trop eleve)
2. **Spread anormaux**: (ask - bid) / mid > 0.15 (spread > 15%)
3. **Inversions put/call**: IV put > IV call pour meme strike/expiry (rare mais possible en marché stresse)
4. **Strikes incoherents**: strikes non-alignes sur grille standard
Pour chaque enregistrement problematic, retourne:
- "action": "keep", "correct", ou "remove"
- "correction": valeur corrigee si applicable
- "reason": explication courte
{sample_json}
Reponse en JSON format strict:
"""
response = self.client.chat.completions.create(
model=self.models[model],
messages=[
{
"role": "system",
"content": "Tu es un expert en analyse quantitative. Reponds UNIQUEMENT en JSON valide."
},
{"role": "user", "content": prompt}
],
temperature=0.1,
response_format={"type": "json_object"}
)
# Parsing et application des corrections
corrections = json.loads(response.choices[0].message.content)
df_corrected = df.copy()
correction_count = 0
for idx, correction in enumerate(corrections.get("corrections", [])):
if idx < len(df_corrected) and correction.get("action") == "correct":
if "iv" in correction:
df_corrected.loc[idx, "iv"] = correction["iv"]
correction_count += 1
elif correction.get("action") == "remove":
df_corrected.loc[idx, "iv"] = None
df_corrected = df_corrected.dropna(subset=["iv"])
print(f"Corrections appliquees: {correction_count}/{len(df)}")
print(f" Cout estime: ${len(sample) * 0.15 / 1000000:.6f}")
return df_corrected
async def interpolate_missing_strikes(
self,
df: pd.DataFrame,
expiry: datetime,
underlying: str,
max_missing_pct: float = 0.10
) -> pd.DataFrame:
"""
Complete les strikes manquants via interpolation IV smoothe.
Utilise Gemini 2.5 Flash ($2.50/MTok) pour le traitement batch
- rapport qualite/vitesse optimal pour cette operation.
"""
# Identification des strikes presents et manquants
strikes_present = set(df["strike"].unique())
strike_min = min(strikes_present)
strike_max = max(strikes_present)
# Grille standard: chaque 1000$ pour BTC, 100$ pour ETH
step = 1000 if underlying == "BTC" else 100
expected_strikes = set(
range(int(strike_min), int(strike_max) + step, step)
)
strikes_missing = expected_strikes - strikes_present
if not strikes_missing:
return df # Pas de interpolation necessaire
missing_pct = len(strikes_missing) / len(expected_strikes)
if missing_pct > max_missing_pct:
print(f"Attention: {missing_pct:.1%} strikes manquants, limite {max_missing_pct:.1%}")
# Construction du prompt pour interpolation
iv_curve_data = df[df["expiry"] == expiry][["strike", "iv", "option_type"]]
iv_curve_json = iv_curve_data.to_dict(orient="records")
prompt = f"""
Pour l'expiration {expiry.isoformat()}, interpole les IV manquantes.
Strikes presents: {sorted(strikes_present)}
Strikes manquants: {sorted(strikes_missing)}
Donnnees IV courbe:
{json.dumps(iv_curve_json, indent=2)}
Methode: interpolation lineaire entre strikes adjacents + ajustement pour skew.
Reponde en JSON:
{{
"interpolated": [
{{"strike": 45000, "iv_call": 0.62, "iv_put": 0.65}},
...
]
}}
"""
response = await self.async_client.chat.completions.create(
model=self.models["gemini_flash"],
messages=[
{"role": "system", "content": "Expert interpolation IV."},
{"role": "user", "content": prompt}
],
temperature=0.1,
response_format={"type": "json_object"}
)
# Application des interpolations
interpolated = json.loads(response.choices[0].message.content)
new_rows = []
for item in interpolated.get("interpolated", []):
new_rows.append({
"timestamp": df["timestamp"].iloc[0],
"expiry": expiry,
"strike": item["strike"],
"iv": (item.get("iv_call", 0) + item.get("iv_put", 0)) / 2,
"interpolated": True
})
df_interpolated = pd.concat([df, pd.DataFrame(new_rows)], ignore_index=True)
print(f"Strikes interpoles: {len(new_rows)}")
return df_interpolated
async def batch_enrich_metadata(
self,
records: List[Dict],
batch_size: int = 100
) -> List[Dict]:
"""
Enrichit les enregistrements avec metadonnees contextuelles.
Utilise DeepSeek V3.2 ($0.42/MTok) pour minimiser les couts
sur les volumes eleves de donnees financieres.
"""
enriched = []
for i in range(0, len(records), batch_size):
batch = records[i:i + batch_size]
prompt = f"""
Analyse ce batch de {len(batch)} enregistrements IV Surface et ajoute:
1. **Moneyness**: OTM/ATM/ITM base sur spot vs strike
2. **Risk category**: LOW (<40 IV), MEDIUM (40-80), HIGH (>80)
3. **Liquidity tier**: ILLIQUID (<$10k OI), STANDARD, LIQUID (>$1M OI)
{json.dumps(batch, indent=2, default=str)}
Reponse JSON:
{{
"enriched": [
{{"moneyness": "OTM", "risk_category": "MEDIUM", "liquidity_tier": "STANDARD", ...}},
...
]
}}
"""
response = await self.async_client.chat.completions.create(
model=self.models["deepseek_v3"],
messages=[
{"role": "system", "content": "Expert metriques financieres."},
{"role": "user", "content": prompt}
],
temperature=0.1,
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
enriched.extend(result.get("enriched", batch))
print(f"Batch {i//batch_size + 1}/{(len(records)-1)//batch_size + 1} traite")
return enriched
=============================================================================
EXECUTION DU PIPELINE HOLYSHEEP
=============================================================================
async def enrich_pipeline(input_path: str, output_path: str):
"""Pipeline complet d'enrichissement avec HolySheep AI."""
# Chargement des donnees brutes
df = pd.read_json(input_path)
print(f"Donnees chargees: {len(df)} enregistrements")
# Initialisation client HolySheep
holysheep = HolySheepAIClient(os.getenv("HOLYSHEEP_API_KEY"))
# Etape 1: Validation des anomalies IV
print("\n=== Etape 1: Validation des anomalies ===")
df = holysheep.validate_iv_anomalies(df, model="deepseek_v3")
# Etape 2: Interpolation strikes manquants
print("\n=== Etape 2: Interpolation strikes ===")
for expiry in df["expiry"].unique():
for underlying in df["underlying"].unique():
df = await holysheep.interpolate_missing_strikes(
df[df["underlying"] == underlying],
expiry,
underlying
)
# Etape 3: Enrichissement metadonnees
print("\n=== Etape 3: Enrichissement metadonnees ===")
records = df.to_dict(orient="records")
records_enriched = await holysheep.batch_enrich_metadata(records)
# Export
df_enriched = pd.DataFrame(records_enriched)
df_enriched.to_json(output_path, orient="records", indent=2)
print(f"\nExport termine: {output_path}")
print(f"Total enregistrements: {len(df_enriched)}")
if __name__ == "__main__":
asyncio.run(enrich_pipeline(
input_path="/tmp/iv_raw.json",
output_path="/tmp/iv_enriched.json"
))3. Conversion et Ingestion Parquet
La phase finale : transformer les JSON enriched en fichiers Parquet optimises pour les analyses. J'utilise la compression Zstd (ratio ~2.5x vs None) et le partitioning par date/expiry pour des requetes 10x plus rapides sur les filtres temporels.
#!/usr/bin/env python3
"""
Parquet Ingestion Pipeline - Optimisation pour Analyse IV Surface
==================================================================
Conversion JSON -> Parquet avec:
- Schema evolution et validation
- Partitionnement temporel
- Compression Zstd
- Optimisation pour requetes analytiques
"""
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.dataset as ds
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional, Dict
import logging
import json
import gc
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class IVSurfaceParquetPipeline:
"""
Pipeline de conversion IV Surface vers Parquet optimise.
Optimisations:
- Partitionnement: date, expiry, underlying
- Compression: Zstd niveau 3 (rapide, bon ratio)
- Encodage: dictionnaires pour colonnes categorielles
- Types: dates en int96 pour compatibilite Hive/Spark
"""
def __init__(
self,
output_dir: str,
partition_by: List[str] = ["year", "month", "underlying"],
compression: str = "zstd",
compression_level: int = 3
):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.partition_by = partition_by
self.compression = compression
self.compression_level = compression_level
# Schema Parquet optimise
self.schema = pa.schema([
("timestamp", pa.timestamp("us")),
("date_partition", pa.date32()),
("year", pa.int16()),
("month", pa.int8()),
("exchange", pa.dictionary(pa.int8(), pa.string())),
("underlying", pa.dictionary(pa.int8(), pa.string())),
("strike", pa.float64()),
("expiry", pa.timestamp("us")),
("expiry_days", pa.int16()),
("option_type", pa.dictionary(pa.int8(), pa.string())),
("iv", pa.float32()),
("iv_adjusted", pa.float32()), # IV avec skew correction
("spot", pa.float64()),
("bid", pa.float64()),
("ask", pa.float64()),
("mid", pa.float64()),
("spread_pct", pa.float32()),
("volume", pa.float64()),
("open_interest", pa.float64()),
("moneyness", pa.dictionary(pa.int8(), pa.string())),
("risk_category", pa.dictionary(pa.int8(), pa.string())),
("liquidity_tier", pa.dictionary(pa.int8(), pa.string())),
("interpolated", pa.bool_()),
("raw_instrument_name", pa.string()),
("data_quality_score", pa.float32()),
("created_at", pa.timestamp("us"))
])
def prepare_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
"""Preparation et enrichment du DataFrame avant conversion."""
# Conversion des dates
df["timestamp"] = pd.to_datetime(df["timestamp"])
df["expiry"] = pd.to_datetime(df["expiry"])
# Partitions temporelles
df["date_partition"] = df["timestamp"].dt.date
df["year"] = df["timestamp"].dt.year.astype("int16")
df["month"] = df["timestamp"].dt.month.astype("int8")
# Calcul du nombre de jours jusqu'a expiration
df["expiry_days"] = (df["expiry"] - df["timestamp"]).dt.days.astype("int16")
# Mid price et spread
df["mid"] = (df["bid"] + df["ask"]) / 2
df["spread_pct"] = (df["ask"] - df["bid"]) / df["mid"]
# Score de qualite base sur complétude et coherence
df["data_quality_score"] = (
(df["iv"].notna().astype(float) * 0.4) +
((df["spread_pct"] < 0.10).astype(float) * 0.3) +
((df["open_interest"] > 1000).astype(float) * 0.3)
)
# Valeurs par defaut pour colonnes optionnelles
df["interpolated"] = df.get("interpolated", False)
df["iv_adjusted"] = df.get("iv_adjusted", df["iv"])
# Timestamp de creation
df["created_at"] = datetime.utcnow()
# Selection et reorder des colonnes selon schema
df = df[[
"timestamp", "date_partition", "year", "month",
"exchange", "underlying", "strike", "expiry", "expiry_days",
"option_type", "iv", "iv_adjusted", "spot",
"bid", "ask", "mid", "spread_pct",
"volume", "open_interest",
"moneyness", "risk_category", "liquidity_tier",
"interpolated", "raw_instrument_name", "data_quality_score",
"created_at"
]]
logger.info(f"DataFrame prepare: {len(df)} lignes, {len(df.columns)} colonnes")
return df
def write_parquet(
self,
df: pd.DataFrame,
table_name: str = "iv_surface",
target_file_size_mb: int = 128
) -> Dict:
"""
Ecriture Parquet avec partitioning et optimisation.
Args:
df: DataFrame pret
table_name: Nom de la table
target_file_size_mb: Taille cible par fichier (optimisation storage/req)
Returns:
Dict avec statistiques d'ecriture
"""
start_time = datetime.now()
# Conversion PyArrow
table = pa.Table.from_pandas(
df,
schema=self.schema,
preserve_index=False
)
# Stats avant compression
row_count = len(df)
original_size_mb = df.memory_usage(deep=True).sum() / 1024 / 1024
# Ecriture avec partitioning
output_path = self.output_dir / table_name
pq.write_to_dataset(
table,
root_path=str(output_path),
partition_cols=self.partition_by,
compression=self.compression,
use_dictionary=True,
write_statistics=True,
data_page_size=8 * 1024 * 1024, # 8MB data pages
)
# Calcul des fichiers generes
file_count = len(list(output_path.rglob("*.parquet")))
# Taille finale estimee (lecture d'un echantillon)
sample_files = list(output_path.rglob("*.parquet"))[:5]
if sample_files:
sample_size = sum(f.stat().st_size for f in sample_files)
avg_size = sample_size / len(sample_files)
estimated_total_mb = (avg_size * file_count) / 1024 / 1024
else:
estimated_total_mb = original_size_mb / 2.5 # Estimation ratio Zstd
duration = (datetime.now() - start_time).total_seconds()
stats = {
"table_name": table_name,
"output_path": str(output_path),
"row_count": row_count,
"original_size_mb": round(original_size_mb, 2),
"estimated_compressed_mb": round(estimated_total_mb, 2),
"compression_ratio": round(original_size_mb / max(estimated_total_mb, 0.1), 2),
"file_count": file_count,
"partition_cols": self.partition_by,
"duration_seconds": round(duration, 2)
}
logger.info(f"""
╔══════════════════════════════════════════════════════════╗
║ PARQUET INGESTION TERMINE ║
╠══════════════════════════════════════════════════════════╣
║ Lignes: {row_count:>10,} ║
║ Taille originale: {original_size_mb:>6.1f} MB ║
║ Taille compressee: {estimated_total_mb:>6.1f} MB (estime) ║
║ Ratio compression: {stats['compression_ratio']:>6.2f}x ║
║ Fichiers generes: {file_count:>5} ║
║ Duree: {duration:>5.2f}s ║
╚══════════════════════════════════════════════════════════╝
""")
return stats
def query_sample(self, n_rows: int = 10) -> pd.DataFrame:
"""Lecture d'un echantillon pour verification."""
table = ds.dataset(
str(self.output_dir / "iv_surface"),
format="parquet"
).to_table()
return table.slice(0, n_rows).to_pandas()
=============================================================================
EXECUTION
=============================================================================
def run_ingestion_pipeline(
input_json: str,
output_dir: str,
table_name: str = "iv_surface"
):
"""Execute le pipeline complet d'ingestion."""
logger.info(f"Chargement depuis: {input_json}")
df = pd.read_json(input_json)
pipeline = IVSurfaceParquetPipeline(
output_dir=output_dir,
partition_by=["year", "month", "underlying"],
compression="zstd",
compression_level=3
)
# Preparation
df_prepared = pipeline.prepare_dataframe(df)
# Ecriture Parquet
stats = pipeline.write_parquet(df_prepared, table_name)
# Verification
sample = pipeline.query_sample(5)
print("\nEchantillon des 5 premieres lignes:")
print(sample[["timestamp", "underlying", "strike", "iv", "exchange"]])
# Export des statistiques
stats_path = Path(output_dir) / table_name / "_stats.json"
with open(stats_path, "w") as f:
json.dump(stats, f, indent=2, default=str)
return stats
if __name__ == "__main__":
stats = run_ingestion_pipeline(
input_json="/tmp/iv_enriched.json",
output_dir="/data/iv_surface_parquet",
table_name="iv_surface"
)4. Requêtes Analytiques sur le Parquet
Une fois les donnees en Parquet, les performances de requete sont dramatiques. Sur mon dataset de 30 jours (~4.2M lignes), une requete sur 7 jours de donnees s'execute en 340ms contre 8.5s sur le JSON equivalent.
#!/usr/bin/env python3
"""
Analytique IV Surface - Exemples de Requetes Parquet
=====================================================
"""
import pyarrow.dataset as ds
import pyarrow.compute as pc
import pandas as pd
from datetime import datetime, timedelta
def query_iv_surface(
data_path: str,
start_date: datetime,
end_date: datetime,
underlying: str = "BTC"
) -> pd.DataFrame:
"""
Requete optimisée sur le Parquet partitionne.
Utilise le partitionnement pour filter seulement