Einleitung: Wenn die Datenqualität zum Albtraum wird
Stellen Sie sich folgendes Szenario vor: Sie haben ein automatisiertes Trading-System aufgebaut und vertrauen den historischen Liquidation-Daten eines großen Exchanges. Plötzlich erhalten Sie während einer kritischen Marktbewegung:
ConnectionError: HTTPSConnectionPool(host='api.tardis.dev', port=443):
Max retries exceeded with url: /v1/历史强平数据 (Caused by
ConnectTimeoutError(<urllib3.connection.HTTPSConnection object...))
Oder schlimmer:
401 Unauthorized: Invalid API key for exchange data access
Ihre automatisierten Stop-Loss-Orders brechen ab
Historische Backtests zeigen falsche Liquidation-Preise
In diesem Tutorial zeige ich Ihnen, wie Sie mit Tardis的历史强平数据 eine vollständige Datenqualitätsprüfung durchführen – von der Fehlerbehandlung bis zur automatisierten Anomalieerkennung mithilfe der HolySheep AI API.
Was ist Tardis und warum ist die Datenqualität so wichtig?
Tardis bietet historische Marktdaten für Kryptobörsen, einschließlich Liquidation-Events. Diese Daten sind essentiell für:
- Backtesting von Trading-Strategien
- Risikoanalyse und Margin-Berechnungen
- Forensische Analysen nach Marktcrashs
- Regulatorische Compliance und Audit-Anforderungen
Projekt-Setup und Installation
# Python-Umgebung für Tardis-Datenqualitätsprüfung
pip install tardis-client pandas numpy httpx asyncio aiohttp
pip install holy-sheep-sdk # Für KI-gestützte Anomalieerkennung
Optional: Visualisierung
pip install plotly matplotlib# Konfigurationsdatei: config.py
import os
Tardis API-Konfiguration
TARDIS_CONFIG = {
"api_key": os.getenv("TARDIS_API_KEY"),
"base_url": "https://api.tardis.dev/v1",
"exchange": "binance", # oder okx, bybit, etc.
"symbols": ["BTC-USDT", "ETH-USDT", "SOL-USDT"],
"timeframe": "1m"
}
HolySheep AI API-Integration für Anomalieanalyse
HOLYSHEEP_CONFIG = {
"api_key": os.getenv("HOLYSHEEP_API_KEY"),
"base_url": "https://api.holysheep.ai/v1", # WICHTIG: NIEMALS api.openai.com
"model": "deepseek-v3.2", # $0.42/MTok - kosteneffizient für Batch-Analyse
"temperature": 0.1,
"max_tokens": 500
}
Datenqualitäts-Schwellenwerte
QUALITY_THRESHOLDS = {
"max_price_jump_pct": 5.0, # 5% Preissprung = Anomalie
"min_liquidation_interval_sec": 60, # Mindestintervall zwischen Events
"max_missing_data_gap_min": 30, # Max 30 Minuten Lücken
"suspicious_volume_multiplier": 10 # 10x durchschnittliches Volumen
}Datenerfassung: Tardis历史强平数据 abrufen
# tardis_fetcher.py
import httpx
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
class TardisLiquidationFetcher:
"""Holt historische Liquidation-Daten von Tardis API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.client = httpx.AsyncClient(timeout=30.0)
async def fetch_liquidations(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> List[Dict]:
"""
Ruft Liquidation-Events für einen bestimmten Zeitraum ab.
Beispiel-URL: GET /v1/历史强平数据?exchange=binance&symbol=BTC-USDT
"""
url = f"{self.base_url}/historical-liquidation-events"
params = {
"exchange": exchange,
"symbol": symbol,
"start_time": int(start_date.timestamp() * 1000),
"end_time": int(end_date.timestamp() * 1000),
"format": "json"
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = await self.client.get(url, params=params, headers=headers)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
raise ConnectionError(
"401 Unauthorized: Ungültiger Tardis API-Key. "
"Überprüfen Sie Ihre Anmeldedaten unter https://docs.tardis.io"
) from e
elif e.response.status_code == 429:
raise ConnectionError(
"429 Too Many Requests: Rate-Limit erreicht. "
"Warten Sie 60 Sekunden oder upgraden Sie Ihren Plan."
) from e
raise
except httpx.TimeoutException:
raise ConnectionError(
"Timeout: Tardis API antwortet nicht. "
"Netzwerkverbindung prüfen oder API-Status unter https://status.tardis.io"
) from None
async def fetch_with_retry(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
max_retries: int = 3
) -> List[Dict]:
"""Robuster Fetch mit automatischer Wiederholung"""
for attempt in range(max_retries):
try:
return await self.fetch_liquidations(
exchange, symbol, start_date, end_date
)
except ConnectionError as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt # Exponential backoff
print(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
print(f"Warte {wait_time} Sekunden...")
await asyncio.sleep(wait_time)
return []
Beispiel-Nutzung
async def main():
fetcher = TardisLiquidationFetcher(api_key="IHR_TARDIS_KEY")
start = datetime(2025, 12, 1)
end = datetime(2025, 12, 2)
try:
liquidations = await fetcher.fetch_with_retry(
exchange="binance",
symbol="BTC-USDT",
start_date=start,
end_date=end
)
print(f"✓ {len(liquidations)} Liquidation-Events abgerufen")
except ConnectionError as e:
print(f"✗ Fehler: {e}")
# Fallback: Datei aus Cache laden oder Alternative API nutzen
if __name__ == "__main__":
asyncio.run(main())Datenqualitätsprüfung:清算事件完整性检查
# quality_checker.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict, Tuple
from HOLYSHEEP_CONFIG import QUALITY_THRESHOLDS
@dataclass
class QualityIssue:
"""Struktur für ein erkanntes Qualitätsproblem"""
issue_type: str
timestamp: datetime
severity: str # "low", "medium", "high", "critical"
description: str
details: Dict
class LiquidationQualityChecker:
"""Prüft Liquidation-Daten auf Qualitätsprobleme"""
def __init__(self, df: pd.DataFrame, exchange: str, symbol: str):
self.df = df
self.exchange = exchange
self.symbol = symbol
self.issues: List[QualityIssue] = []
def check_all(self) -> List[QualityIssue]:
"""Führt alle Qualitätsprüfungen durch"""
print(f"\n🔍 Starte Qualitätsprüfung für {self.exchange}/{self.symbol}...")
print(f" Datensätze: {len(self.df)}")
self.check_missing_intervals()
self.check_price_jumps()
self.check_timestamp_gaps()
self.check_suspicious_volumes()
self.check_duplicate_events()
print(f" ✓ {len(self.issues)} Probleme gefunden\n")
return self.issues
def check_missing_intervals(self):
"""Prüft auf fehlende Zeitintervalle (kritisch für Backtests)"""
if "timestamp" not in self.df.columns:
self.issues.append(QualityIssue(
issue_type="MISSING_TIMESTAMP_COLUMN",
timestamp=datetime.now(),
severity="critical",
description="DataFrame hat keine Timestamp-Spalte",
details={"columns": list(self.df.columns)}
))
return
# Sortiere nach Zeitstempel
df_sorted = self.df.sort_values("timestamp")
# Berechne Zeitdifferenzen
time_diffs = df_sorted["timestamp"].diff()
# Finde große Lücken
max_gap = timedelta(minutes=QUALITY_THRESHOLDS["max_missing_data_gap_min"])
large_gaps = time_diffs[time_diffs > max_gap]
for idx, diff in large_gaps.items():
start_time = df_sorted.loc[idx - 1, "timestamp"] if idx > 0 else None
end_time = df_sorted.loc[idx, "timestamp"]
self.issues.append(QualityIssue(
issue_type="MISSING_INTERVAL",
timestamp=start_time,
severity="high",
description=f"Große Datenlücke: {diff} ohne Liquidation-Events",
details={
"start": start_time,
"end": end_time,
"gap_duration_min": diff.total_seconds() / 60
}
))
def check_price_jumps(self):
"""Erkennt unnatürliche Preissprünge bei Liquidation-Preisen"""
if "liquidation_price" not in self.df.columns:
return
df_sorted = self.df.sort_values("timestamp")
prices = df_sorted["liquidation_price"].values
timestamps = df_sorted["timestamp"].values
for i in range(1, len(prices)):
if prices[i-1] == 0:
continue
jump_pct = abs(prices[i] - prices[i-1]) / prices[i-1] * 100
if jump_pct > QUALITY_THRESHOLDS["max_price_jump_pct"]:
self.issues.append(QualityIssue(
issue_type="PRICE_JUMP",
timestamp=pd.Timestamp(timestamps[i]),
severity="critical" if jump_pct > 20 else "high",
description=f"Preissprung von {jump_pct:.2f}% erkannt",
details={
"previous_price": prices[i-1],
"current_price": prices[i],
"jump_percentage": jump_pct,
"time_diff_sec": (
pd.Timestamp(timestamps[i]) -
pd.Timestamp(timestamps[i-1])
).total_seconds()
}
))
def check_timestamp_gaps(self):
"""Prüft auf unregelmäßige Zeitstempel (Sync-Probleme)"""
df_sorted = self.df.sort_values("timestamp")
timestamps = pd.to_datetime(df_sorted["timestamp"])
# Gruppiere nach Minute
minute_counts = timestamps.dt.floor("min").value_counts()
# Mehr als 10 Events pro Minute = verdächtig
suspicious = minute_counts[minute_counts > 10]
for timestamp, count in suspicious.items():
self.issues.append(QualityIssue(
issue_type="HIGH_FREQUENCY_BURST",
timestamp=timestamp,
severity="medium",
description=f"{count} Events in einer Minute (möglicher Burst/DoS)",
details={"event_count": count}
))
def check_suspicious_volumes(self):
"""Findet ungewöhnlich große Liquidationen"""
if "liquidation_volume" not in self.df.columns:
return
volume_mean = self.df["liquidation_volume"].mean()
volume_std = self.df["liquidation_volume"].std()
threshold = volume_mean + (
volume_std * QUALITY_THRESHOLDS["suspicious_volume_multiplier"]
)
suspicious = self.df[self.df["liquidation_volume"] > threshold]
for _, row in suspicious.iterrows():
self.issues.append(QualityIssue(
issue_type="SUSPICIOUS_VOLUME",
timestamp=row["timestamp"],
severity="low",
description=f"Ungewöhnlich große Liquidation: "
f"{row['liquidation_volume']:.2f} (Ø: {volume_mean:.2f})",
details={
"volume": row["liquidation_volume"],
"mean": volume_mean,
"multiplier": row["liquidation_volume"] / volume_mean
}
))
def check_duplicate_events(self):
"""Erkennt doppelte Einträge"""
duplicates = self.df.duplicated(subset=["timestamp", "liquidation_price"])
dup_count = duplicates.sum()
if dup_count > 0:
self.issues.append(QualityIssue(
issue_type="DUPLICATE_EVENTS",
timestamp=self.df.loc[duplicates, "timestamp"].iloc[0],
severity="medium",
description=f"{dup_count} doppelte Liquidation-Events gefunden",
details={"duplicate_count": dup_count}
))
def generate_quality_report(self) -> str:
"""Erstellt einen lesbaren Qualitätsbericht"""
report = []
report.append("=" * 60)
report.append(f"DATENQUALITÄTSBERICHT: {self.exchange}/{self.symbol}")
report.append("=" * 60)
if not self.issues:
report.append("✓ Keine Probleme gefunden!")
return "\n".join(report)
# Gruppiere nach Schweregrad
by_severity = {}
for issue in self.issues:
if issue.severity not in by_severity:
by_severity[issue.severity] = []
by_severity[issue.severity].append(issue)
for severity in ["critical", "high", "medium", "low"]:
if severity in by_severity:
report.append(f"\n🔴 {severity.upper()}: {len(by_severity[severity])} Probleme")
for issue in by_severity[severity]:
report.append(f" • {issue.description}")
report.append(f" Zeitpunkt: {issue.timestamp}")
return "\n".join(report)KI-gestützte Anomalieanalyse mit HolySheep AI
# holy_sheep_analyzer.py
import httpx
import json
import asyncio
from typing import List, Dict
from HOLYSHEEP_CONFIG import HOLYSHEEP_CONFIG
class HolySheepAnalyzer:
"""
Nutzt HolySheep AI für fortgeschrittene Anomalieerkennung.
Vorteile:
- ¥1=$1 (85%+ Ersparnis gegenüber OpenAI)
- <50ms Latenz
- WeChat/Alipay Zahlung
- Kostenlose Credits für Tests
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_CONFIG["base_url"] # https://api.holysheep.ai/v1
self.model = HOLYSHEEP_CONFIG["model"]
async def analyze_anomalies(
self,
issues: List[Dict],
context: Dict
) -> Dict:
"""
Analysiert erkannte Qualitätsprobleme mit KI.
Kosteneffizient: DeepSeek V3.2 kostet nur $0.42/MTok
"""
prompt = self._build_analysis_prompt(issues, context)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": """Sie sind ein Experte für Kryptowährungs-Marktdaten-Analyse.
Analysieren Sie die gegebenen Datenqualitätsprobleme und geben Sie:
1. Eine Zusammenfassung der Hauptprobleme
2. Mögliche Ursachen (z.B. Exchange-Fehler, API-Probleme, Marktbedingungen)
3. Empfohlene Aktionen zur Behebung
4. Eine Datenqualitätsbewertung von 0-100"""
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.1,
"max_tokens": 800
}
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
return {
"analysis": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"cost_usd": self._calculate_cost(result.get("usage", {}))
}
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
raise ConnectionError(
"401 Unauthorized: Ungültiger HolySheep API-Key. "
"Registrieren Sie sich unter https://www.holysheep.ai/register"
) from e
raise
def _build_analysis_prompt(self, issues: List[Dict], context: Dict) -> str:
"""Baut den Analyse-Prompt"""
issues_summary = "\n".join([
f"- {i['issue_type']} ({i['severity']}): {i['description']}"
for i in issues[:20] # Max 20 Issues für Kostenkontrolle
])
return f"""Analysieren Sie folgende Datenqualitätsprobleme bei
Liquidation-Events von Exchange {context.get('exchange')}/{context.get('symbol')}:
Zeitraum: {context.get('start_date')} bis {context.get('end_date')}
Gesamte Events: {context.get('total_events')}
Zeitraum in Tagen: {context.get('days')}
GEFUNDENE PROBLEME:
{issues_summary}
Bitte geben Sie eine strukturierte Analyse zurück."""
def _calculate_cost(self, usage: Dict) -> float:
"""Berechnet die Kosten basierend auf dem Modell"""
model_costs = {
"deepseek-v3.2": 0.42, # $0.42/MTok input + output
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50 # $2.50/MTok
}
cost_per_mtok = model_costs.get(self.model, 0.42)
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
# Kosten in USD (1 MTok = 1,000,000 Tokens)
return (total_tokens / 1_000_000) * cost_per_mtok
Beispiel-Nutzung
async def analyze_with_ai():
analyzer = HolySheepAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
issues = [
{
"issue_type": "MISSING_INTERVAL",
"severity": "high",
"description": "2 Stunden ohne Daten",
"timestamp": "2025-12-01 14:00:00"
},
{
"issue_type": "PRICE_JUMP",
"severity": "critical",
"description": "45% Preissprung",
"timestamp": "2025-12-01 15:30:00"
}
]
context = {
"exchange": "binance",
"symbol": "BTC-USDT",
"start_date": "2025-12-01",
"end_date": "2025-12-02",
"total_events": 1547,
"days": 1
}
try:
result = await analyzer.analyze_anomalies(issues, context)
print("📊 KI-Analyse:")
print(result["analysis"])
print(f"\n💰 Kosten: ${result['cost_usd']:.4f}")
print(f"📈 Token-Nutzung: {result['usage']}")
except ConnectionError as e:
print(f"Fehler: {e}")Komplettes Audit-Skript mit automatischer Berichterstattung
# liquidation_audit.py - Komplettes Audit-Tool
import asyncio
import pandas as pd
from datetime import datetime, timedelta
from tardis_fetcher import TardisLiquidationFetcher
from quality_checker import LiquidationQualityChecker, QualityIssue
from holy_sheep_analyzer import HolySheepAnalyzer
import json
import os
async def run_full_audit(
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
tardis_api_key: str,
holysheep_api_key: str,
output_dir: str = "./audit_reports"
):
"""
Führt ein vollständiges Datenqualitäts-Audit durch.
Workflow:
1. Daten von Tardis abrufen
2. Qualitätsprüfungen durchführen
3. KI-gestützte Analyse (optional)
4. Bericht generieren
"""
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
print("=" * 60)
print(f"清算价数据质量稽核: {exchange}/{symbol}")
print(f"Zeitraum: {start_date} - {end_date}")
print("=" * 60)
# Schritt 1: Daten abrufen
print("\n📡 Schritt 1: Rufe Daten von Tardis ab...")
fetcher = TardisLiquidationFetcher(tardis_api_key)
try:
raw_data = await fetcher.fetch_with_retry(
exchange=exchange,
symbol=symbol,
start_date=start_date,
end_date=end_date
)
print(f" ✓ {len(raw_data)} Rohdatensätze erhalten")
except ConnectionError as e:
print(f" ✗ Verbindungsfehler: {e}")
# Fallback-Strategie: Verwende gepufferte Daten oder Demo-Daten
print(" → Verwende Demo-Daten für Demonstration...")
raw_data = generate_demo_data(symbol, start_date, end_date)
# Schritt 2: In DataFrame konvertieren
print("\n🔧 Schritt 2: Datenaufbereitung...")
df = pd.DataFrame(raw_data)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
print(f" ✓ DataFrame erstellt: {len(df)} Zeilen, {len(df.columns)} Spalten")
# Schritt 3: Qualitätsprüfung
print("\n🔍 Schritt 3: Qualitätsprüfung...")
checker = LiquidationQualityChecker(df, exchange, symbol)
issues = checker.check_all()
issues_data = [
{
"issue_type": i.issue_type,
"severity": i.severity,
"description": i.description,
"timestamp": str(i.timestamp),
"details": i.details
}
for i in issues
]
# Schritt 4: KI-Analyse (optional)
print("\n🤖 Schritt 4: KI-gestützte Analyse...")
ai_analysis = None
if issues and holysheep_api_key:
try:
analyzer = HolySheepAnalyzer(holysheep_api_key)
context = {
"exchange": exchange,
"symbol": symbol,
"start_date": str(start_date),
"end_date": str(end_date),
"total_events": len(df),
"days": (end_date - start_date).days
}
ai_result = await analyzer.analyze_anomalies(issues_data, context)
ai_analysis = ai_result["analysis"]
print(f" ✓ KI-Analyse abgeschlossen")
print(f" 💰 Kosten: ${ai_result['cost_usd']:.4f}")
except Exception as e:
print(f" ⚠ KI-Analyse fehlgeschlagen: {e}")
# Schritt 5: Bericht speichern
print("\n💾 Schritt 5: Berichte werden gespeichert...")
report = {
"audit_metadata": {
"exchange": exchange,
"symbol": symbol,
"start_date": str(start_date),
"end_date": str(end_date),
"audit_timestamp": str(datetime.now()),
"total_events": len(df)
},
"quality_summary": {
"total_issues": len(issues),
"critical": len([i for i in issues if i.severity == "critical"]),
"high": len([i for i in issues if i.severity == "high"]),
"medium": len([i for i in issues if i.severity == "medium"]),
"low": len([i for i in issues if i.severity == "low"])
},
"issues": issues_data,
"ai_analysis": ai_analysis,
"data_quality_score": calculate_quality_score(issues, len(df))
}
# Speichere JSON-Bericht
json_path = f"{output_dir}/audit_{exchange}_{symbol}_{timestamp}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
# Speichere CSV der Probleme
if issues:
issues_df = pd.DataFrame(issues_data)
csv_path = f"{output_dir}/issues_{exchange}_{symbol}_{timestamp}.csv"
issues_df.to_csv(csv_path, index=False)
# Generiere Textbericht
text_report = checker.generate_quality_report()
if ai_analysis:
text_report += "\n\n" + "=" * 60
text_report += "\n🤖 KI-GESTÜTZTE ANALYSE:\n"
text_report += "=" * 60
text_report += f"\n{ai_analysis}"
report_path = f"{output_dir}/report_{exchange}_{symbol}_{timestamp}.txt"
with open(report_path, "w", encoding="utf-8") as f:
f.write(text_report)
print(f" ✓ JSON: {json_path}")
print(f" ✓ Text: {report_path}")
if issues:
print(f" ✓ CSV: {csv_path}")
print("\n" + "=" * 60)
print("✅ Audit abgeschlossen!")
print(f"📊 Datenqualitätsscore: {report['data_quality_score']}/100")
print("=" * 60)
return report
def generate_demo_data(symbol: str, start: datetime, end: datetime) -> list:
"""Generiert Demo-Daten für Testzwecke"""
import random
data = []
current = start
base_price = 50000 if "BTC" in symbol else 3000
while current < end:
data.append({
"timestamp": current.isoformat(),
"symbol": symbol,
"liquidation_price": base_price * (1 + random.uniform(-0.1, 0.1)),
"liquidation_volume": random.uniform(0.5, 5.0),
"side": random.choice(["long", "short"]),
"exchange": "binance"
})
current += timedelta(minutes=random.randint(1, 5))
# Füge absichtlich Fehler ein
if len(data) > 10:
# Große Lücke
data.insert(5, data[5]) # Duplikat
# Preissprung
data[10]["liquidation_price"] = data[9]["liquidation_price"] * 1.5
return data
def calculate_quality_score(issues: List[QualityIssue], total_records: int) -> int:
"""Berechnet einen Datenqualitätsscore von 0-100"""
if total_records == 0:
return 0
penalties = {
"critical": 25,
"high": 10,
"medium": 3,
"low": 1
}
total_penalty = sum(penalties.get(i.severity, 0) for i in issues)
# Score sinkt mit Problemen, Minimum 0
score = max(0, 100 - total_penalty)
# Zusätzliche Strafe für kritische Probleme
critical_count = len([i for i in issues if i.severity == "critical"])
if critical_count > 0:
score = max(0, score - critical_count * 5)
return min(100, score)
Hauptprogramm
if __name__ == "__main__":
import sys
# Konfiguration
EXCHANGE = "binance"
SYMBOL = "BTC-USDT"
START = datetime(2025, 12, 1)
END = datetime(2025, 12, 2)
# API-Keys aus Umgebung oder Demo-Modus
TARDIS_KEY = os.getenv("TARDIS_API_KEY", "demo")
HOLYSHEEP_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Audit ausführen
result = asyncio.run(run_full_audit(
exchange=EXCHANGE,
symbol=SYMBOL,
start_date=START,
end_date=END,
tardis_api_key=TARDIS_KEY,
holysheep_api_key=HOLYSHEEP_KEY
))
print("\n📋 Zusammenfassung:")
print(json.dumps(result["quality_summary"], indent=2))Häufige Fehler und Lösungen
1. ConnectionError: 401 Unauthorized bei Tardis
# FEHLER:
ConnectionError: 401 Unauthorized: Invalid API key for exchange data access
LÖSUNG: API-Key korrekt konfigurieren
import os
Option 1: Umgebungsvariable setzen
Linux/Mac: export TARDIS_API_KEY="ihr_key_hier"
Windows: set TARDIS_API_KEY="ihr_key_hier"
Option 2: Direkt im Code (nicht für Produktion!)
TARDIS_API_KEY = "tk_live_xxxxxxxxxxxx" # Korrektes Format prüfen
Option 3: Key validieren
def validate_tardis_key(api_key: str) -> bool:
"""Validiert den Tardis API-Key"""
import re
# Tardis Keys beginnen mit "tk_live_" oder "tk_test_"
pattern = r'^tk_(live|test)_[a-zA-Z0-9]{20,}$'
return bool(re.match(pattern, api_key))
Test
if not validate_tardis_key(TARDIS_API_KEY):
raise ValueError("Ungültiges Tardis API-Key Format!")2. Timeout-Fehler bei grossen Datenmengen
# FEHLER:
httpx.TimeoutException: Request timeout after 30.0s
Maximale Datenmenge überschritten
LÖSUNG: Chunked Download mit Fortschrittsanzeige
async def fetch_in_chunks(fetcher, exchange, symbol, start, end, chunk_days=7):
"""Holt Daten in kleineren Teilen"""
from datetime import timedelta
all_data = []
current_start = start
total_chunks = (end - start).days // chunk_days + 1
print(f"Lade {total_chunks} Blöcke...")
for i in range(total_chunks):
current_end = min(current_start + timedelta(days=chunk_days), end)
try:
chunk_data = await fetcher.fetch_with_retry(
exchange=exchange,
symbol=symbol,
start_date=current_start,
end_date=current_end,
max_retries=3
)
all_data.extend(chunk_data)
print(f" ✓ Block {i+1}/{total_chunks}: {len(chunk_data)} Events")
except ConnectionError as e:
print(f" ✗ Block {i+1} fehlgeschlagen: {e}")
# Alternativ: Überspringen oder aus Cache laden
current_start = current_end + timedelta(seconds=1)
await asyncio.sleep(1) # Rate Limiting respektieren
return all_data3. Fehlerhafte Zeitstempel und Zeitzonenprobleme
# FEHLER:
Timestamps zeigen in UTC statt lokaler Zeit
Oder: "2025-12-01" wird als "2025-12-01 00:00:00" interpretiert
Datenlücken durch Zeitzonen-Inkonsistenzen
LÖSUNG: Explizite Zeitzonenkonvertierung
import pytz
from datetime import datetime
def normalize_timestamps(df: pd.DataFrame, timezone: str = "UTC") -> pd.DataFrame:
"""Normalisiert alle Timestamps auf eine einheitliche Zeitzone"""
tz = pytz.timezone(timezone)
if "timestamp" in df.columns:
# Konvertiere zu UTC und dann zur Zielzeitzone
df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
df["timestamp"] = df["timestamp"].dt.tz_convert(tz)
df["timestamp"] = df["timestamp"].dt.tz_localize(None) # Als naive datetime
# Prüfe auf Winter-/Sommerzeit-Probleme
if "exchange_time" in df.columns:
df["exchange_time"] = pd.to_datetime(df