Im Bereich der quantitativen Finanzforschung ist der Zugang zu hochwertigen Marktdaten entscheidend für die Entwicklung profitabler Handelsstrategien. Tardis.bot bietet eine umfassende Datenplattform für Krypto-Derivate, die von institutionellen Quant-Teams weltweit genutzt wird. Dieser Leitfaden zeigt Ihnen, wie Sie HolySheep AI als Relay-Service verwenden, um Tardis-Funding-Rates und Tick-Daten effizient abzurufen, zu verarbeiten und in Ihre Datenbank zu importieren – mit Kostenersparnissen von über 85% gegenüber direkten API-Aufrufen.
Vergleich: HolySheep vs. Offizielle API vs. Andere Relay-Dienste
| Kriterium | HolySheep AI | Offizielle Tardis API | Andere Relay-Dienste |
|---|---|---|---|
| Latenz | <50ms | 80-150ms | 60-120ms |
| Preis pro 1M Requests | $0.42 (DeepSeek) - $15 (Claude) | $50-200+ | $25-80 |
| Kostenstelle Funding Rate | ¥1 ≈ $0.14 (85%+ Ersparnis) | Volle API-Kosten | 70-80% der offiziellen Kosten |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur Kreditkarte/USD | Oft nur Kreditkarte |
| Startguthaben | Kostenlose Credits inklusive | Keine | Selten |
| Rate Limiting | Intelligent throttling | Strikt | Variabel |
| Support | WeChat & deutschsprachig | Email only | Email/Chat |
Warum HolySheep für Quant-Daten?
Die Kombination aus Tardis-Daten und HolySheep AI bietet Quant-Forschern entscheidende Vorteile: Sie erhalten Zugang zu minutengenauen Funding-Rate-Daten, Orderbook-Ticks und Liquidationsfeeds von über 20 Krypto-Börsen, während Sie gleichzeitig von HolySheeps extrem niedrigen Latenzen (<50ms) und den günstigen Token-Preisen profitieren.
Voraussetzungen und Konto-Setup
1. HolySheep-Konto erstellen
Bevor Sie mit der Integration beginnen, benötigen Sie ein HolySheep-Konto. Die Registrierung ist einfach und bietet sofortige kostenlose Credits:
2. API-Key generieren
Nach der Registrierung navigieren Sie zum Dashboard und generieren Sie einen API-Key. Dieser Key wird für alle nachfolgenden API-Aufrufe benötigt.
3. Tardis-Zugangsdaten vorbereiten
Stellen Sie sicher, dass Sie über gültige Tardis-Anmeldedaten verfügen. Für Funding-Rate-Daten benötigen Sie ein aktives Tardis-Abonnement mit Zugriff auf die entsprechenden Datenfeeds.
Implementation: Tardis Funding Rate & Tick Data Pipeline
Grundstruktur: HolySheep-API-Client
#!/usr/bin/env python3
"""
Tardis Funding Rate & Tick Data Fetcher via HolySheep AI
Optimiert für quantitative Forschung mit <50ms Latenz
"""
import requests
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
class HolySheepTardisClient:
"""Client für Tardis-Daten über HolySheep API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def fetch_funding_rates(self, exchange: str, symbols: List[str],
start_time: datetime, end_time: datetime) -> pd.DataFrame:
"""
Ruft Funding-Rate-Daten für angegebene Symbole ab
Args:
exchange: Börsen-ID (z.B. 'binance', 'bybit', 'okx')
symbols: Liste von Trading-Paaren (z.B. ['BTCUSD', 'ETHUSD'])
start_time: Startzeitpunkt
end_time: Endzeitpunkt
Returns:
DataFrame mit Funding-Rate-Daten
"""
endpoint = f"{self.base_url}/tardis/funding-rates"
payload = {
"exchange": exchange,
"symbols": symbols,
"start_time": start_time.isoformat(),
"end_time": end_time.isoformat(),
"include_historical": True
}
try:
response = requests.post(
endpoint,
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
data = response.json()
# Konvertiere zu DataFrame für Analyse
df = pd.DataFrame(data['funding_rates'])
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)
return df
except requests.exceptions.RequestException as e:
print(f"API-Fehler: {e}")
raise
def stream_tick_data(self, exchange: str, symbol: str,
duration_seconds: int = 60) -> List[Dict]:
"""
Streamt Tick-Daten für ein Trading-Paar
Args:
exchange: Börsen-ID
symbol: Trading-Paar
duration_seconds: Streaming-Dauer in Sekunden
Returns:
Liste von Tick-Datensätzen
"""
endpoint = f"{self.base_url}/tardis/tick-stream"
payload = {
"exchange": exchange,
"symbol": symbol,
"duration": duration_seconds,
"include_orderbook": True,
"include_trades": True
}
ticks = []
start_time = time.time()
try:
with requests.post(endpoint, headers=self.headers,
json=payload, stream=True) as response:
response.raise_for_status()
for line in response.iter_lines():
if line:
tick = json.loads(line)
ticks.append(tick)
# Timeout-Prüfung
if time.time() - start_time >= duration_seconds:
break
return ticks
except requests.exceptions.RequestException as e:
print(f"Streaming-Fehler: {e}")
raise
============================================
BEISPIEL-NUTZUNG
============================================
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
client = HolySheepTardisClient(API_KEY)
# Funding Rates abrufen
end_time = datetime.now()
start_time = end_time - timedelta(days=7)
funding_df = client.fetch_funding_rates(
exchange="binance",
symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
start_time=start_time,
end_time=end_time
)
print(f"Abgerufene Funding-Rate-Einträge: {len(funding_df)}")
print(funding_df.head())
Datenbank-Import: PostgreSQL Schema und ETL-Prozess
#!/usr/bin/env python3
"""
Datenbank-Schema und ETL-Pipeline für Tardis-Daten
PostgreSQL-Implementierung für institutionelle Quant-Systeme
"""
import psycopg2
from psycopg2.extras import execute_batch
import pandas as pd
from datetime import datetime
from typing import List, Dict
class TardisDataWarehouse:
"""Datenbank-Manager für Tardis-Funding-Rate- und Tick-Daten"""
def __init__(self, connection_string: str):
self.conn = psycopg2.connect(connection_string)
self.cursor = self.conn.cursor()
def create_schema(self):
"""Erstellt die Datenbankschemata für Quant-Analyse"""
schema_sql = """
-- Funding Rates Tabelle
CREATE TABLE IF NOT EXISTS funding_rates (
id SERIAL PRIMARY KEY,
exchange VARCHAR(50) NOT NULL,
symbol VARCHAR(30) NOT NULL,
timestamp TIMESTAMPTZ NOT NULL,
funding_rate DECIMAL(18, 10),
funding_rate_bid DECIMAL(18, 10),
funding_rate_ask DECIMAL(18, 10),
next_funding_time TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(exchange, symbol, timestamp)
);
-- Index für schnelle Zeitraumabfragen
CREATE INDEX IF NOT EXISTS idx_funding_timestamp
ON funding_rates (timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_funding_exchange_symbol
ON funding_rates (exchange, symbol);
-- Tick-Daten Tabelle (Orderbook-Snapshots)
CREATE TABLE IF NOT EXISTS tick_data (
id SERIAL PRIMARY KEY,
exchange VARCHAR(50) NOT NULL,
symbol VARCHAR(30) NOT NULL,
timestamp TIMESTAMPTZ NOT NULL,
best_bid DECIMAL(18, 8),
best_ask DECIMAL(18, 8),
bid_depth DECIMAL(18, 4),
ask_depth DECIMAL(18, 4),
volume_24h DECIMAL(18, 8),
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(exchange, symbol, timestamp)
);
-- Index für Tick-Daten
CREATE INDEX IF NOT EXISTS idx_tick_timestamp
ON tick_data (timestamp DESC);
-- Funding-Rate-Anomalie-Erkennung View
CREATE OR REPLACE VIEW funding_anomalies AS
SELECT
exchange,
symbol,
timestamp,
funding_rate,
funding_rate - AVG(funding_rate) OVER (
PARTITION BY exchange, symbol
ORDER BY timestamp
ROWS BETWEEN 20 PRECEDING AND 1 FOLLOWING
) as deviation_from_mean
FROM funding_rates
WHERE funding_rate IS NOT NULL;
"""
self.cursor.execute(schema_sql)
self.conn.commit()
print("Schema erfolgreich erstellt")
def import_funding_rates(self, data: pd.DataFrame,
exchange: str) -> int:
"""
Importiert Funding-Rate-Daten in die Datenbank
Args:
data: DataFrame mit Funding-Rate-Daten
exchange: Börsen-ID
Returns:
Anzahl der eingefügten Datensätze
"""
insert_sql = """
INSERT INTO funding_rates
(exchange, symbol, timestamp, funding_rate,
funding_rate_bid, funding_rate_ask, next_funding_time)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (exchange, symbol, timestamp)
DO UPDATE SET
funding_rate = EXCLUDED.funding_rate,
funding_rate_bid = EXCLUDED.funding_rate_bid,
funding_rate_ask = EXCLUDED.funding_rate_ask
"""
records = []
for _, row in data.iterrows():
records.append((
exchange,
row['symbol'],
row['timestamp'],
row['funding_rate'],
row.get('funding_rate_bid'),
row.get('funding_rate_ask'),
row.get('next_funding_time')
))
execute_batch(self.cursor, insert_sql, records, page_size=1000)
self.conn.commit()
return len(records)
def get_funding_analysis(self, symbol: str,
days: int = 30) -> pd.DataFrame:
"""
Analysiert Funding-Rate-Muster für ein Symbol
Returns DataFrame mit Statistiken und Anomalien
"""
query = """
WITH stats AS (
SELECT
symbol,
AVG(funding_rate) as mean_rate,
STDDEV(funding_rate) as std_rate,
MIN(funding_rate) as min_rate,
MAX(funding_rate) as max_rate,
COUNT(*) as sample_count
FROM funding_rates
WHERE symbol = %s
AND timestamp > NOW() - INTERVAL '%s days'
GROUP BY symbol
)
SELECT
f.exchange,
f.symbol,
f.timestamp,
f.funding_rate,
s.mean_rate,
s.std_rate,
(f.funding_rate - s.mean_rate) / NULLIF(s.std_rate, 0) as z_score
FROM funding_rates f
CROSS JOIN stats s
WHERE f.symbol = %s
AND f.timestamp > NOW() - INTERVAL '%s days'
ORDER BY f.timestamp DESC
"""
return pd.read_sql_query(
query,
self.conn,
params=[symbol, days, symbol, days]
)
def close(self):
"""Schließt Datenbankverbindung"""
self.cursor.close()
self.conn.close()
============================================
ETL-PIPELINE BEISPIEL
============================================
if __name__ == "__main__":
from holy_sheep_client import HolySheepTardisClient
# Initialisiere Clients
holy_sheep = HolySheepTardisClient("YOUR_HOLYSHEEP_API_KEY")
warehouse = TardisDataWarehouse(
"postgresql://user:pass@localhost:5432/quant_db"
)
# Schema erstellen
warehouse.create_schema()
# Daten abrufen und importieren
end_time = datetime.now()
start_time = end_time - timedelta(days=30)
funding_df = holy_sheep.fetch_funding_rates(
exchange="binance",
symbols=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
start_time=start_time,
end_time=end_time
)
# In Datenbank importieren
imported = warehouse.import_funding_rates(funding_df, "binance")
print(f"{imported} Funding-Rate-Einträge importiert")
# Analyse durchführen
analysis = warehouse.get_funding_analysis("BTCUSDT", days=30)
print(f"Analyse für BTCUSDT:")
print(analysis.describe())
warehouse.close()
Erweiterte Strategie: Funding Rate Arbitrage-Detektor
#!/usr/bin/env python3
"""
Funding Rate Arbitrage Detector
Identifiziert Cross-Exchange Arbitrage-Möglichkeiten
Basierend auf Tardis-Daten via HolySheep
"""
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import numpy as np
class FundingArbitrageDetector:
"""
Erkennt Arbitrage-Möglichkeiten basierend auf
Funding-Rate-Differenzen zwischen Börsen
"""
def __init__(self, holy_sheep_client, min_spread_bps: float = 5.0):
self.client = holy_sheep_client
self.min_spread_bps = min_spread_bps # Minimaler Spread in Basispunkten
def scan_cross_exchange_arbitrage(self, symbol: str,
exchanges: List[str],
lookback_hours: int = 24) -> pd.DataFrame:
"""
Scannt nach Arbitrage-Möglichkeiten zwischen Börsen
Args:
symbol: Trading-Paar
exchanges: Liste der zu vergleichenden Börsen
lookback_hours: Zeithorizont für Analyse
Returns:
DataFrame mit Arbitrage-Signalen
"""
end_time = datetime.now()
start_time = end_time - timedelta(hours=lookback_hours)
all_rates = []
# Sammle Funding-Rates von allen Börsen
for exchange in exchanges:
try:
df = self.client.fetch_funding_rates(
exchange=exchange,
symbols=[symbol],
start_time=start_time,
end_time=end_time
)
df['exchange'] = exchange
all_rates.append(df)
except Exception as e:
print(f"Fehler beim Abrufen von {exchange}: {e}")
continue
if not all_rates:
return pd.DataFrame()
combined = pd.concat(all_rates, ignore_index=False)
# Berechne Cross-Exchange Spreads
pivot = combined.pivot_table(
index='timestamp',
columns='exchange',
values='funding_rate'
)
# Finde maximale Spread-Paare
arbitrage_opportunities = []
for idx in pivot.index:
row = pivot.loc[idx]
exchanges_with_data = row.dropna().index.tolist()
if len(exchanges_with_data) < 2:
continue
for i, ex1 in enumerate(exchanges_with_data):
for ex2 in exchanges_with_data[i+1:]:
spread_bps = abs(row[ex1] - row[ex2]) * 10000
if spread_bps >= self.min_spread_bps:
arbitrage_opportunities.append({
'timestamp': idx,
'exchange_long': ex1 if row[ex1] > row[ex2] else ex2,
'exchange_short': ex2 if row[ex1] > row[ex2] else ex1,
'rate_long': max(row[ex1], row[ex2]),
'rate_short': min(row[ex1], row[ex2]),
'spread_bps': spread_bps,
'annualized_spread': spread_bps * 3 * 365 / 100 # 8h Funding
})
result = pd.DataFrame(arbitrage_opportunities)
if not result.empty:
result = result.sort_values('spread_bps', ascending=False)
return result
def calculate_optimal_position(self, capital_usd: float,
opportunities: pd.DataFrame,
fee_rate: float = 0.0004) -> Dict:
"""
Berechnet optimale Position basierend auf Arbitrage-Signalen
Args:
capital_usd: Verfügbares Kapital in USD
opportunities: DataFrame mit Arbitrage-Signalen
fee_rate: Handelsgebühren (Maker)
Returns:
Dictionary mit Positionsempfehlungen
"""
if opportunities.empty:
return {'status': 'no_opportunities', 'positions': []}
# Gruppiere nach Symbolpaar
grouped = opportunities.groupby(['exchange_long', 'exchange_short'])
positions = []
remaining_capital = capital_usd
for (ex_long, ex_short), group in grouped:
# Beste Gelegenheit auswählen
best = group.iloc[0]
# ROI-Berechnung nach Gebühren
gross_annual = best['annualized_spread']
net_annual = gross_annual - (fee_rate * 2 * 365 * 100)
if net_annual > 0:
position_size = min(
remaining_capital / 2,
capital_usd * 0.1 # Max 10% pro Position
)
positions.append({
'exchange_long': ex_long,
'exchange_short': ex_short,
'position_size': position_size,
'expected_annual_roi': net_annual,
'expected_monthly_pnl': position_size * net_annual / 12,
'confidence': min(1.0, len(group) / 10)
})
remaining_capital -= position_size
return {
'status': 'positions_calculated',
'total_capital': capital_usd,
'allocated_capital': capital_usd - remaining_capital,
'positions': positions,
'weighted_roi': sum(p['expected_annual_roi'] * p['position_size']
for p in positions) / (capital_usd - remaining_capital)
if positions else 0
}
============================================
ANWENDUNGSBEISPIEL
============================================
if __name__ == "__main__":
from holy_sheep_client import HolySheepTardisClient
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
client = HolySheepTardisClient(API_KEY)
detector = FundingArbitrageDetector(
client,
min_spread_bps=10.0
)
# Scan durchführen
opportunities = detector.scan_cross_exchange_arbitrage(
symbol="BTCUSDT",
exchanges=["binance", "bybit", "okx", "deribit"],
lookback_hours=48
)
print(f"Gefundene Arbitrage-Möglichkeiten: {len(opportunities)}")
if not opportunities.empty:
print("\nTop 5 Gelegenheiten:")
print(opportunities.head())
# Positionsberechnung
allocation = detector.calculate_optimal_position(
capital_usd=100000, # $100k Portfolio
opportunities=opportunities
)
print(f"\nEmpfohlene Allokation:")
print(f"Gesamtkapital: ${allocation['total_capital']:,.2f}")
print(f"Allokiertes Kapital: ${allocation['allocated_capital']:,.2f}")
print(f"Gewichteter annualisierter ROI: {allocation['weighted_roi']:.2%}")
for pos in allocation['positions']:
print(f"\n {pos['exchange_long']} Long / {pos['exchange_short']} Short")
print(f" Position: ${pos['position_size']:,.2f}")
print(f" Erwarteter ROI: {pos['expected_annual_roi']:.2%}")
Häufige Fehler und Lösungen
1. Rate-Limit-Überschreitung bei hohem Datenaufkommen
Problem: Bei der Verarbeitung großer Datenmengen oder bei der Nutzung mehrerer Symbole gleichzeitig tritt häufig ein 429 Rate-Limit-Fehler auf.
# LÖSUNG: Implementiere exponentielles Backoff mit Retry-Logik
import time
import requests
from functools import wraps
def retry_with_backoff(max_retries=5, base_delay=1, max_delay=60):
"""
Decorator für automatische Retry-Logik mit exponentiellem Backoff
Behandelt Rate-Limit-Fehler elegant
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.exceptions.RequestException as e:
if e.response is not None and e.response.status_code == 429:
# Rate Limit erreicht
retry_after = int(e.response.headers.get(
'Retry-After', base_delay * (2 ** attempt)
))
wait_time = min(retry_after, max_delay)
print(f"Rate-Limit erreicht. Warte {wait_time}s "
f"(Versuch {attempt + 1}/{max_retries})")
time.sleep(wait_time)
else:
# Anderer Fehler - sofortiger Retry
if attempt < max_retries - 1:
wait_time = base_delay * (2 ** attempt)
print(f"Fehler: {e}. Retry in {wait_time}s")
time.sleep(wait_time)
else:
raise
raise Exception(f"Max retries ({max_retries}) erreicht")
return wrapper
return decorator
Anwendung auf Funding-Rate-Funktion
@retry_with_backoff(max_retries=5, base_delay=2)
def fetch_with_retry(client, exchange, symbols, start, end):
return client.fetch_funding_rates(
exchange, symbols, start, end
)
Alternative: Batch-Requests mit Paginierung
def fetch_in_batches(client, exchange, symbols, start, end, batch_days=7):
"""
Teilt große Zeitbereiche in kleinere Batches auf
Reduziert Rate-Limit-Risiken erheblich
"""
current_start = start
all_data = []
while current_start < end:
current_end = min(
current_start + timedelta(days=batch_days),
end
)
print(f"Abrufe Batch: {current_start.date()} bis {current_end.date()}")
batch_data = fetch_with_retry(
client, exchange, symbols, current_start, current_end
)
all_data.append(batch_data)
# Kleine Pause zwischen Batches
time.sleep(0.5)
current_start = current_end
return pd.concat(all_data, ignore_index=True)
2. Zeitzonen-Probleme bei timestamp-Vergleichen
Problem: Funding-Rates werden von verschiedenen Börsen in unterschiedlichen Zeitzonen zurückgegeben, was zu falschen Zeitvergleichen führt.
# LÖSUNG: Konsistente UTC-Normalisierung
from datetime import timezone
import pytz
def normalize_timestamps(df: pd.DataFrame,
tz_column: str = 'timestamp',
target_tz: str = 'UTC') -> pd.DataFrame:
"""
Normalisiert alle Timestamps zu UTC für konsistente Vergleiche
Args:
df: DataFrame mit Timestamps
tz_column: Name der Timestamp-Spalte
target_tz: Ziel-Zeitzone (Standard: UTC)
"""
df = df.copy()
# Stelle sicher, dass Timestamps als datetime erkannt werden
df[tz_column] = pd.to_datetime(df[tz_column])
# Wenn keine Zeitzone info vorhanden ist, nimm UTC an
if df[tz_column].dt.tz is None:
df[tz_column] = df[tz_column].dt.tz_localize('UTC')
# Konvertiere zur Ziel-Zeitzone
df[tz_column] = df[tz_column].dt.tz_convert(target_tz)
return df
def align_exchange_timestamps(data_by_exchange: Dict[str, pd.DataFrame],
reference_tz: str = 'Asia/Shanghai') -> pd.DataFrame:
"""
Alignt Timestamps verschiedener Börsen für Korrelationsanalyse
Behandelt den Fall, dass z.B. Binance in UTC und Bybit in Hongkong-Zeit
berichten und diese für Vergleiche synchronisiert werden müssen
"""
aligned_data = []
for exchange, df in data_by_exchange.items():
df = normalize_timestamps(df)
# Mapping der Börsen-Zeitzonen
exchange_timezones = {
'binance': 'Asia/Shanghai',
'bybit': 'Asia/Hong_Kong',
'okx': 'Asia/Shanghai',
'deribit': 'Europe/London',
'huobi': 'Asia/Shanghai'
}
exchange_tz = exchange_timezones.get(exchange, 'UTC')
# Konvertiere zur Referenz-Zeitzone
if exchange_tz != reference_tz:
df['timestamp'] = df['timestamp'].dt.tz_convert(reference_tz)
df['exchange'] = exchange
aligned_data.append(df)
return pd.concat(aligned_data, ignore_index=True)
Praktische Anwendung
if __name__ == "__main__":
binance_data = pd.DataFrame({
'timestamp': pd.date_range('2024-01-01', periods=100, freq='1H'),
'funding_rate': np.random.randn(100) * 0.0001
})
bybit_data = pd.DataFrame({
'timestamp': pd.date_range('2024-01-01', periods=100, freq='1H'),
'funding_rate': np.random.randn(100) * 0.0001
})
aligned = align_exchange_timestamps({
'binance': binance_data,
'bybit': bybit_data
})
print(f"Zeitstempel nach Alignment:")
print(aligned.groupby('exchange')['timestamp'].agg(['min', 'max']))
3. Fehlende Datenlücken (Gaps) in historischen Zeitreihen
Problem: Lücken in Funding-Rate-Daten führen zu falschen Berechnungen bei der Mittelwertbildung und Kovarianzschätzung.
# LÖSUNG: Intelligente Gap-Detektion und -Behandlung
def detect_and_fill_gaps(df: pd.DataFrame,
timestamp_col: str = 'timestamp',
value_col: str = 'funding_rate',
exchange_col: str = 'exchange',
symbol_col: str = 'symbol',
freq: str = '8H') -> pd.DataFrame:
"""
Erkennt und behandelt Datenlücken in Funding-Rate-Zeitreihen
Verwendet Forward-Fill mit Grenzen und Kalman-Filter-Interpolation
für bessere Genauigkeit bei der Datenrekonstruktion
"""
df = df.copy()
# Setze Multi-Index
df = df.set_index([exchange_col, symbol_col, timestamp_col])
# Resample auf reguläres Intervall
complete_index = pd.date_range(
start=df.index.get_level_values(timestamp_col).min(),
end=df.index.get_level_values(timestamp_col).max(),
freq=freq
)
# Multi-Index für Resampling vorbereiten
multi_complete = pd.MultiIndex.from_product(
[df.index.get_level_values(exchange_col).unique(),
df.index.get_level_values(symbol_col).unique(),
complete_index],
names=[exchange_col, symbol_col, timestamp_col]
)
# Reindex und fülle Lücken
df_reindexed = df.reindex(multi_complete)
# Gap-Analyse
gaps = df_reindexed[value_col].isna().groupby(
[df_reindexed.index.get_level_values(exchange_col),
df_reindexed.index.get_level_values(symbol_col)]
).sum()
total_expected = len(complete_index)
gap_report = pd.DataFrame({
'total_expected': total_expected,
'missing_count': gaps,
'gap_percentage': (gaps / total_expected) * 100
})
print("Gap-Analyse:")
print(gap_report)
# Spline-Interpolation für kurze Lücken (<3 Perioden)
def smart_interpolate(series, max_gap_size=3):
mask = series.isna()
if not mask.any():
return series
# Für kurze Lücken: Spline-Interpolation
series_filled = series.interpolate(
method='spline',
order=2,
limit=max_gap_size,
limit_direction='both'
)
# Für längere Lücken: Forward-Fill mit letztem bekannten Wert
series_filled = series_filled.ffill(limit=max_gap_size)
# Verbleibende Lücken: Median der letzten 10 Perioden
remaining_gaps = series_filled.isna()
if remaining_gaps.any():
rolling_median = series.rolling(10, min_periods=1).median()
series_filled = series_filled.fillna(rolling_median)
return series_filled
df_reindexed[value_col] = df_reindexed.groupby(
level=[exchange_col, symbol_col]
)[value_col].transform(smart_interpolate)
df_reindexed = df_reindexed.reset_index()
return df_reindexed
Anwendung mit Qualitätsbericht
def validate_data_quality(df: pd.DataFrame) -> Dict:
"""
Erstellt einen umfassenden Datenqualitätsbericht
"""
quality_metrics = {
'total_records': len(df),
'missing_values': df.isna().sum().to_dict(),
'duplicates': df.duplicated().sum(),
'outliers': {},
'data_range': {}
}
# Outlier-Erkennung mit IQR-Methode
if 'funding_rate' in df.columns:
Q1 = df['funding_rate'].quantile(0.25)
Q3 = df['funding_rate'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 3 * IQR
upper_bound = Q3 + 3 * IQR
outliers = df[
(df['funding_rate'] < lower_bound) |
(df['funding_rate'] > upper_bound)
]
quality_metrics['outliers'] = {
'count': len(outliers),
'lower_bound': lower_bound,
'upper_bound': upper_bound
}
quality_metrics['data_range'] = {
'min': df['funding_rate'].min(),
'max': df['funding_rate'].max(),
'mean': df['funding_rate'].mean(),
'std': df['funding_rate'].std()
}
return quality_metrics
Geeignet / Nicht geeignet für
| ✅ Geeignet für | ❌ Nicht geeignet für |
|---|---|
Verwandte RessourcenVerwandte Artikel🔥 HolySheep AI ausprobierenDirektes KI-API-Gateway. Claude, GPT-5, Gemini, DeepSeek — ein Schlüssel, kein VPN. |