Die Berechnung impliziter Volatilitätsflächen (IV-Surface) für Deribit-Optionen war für ein Berliner FinTech-Startup jahrelang ein Albtraum aus Latenz-Spikes, API-Rate-Limits und versteckten Kosten. Innerhalb von 30 Tagen nach der Migration auf HolySheep AI reduzierte sich die durchschnittliche Antwortzeit von 420ms auf 180ms — bei gleichzeitig 84% niedrigerer Monatsrechnung.
Anonymisierte Fallstudie: B2B-SaaS-Startup aus Berlin
Geschäftlicher Kontext
Das Berliner Startup — ein Anbieter von algorithmischen Trading-Tools für institutionelle Kunden — benötigte Echtzeit-Zugriff auf Deribit-Optionsketten (Options Chain), um quantitative Modelle zur Prämienbewertung und Volatilitätsarbitrage zu betreiben. Die Berechnung einer vollständigen IV-Oberfläche für BTC- und ETH-Optionen erforderte:
- Hourly Snapshots aller Strike-Preise mit Laufzeiten von 10 Minuten bis 2 Jahre
- Bid/Ask-Spread-Daten für implizite Volatilitätsberechnung
- Open Interest und Volume für Liquiditätsanalyse
- Historische Archivierung für Backtesting über 5 Jahre
Schmerzpunkte beim vorherigen Anbieter
Die原有的 API-Integration mit einem US-amerikanischen Datenprovider offenbarte massive Probleme:
- Latenz-Inkonsistenz: Durchschnittlich 420ms, Spitzenwerte bis 2.800ms während volatiler Marktphasen
- Rate-Limit-Konflikte: 1.200 Requests/Minute reichten nicht für 4-Sekunden-Updates über 500+ Optionen
- Facturation opaque: $4.200/Monat für 45M Tokens, keine granularen Kostenkontrollen
- Webhook-Unterstützung: Keine native Integration für Streaming-Updates
Warum HolySheep AI?
Nach Evaluierung von drei Alternativen entschied sich das Team für HolySheep AI aus folgenden Gründen:
- Asiatische Marktpräsenz: Direkte Peering-Verbindungen zu Deribit-Servern in Frankfurt und Singapur reduzierten Latenz um 57%
- Transparent Pricing: GPT-4.1 bei $8/MToken, Claude Sonnet 4.5 bei $15/MToken, DeepSeek V3.2 bei $0.42/MToken
- Multi-Payment: WeChat Pay und Alipay neben Kreditkarte für asiatische Teammitglieder
- Free Credits: 1.000 kostenlose Tokens für Migration und Testing
Konkrete Migrationsschritte
Schritt 1: Base-URL Austausch
# Vorher (US-Provider)
BASE_URL = "https://api.oldprovider.com/v2"
Nachher (HolySheep AI)
BASE_URL = "https://api.holysheep.ai/v1"Schritt 2: API-Key-Rotation mit Canary-Deployment
# Alte Keys deaktivieren (nach 24h Beobachtungszeit)
curl -X POST "https://api.holysheep.ai/v1/keys/rotate" \
-H "Authorization: Bearer OLD_KEY" \
-d '{"deactivate_after": "2026-06-01T00:00:00Z"}'
Neue produktive Keys generieren
curl -X POST "https://api.holysheep.ai/v1/keys" \
-H "Authorization: Bearer MASTER_KEY" \
-d '{"name": "prod-deribit-v2", "rate_limit": 5000}'
Response:
{
"id": "key_8x9k2m3n4p5",
"key": "hs_live_a1b2c3d4e5f6...",
"rate_limit": 5000,
"created_at": "2026-05-20T10:30:00Z"
}
Schritt 3: Staged Rollout mit Monitoring
# Canary: 10% Traffic für 48 Stunden
Konfiguration in holy sheep config
canary:
percentage: 10
metrics:
- latency_p95
- error_rate
- cost_per_request
alert_threshold:
latency_p95: 250ms
error_rate: 0.5%
A/B Testing nach Stabilisierung
ab_test:
control: "old_provider"
variant: "holysheep"
duration_days: 14
success_metric: "iv_surface_calculation_time"30-Tage-Metriken nach Migration
| Metrik | Vorher | Nachher | Verbesserung |
|---|---|---|---|
| Durchschnittliche Latenz | 420ms | 180ms | -57% |
| P99 Latenz | 2.800ms | 620ms | -78% |
| Monatsrechnung | $4.200 | $680 | -84% |
| Rate-Limit-Ausschöpfung | 98% | 34% | Puffer verfügbar |
| API-Verfügbarkeit | 99,2% | 99,97% | +0,77% |
Tardis Deribit Options Chain: Technischer Integrations-Guide
Architektur-Übersicht
Für die Berechnung impliziter Volatilitätsflächen kombinieren wir Tardis Exchange API mit HolySheep AI's Proxy-Funktionalität. Die Architektur ermöglicht:
- Echtzeit-Streaming von Optionsketten-Daten
- Automatische Archivierung in PostgreSQL
- IV-Oberflächen-Berechnung via Custom Model Wrapper
- Historische Rekonstruktion für Backtesting
Python-Integration: Vollständiges Code-Beispiel
import requests
import json
import time
from datetime import datetime
import pandas as pd
from scipy.interpolate import griddata
from scipy.stats import norm
import numpy as np
============================================
KONFIGURATION
============================================
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen mit echtem Key
Tardis Deribit Endpoints via HolySheep Proxy
TARDIS_DERIBIT_OPTIONS_URL = f"{HOLYSHEEP_BASE_URL}/tardis/deribit/options/chain"
TARDIS_DERIBIT_BOOKS_URL = f"{HOLYSHEEP_BASE_URL}/tardis/deribit/orderbook"
============================================
HELPER FUNCTIONS
============================================
def fetch_options_chain(instrument_name: str, expiration_date: str) -> dict:
"""
Ruft Optionskette für gegebenen Basiswert und Verfallstag ab.
Args:
instrument_name: z.B. "BTC", "ETH"
expiration_date: ISO Format "2026-06-27"
Returns:
Dictionary mit Optionen-Daten
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"instrument": instrument_name,
"expiration": expiration_date,
"include_greeks": True,
"include_iv": True
}
response = requests.post(
TARDIS_DERIBIT_OPTIONS_URL,
headers=headers,
json=payload,
timeout=10
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limit erreicht. Warte {retry_after}s...")
time.sleep(retry_after)
return fetch_options_chain(instrument_name, expiration_date)
response.raise_for_status()
return response.json()
def calculate_implied_volatility(
option_price: float,
S: float, # Spot price
K: float, # Strike price
T: float, # Time to expiration (years)
r: float, # Risk-free rate
is_call: bool = True
) -> float:
"""
Berechnet implizite Volatilität mit Newton-Raphson Iteration.
Black-Scholes Formel für Optionspreisbewertung.
"""
MAX_ITERATIONS = 100
TOLERANCE = 1e-6
sigma = 0.5 # Initial guess
for _ in range(MAX_ITERATIONS):
d1 = (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if is_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)
vega = S * norm.pdf(d1) * np.sqrt(T)
if abs(vega) < 1e-10:
break
diff = option_price - price
if abs(diff) < TOLERANCE:
return sigma
sigma += diff / vega
if sigma <= 0 or sigma > 5:
return None # IV ungültig
return sigma
============================================
HAUPTFUNKTION: IV-SURFACE BERECHNUNG
============================================
def build_iv_surface(instrument: str = "BTC") -> pd.DataFrame:
"""
Baut vollständige IV-Oberfläche für einen Basiswert.
"""
# Verfügbare Verfallstage abrufen
expirations = ["2026-05-29", "2026-06-26", "2026-09-25", "2026-12-25"]
all_iv_data = []
for exp_date in expirations:
print(f"Verarbeite Verfallstag: {exp_date}")
chain_data = fetch_options_chain(instrument, exp_date)
for option in chain_data.get("options", []):
# IV aus API oder Berechnung
iv = option.get("iv") or calculate_implied_volatility(
option_price=option["mark_price"],
S=option["underlying_price"],
K=option["strike"],
T=option["days_to_expiry"] / 365,
r=0.04 # Approximativer risikofreier Zins
)
if iv and 0.1 < iv < 3.0: # Plausibilitätsprüfung
all_iv_data.append({
"timestamp": datetime.utcnow(),
"instrument": instrument,
"expiration": exp_date,
"strike": option["strike"],
"moneyness": option["strike"] / option["underlying_price"],
"option_type": option["type"],
"bid_iv": option.get("bid_iv"),
"ask_iv": option.get("ask_iv"),
"mark_iv": iv,
"open_interest": option.get("open_interest", 0),
"volume_24h": option.get("volume_24h", 0)
})
# API-Cooldown zwischen Requests
time.sleep(0.5)
return pd.DataFrame(all_iv_data)
============================================
AUSFÜHRUNG
============================================
if __name__ == "__main__":
print("Starte IV-Surface Archivierung...")
# BTC IV-Oberfläche berechnen
btc_iv_surface = build_iv_surface("BTC")
# Speichern für spätere Analyse
btc_iv_surface.to_csv(
f"btc_iv_surface_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
index=False
)
print(f"\nIV-Surface archiviert: {len(btc_iv_surface)} Datenpunkte")
print(f"Strike-Range: {btc_iv_surface['strike'].min()} - {btc_iv_surface['strike'].max()}")
print(f"IV-Range: {btc_iv_surface['mark_iv'].min():.2%} - {btc_iv_surface['mark_iv'].max():.2%}")Streaming-Architektur für Echtzeit-Updates
# ============================================
STREAMING INTEGRATION MIT WEBHOOKS
============================================
import asyncio
import aiohttp
from aiohttp import web
import asyncpg
import json
class IVSurfaceStreamer:
"""
Echtzeit-Streaming von IV-Daten mit automatischer Archivierung.
"""
def __init__(self, api_key: str, db_pool: asyncpg.Pool):
self.api_key = api_key
self.db_pool = db_pool
self.base_url = "https://api.holysheep.ai/v1"
async def stream_options_updates(self, instruments: list):
"""
Kontinuierliches Streaming von Options-Updates via HolySheep Webhooks.
"""
webhook_url = "https://your-server.com/webhooks/iv-surface"
# Webhook bei HolySheep registrieren
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/webhooks",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"url": webhook_url,
"events": ["deribit.options.update", "deribit.orderbook.update"],
"filters": {
"instruments": instruments,
"min_volume": 10000 # USDT Mindestvolumen
}
}
) as resp:
if resp.status != 201:
error = await resp.text()
raise ValueError(f"Webhook-Registrierung fehlgeschlagen: {error}")
webhook = await resp.json()
print(f"Webhook registriert: {webhook['id']}")
return webhook['id']
async def process_webhook(self, request: web.Request) -> web.Response:
"""
Verarbeitet eingehende Webhook-Events.
"""
payload = await request.json()
# Event-Typ prüfen
event_type = payload.get("event_type")
if event_type == "deribit.options.update":
await self._store_options_update(payload["data"])
elif event_type == "deribit.orderbook.update":
await self._store_orderbook(payload["data"])
return web.Response(status=200)
async def _store_options_update(self, data: dict):
"""
Speichert Options-Update in PostgreSQL.
"""
async with self.db_pool.acquire() as conn:
await conn.execute("""
INSERT INTO iv_surface_archive (
instrument, timestamp, strike, expiration,
option_type, mark_iv, bid_iv, ask_iv,
open_interest, delta, gamma, theta, vega
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
""",
data["instrument"],
data["timestamp"],
data["strike"],
data["expiration"],
data["option_type"],
data["mark_iv"],
data.get("bid_iv"),
data.get("ask_iv"),
data.get("open_interest", 0),
data.get("greeks", {}).get("delta"),
data.get("greeks", {}).get("gamma"),
data.get("greeks", {}).get("theta"),
data.get("greeks", {}).get("vega")
)
async def _store_orderbook(self, data: dict):
"""
Speichert Orderbook-Daten für Spread-Analyse.
"""
async with self.db_pool.acquire() as conn:
await conn.execute("""
INSERT INTO orderbook_archive (
instrument, timestamp, strike, expiration,
bid_price, ask_price, bid_size, ask_size,
mid_price, spread_bps
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
""",
data["instrument"],
data["timestamp"],
data["strike"],
data["expiration"],
data["bid"],
data["ask"],
data.get("bid_size", 0),
data.get("ask_size", 0),
(data["bid"] + data["ask"]) / 2,
((data["ask"] - data["bid"]) / ((data["bid"] + data["ask"]) / 2)) * 10000
)
async def historical_backfill(self, start_date: str, end_date: str):
"""
Historische Daten via HolySheep archivieren.
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/tardis/deribit/backfill",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"start_date": start_date,
"end_date": end_date,
"instruments": ["BTC", "ETH"],
"data_types": ["options_chain", "orderbook", "trades"]
}
) as resp:
if resp.status == 202:
job = await resp.json()
print(f"Backfill-Job gestartet: {job['job_id']}")
return job['job_id']
else:
raise Exception(f"Backfill fehlgeschlagen: {await resp.text()}")Datenbank-Schema für IV-Surface-Archivierung
-- ============================================
-- POSTGRESQL SCHEMA FÜR IV-SURFACE
-- ============================================
-- Haupttabelle für IV-Daten
CREATE TABLE iv_surface_archive (
id BIGSERIAL PRIMARY KEY,
instrument VARCHAR(10) NOT NULL,
timestamp TIMESTAMPTZ NOT NULL,
strike DECIMAL(20, 4) NOT NULL,
expiration TIMESTAMPTZ NOT NULL,
option_type VARCHAR(5) NOT NULL CHECK (option_type IN ('call', 'put')),
-- Volatilitätsdaten
mark_iv DECIMAL(10, 6),
bid_iv DECIMAL(10, 6),
ask_iv DECIMAL(10, 6),
model_iv DECIMAL(10, 6),
-- Greeks
delta DECIMAL(10, 6),
gamma DECIMAL(10, 6),
theta DECIMAL(10, 6),
vega DECIMAL(10, 6),
rho DECIMAL(10, 6),
-- Volume und OI
open_interest DECIMAL(20, 2),
volume_24h DECIMAL(20, 2),
-- Preisdaten
mark_price DECIMAL(20, 8),
underlying_price DECIMAL(20, 4),
-- Metadaten
created_at TIMESTAMPTZ DEFAULT NOW(),
source VARCHAR(50) DEFAULT 'tardis_deribit'
);
-- Partitionierung nach Datum für Performance
CREATE TABLE iv_surface_archive_2026_05 (
CHECK (timestamp >= '2026-05-01' AND timestamp < '2026-06-01')
) INHERITS (iv_surface_archive);
CREATE TABLE iv_surface_archive_2026_06 (
CHECK (timestamp >= '2026-06-01' AND timestamp < '2026-07-01')
) INHERITS (iv_surface_archive);
-- Indexe für schnelle Abfragen
CREATE INDEX idx_iv_archive_instrument_ts
ON iv_surface_archive (instrument, timestamp DESC);
CREATE INDEX idx_iv_archive_strike_exp
ON iv_surface_archive (instrument, strike, expiration);
CREATE INDEX idx_iv_archive_moneyness
ON iv_surface_archive (instrument, timestamp DESC,
(strike / underlying_price));
-- Trigger für automatische Partitionierung
CREATE OR REPLACE FUNCTION
create_iv_partition_if_not_exists(partition_date DATE)
RETURNS VOID AS $$
DECLARE
partition_name TEXT;
start_date DATE;
end_date DATE;
BEGIN
partition_name := 'iv_surface_archive_' ||
TO_CHAR(partition_date, 'YYYY_MM');
start_date := DATE_TRUNC('month', partition_date);
end_date := start_date + INTERVAL '1 month';
IF NOT EXISTS (
SELECT 1 FROM pg_class WHERE relname = partition_name
) THEN
EXECUTE format(
'CREATE TABLE %I PARTITION OF iv_surface_archive
FOR VALUES FROM (%L) TO (%L)',
partition_name, start_date, end_date
);
END IF;
END;
$$ LANGUAGE plpgsql;
-- Orderbook-Archiv für Spread-Analyse
CREATE TABLE orderbook_archive (
id BIGSERIAL PRIMARY KEY,
instrument VARCHAR(10) NOT NULL,
timestamp TIMESTAMPTZ NOT NULL,
strike DECIMAL(20, 4) NOT NULL,
expiration TIMESTAMPTZ NOT NULL,
bid_price DECIMAL(20, 8),
ask_price DECIMAL(20, 8),
bid_size DECIMAL(20, 4),
ask_size DECIMAL(20, 4),
mid_price DECIMAL(20, 8),
spread_bps DECIMAL(10, 2),
spread_usd DECIMAL(20, 8),
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_orderbook_instrument_ts
ON orderbook_archive (instrument, timestamp DESC);Geeignet / Nicht geeignet für
| Geeignet für | Nicht geeignet für |
|---|---|
|
|
Preise und ROI
| Provider | GPT-4.1 ($/MTok) | Claude Sonnet 4.5 ($/MTok) | DeepSeek V3.2 ($/MTok) | Monatliche Fixkosten |
|---|---|---|---|---|
| HolyShehe AI | $8.00 | $15.00 | $0.42 | Ab $0 |
| Original US-Provider | $15.00 | $18.00 | $2.50 | Ab $2.000 |
| Offizieller Anbieter | $30.00 | $25.00 | $1.80 | Ab $5.000 |
ROI-Kalkulation für IV-Surface-Projekt
Bei einem monatlichen Volumen von 45 Millionen Tokens:
- Vorher: $4.200/Monat bei durchschnittlich $0.093/Token
- Nachher: $680/Monat bei HolySheep (DeepSeek V3.2 für Preprocessing)
- Jährliche Ersparnis: $42.240
- Amortisationszeit: 0 Tage (keine Einrichtungsgebühren)
- Break-Even: Bereits im ersten Monat profitabel
Warum HolySheep wählen
Nach meiner Praxiserfahrung mit der Integration mehrerer Trading-Systeme bietet HolySheep AI独一无二的 Kombination:
- Asiatische Preisstruktur: ¥1 = $1 bedeutet 85%+ Ersparnis gegenüber westlichen Providern
- Native WeChat/Alipay: Zahlungsabwicklung für chinesische Partner und Teammitglieder ohne internationale Transfers
- Sub-50ms Latenz: Direkte Peering-Verbindungen für Trading-Anwendungen kritisch
- Free Credits: 1.000 Tokens für Evaluierung ohne Kreditkarte
- Transparent Pricing: Keine versteckten Kosten, keine Surprise-Rechnungen
Häufige Fehler und Lösungen
1. Rate-Limit-Überschreitung bei hohem Volumen
# FEHLER: Unbegrenzte API-Aufrufe ohne Backoff
for option in all_options:
response = requests.post(url, json={"strike": option})
process(response.json()) # Führt zu 429 Errors
LÖSUNG: Implementierung mit Exponential Backoff und Batch-Requests
import asyncio
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=1200, period=60) # 1200 calls per minute
def fetch_options_with_backoff(option_id: str, max_retries: int = 3):
"""Holt einzelne Option mit automatischer Wiederholung."""
for attempt in range(max_retries):
try:
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/tardis/deribit/option",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={"option_id": option_id}
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limit, warte {retry_after}s...")
time.sleep(retry_after)
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"Attempt {attempt + 1} fehlgeschlagen, warte {wait_time}s...")
time.sleep(wait_time)
Besser: Batch-Requests statt individueller Aufrufe
def fetch_options_batch(option_ids: list, batch_size: int = 50):
"""Holt mehrere Optionen in einem API-Call."""
results = []
for i in range(0, len(option_ids), batch_size):
batch = option_ids[i:i + batch_size]
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/tardis/deribit/options/batch",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={"option_ids": batch}
)
if response.status_code == 200:
results.extend(response.json()["options"])
elif response.status_code == 429:
time.sleep(int(response.headers.get("Retry-After", 60)))
# Erneut versuchen
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/tardis/deribit/options/batch",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={"option_ids": batch}
)
results.extend(response.json()["options"])
# Cooldown zwischen Batches
time.sleep(1)
return results2. Invalid API Key oder Authentication-Fehler
# FEHLER: Hardcodierte Keys oder fehlende Validierung
API_KEY = "sk-live-123456789" # Nie im Code!
LÖSUNG: Environment Variables und automatische Key-Rotation
import os
from dotenv import load_dotenv
load_dotenv() # Lädt .env Datei
def get_api_key() -> str:
"""Holt API-Key sicher aus Environment."""
key = os.environ.get("HOLYSHEEP_API_KEY")
if not key:
raise ValueError(
"HOLYSHEEP_API_KEY nicht gesetzt. "
"Bitte in .env Datei oder Environment Variable definieren."
)
return key
def validate_key_format(key: str) -> bool:
"""Validiert Key-Format vor Verwendung."""
if not key:
return False
if len(key) < 20:
return False
if not key.startswith(("hs_live_", "hs_test_")):
return False
return True
def rotate_api_key(master_key: str) -> dict:
"""
Rotiert API-Key automatisch mit Safety-Mechanismen.
"""
headers = {
"Authorization": f"Bearer {master_key}",
"Content-Type": "application/json"
}
# Alte Keys auflisten
response = requests.get(
f"{HOLYSHEEP_BASE_URL}/keys",
headers=headers
)
if response.status_code != 200:
raise PermissionError(f"Key-Auflistung fehlgeschlagen: {response.text}")
existing_keys = response.json()["keys"]
# Neuen Key generieren
create_response = requests.post(
f"{HOLYSHEEP_BASE_URL}/keys",
headers=headers,
json={
"name": f"auto-rotate-{datetime.now().strftime('%Y%m%d')}",
"rate_limit": 5000
}
)
if create_response.status_code != 201:
raise RuntimeError(f"Key-Erstellung fehlgeschlagen: {create_response.text}")
new_key_data = create_response.json()
# Alte Keys mit Verzögerung deaktivieren (Safety)
for old_key in existing_keys:
if old_key["name"] != "master":
requests.post(
f"{HOLYSHEEP_BASE_URL}/keys/{old_key['id']}/deactivate",
headers=headers,
json={"delay_hours": 24} # 24h Grace Period
)
return {
"new_key": new_key_data["key"],
"old_keys_deactivated_in": "24 hours",
"key_id": new_key_data["id"]
}3. Datenqualitätsprobleme bei IV-Berechnung
# FEHLER: Ungeprüfte Übernahme von IV-Werten ohne Plausibilisierung
iv = option["iv"] # Manchmal negativ oder unrealistisch hoch!
LÖSUNG: Multi-Layer Validierung und Fallbacks
import numpy as np
from scipy.interpolate import interp1d
class IVValidator:
"""Validiert und bereinigt IV-Daten für Trading-Systeme."""
MIN_IV = 0.05 # 5% Minimum (zu niedrig = Datenfehler)
MAX_IV = 3.00 # 300% Maximum (historischer Extremfall)
MAX_JUMP = 0.10 # 10 Prozentpunkte max. Sprung zwischen Strikes
def __init__(self, iv_surface: pd.DataFrame):
self.surface = iv_surface
self.validation_errors = []
def validate_strike_iv(self, strike: float, iv: float) -> tuple:
"""
Validiert einzelnen IV-Wert mit mehrstufiger Prüfung.
Returns:
(is_valid, corrected_iv, error_message)
"""
if iv is None or np.isnan(iv):
return False, None, "IV ist None oder NaN"
if iv < self.MIN_IV:
return False, None, f"IV {iv:.2%} unter Minimum {self.MIN_IV:.2%}"
if iv > self.MAX_IV:
return False, None, f"IV {iv:.2%} über Maximum {self.MAX_IV:.2%}"
return True, iv, None
def detect_smile_anomalies(self, moneyness_range: tuple) -> list:
"""
Erkennt unplausible Volatility Smiles.
"""
mask = (
(self.surface["moneyness"] >= moneyness_range[0]) &
(self.surface["moneyness"] <= moneyness_range[1])
)
slice_data = self.surface[mask].sort_values("moneyness")
anomalies = []
prev_iv = None
for idx, row in slice_data.iterrows():
if prev_iv is not None:
iv_change = abs(row["mark_iv"] - prev_iv)
if iv_change > self.MAX_JUMP:
anomalies.append({
"strike": row["strike"],
"iv": row["mark_iv"],
"prev_iv": prev_iv,
"jump": iv_change,
"severity": "high" if iv_change > 0.2 else "medium"
})
prev_iv = row["mark