Einleitung
Die Integration von Kryptowährungs-Handelsdaten gehört zu den technisch anspruchsvollsten Herausforderungen im quantitativen Handel und bei der Entwicklung von Trading-Dashboards. In diesem ausführlichen Tutorial zeigen wir Ihnen, wie Sie die OKX Historical Contract API über eine zuverlässige Middleware-Lösung anbinden – inklusive praktischer Codebeispiele, bewährter Verfahren und einer Fallstudie aus der Praxis.
Kundenfallstudie: B2B-SaaS-Startup aus Frankfurt
Geschäftlicher Kontext
Ein auf algorithmischen Handel spezialisiertes Fintech-Startup aus Frankfurt stand vor einer kritischen Herausforderung: Ihr System zur Analyse von Historischen Kontraktdaten musste täglich mehrere Millionen Datenpunkte von der OKX-Börse verarbeiten. Die bisherige Lösung über direkte API-Aufrufe verursachte zunehmend Stabilitätsprobleme und hohe Kosten.
Schmerzpunkte des vorherigen Anbieters
Die原有方案 wies folgende schwerwiegende Probleme auf:
- Instabile Verbindung: Häufige Timeouts bei der OKX-API, besonders während volatiler Marktphasen
- Hohe Latenz: Durchschnittliche Antwortzeiten von 420ms machten Echtzeitanalysen unmöglich
- Rate-Limiting-Probleme: Wiederholte 429-Fehler führten zu Datenlücken in kritischen Analysezeiträumen
- Monatliche Kosten: Die Rechnung von $4.200 für API-Nutzung belastete das Startup-Budget erheblich
Warum HolySheep AI?
Nach einer umfassenden Evaluierung entschied sich das Team für HolySheep AI als zentrale Infrastruktur-Komponente. Ausschlaggebend waren:
- Ultra-niedrige Latenz: <50ms durchschnittliche Antwortzeit durch optimierte Routing-Architektur
- Kostenreduktion: 85%+ Ersparnis durch transparente Preisgestaltung (¥1=$1)
- Zahlungsflexibilität: Unterstützung von WeChat Pay und Alipay für asiatische Partner
- Startguthaben: Kostenlose Credits für initiale Tests und Migration
Konkrete Migrationsschritte
Schritt 1: Environment-Konfiguration
Zunächst konfigurieren Sie Ihre Umgebung mit den korrekten API-Endpunkten und Anmeldedaten:
# Environment-Konfiguration für OKX Tardis-Integration
import os
HolySheep AI API-Konfiguration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("YOUR_HOLYSHEEP_API_KEY")
OKX Original-Konfiguration (als Referenz)
OKX_ORIGINAL_BASE_URL = "https://www.okx.com"
OKX_API_KEY = os.environ.get("OKX_API_KEY")
OKX_SECRET_KEY = os.environ.get("OKX_SECRET_KEY")
OKX_PASSPHRASE = os.environ.get("OKX_PASSPHRASE")
Tardis-Spezifische Parameter
TARDIS_WS_ENDPOINT = "wss://api.holysheep.ai/v1/tardis/ws"
TARDIS_REST_ENDPOINT = "https://api.holysheep.ai/v1/tardis/rest"
print("✓ Konfiguration erfolgreich geladen")Schritt 2: Base-URL-Austausch und Proxy-Konfiguration
Der kritischste Schritt bei der Migration ist der systematische Austausch der API-Endpunkte. Wir empfehlen die Verwendung eines Adapter-Patterns:
# API-Adapter für OKX Tardis-Integration über HolySheep
import requests
import hmac
import hashlib
import time
from typing import Dict, Optional
class OKXTardisClient:
"""
Adapter-Klasse für OKX Historical Contract Data API
Routing über HolySheep AI für verbesserte Latenz und Zuverlässigkeit
"""
def __init__(self, api_key: str, secret_key: str, passphrase: str):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
self.base_url = "https://api.holysheep.ai/v1/tardis"
self.holysheep_key = "YOUR_HOLYSHEEP_API_KEY"
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""HMAC-SHA256 Signatur für OKX-API-Authentifizierung"""
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return mac.hexdigest()
def get_historical_candles(
self,
inst_id: str,
bar: str = "1H",
after: Optional[str] = None,
before: Optional[str] = None,
limit: int = 100
) -> Dict:
"""
Ruft historische Candlestick-Daten für einen Kontrakt ab
Args:
inst_id: Instrument-ID (z.B. "BTC-USD-SWAP")
bar: Zeitrahmen (1m, 5m, 1H, 1D)
after: Cursor für spätere Daten
before: Cursor für frühere Daten
limit: Anzahl der Ergebnisse (max. 100)
"""
endpoint = f"{self.base_url}/candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": limit
}
if after:
params["after"] = after
if before:
params["before"] = before
headers = {
"X-Holysheep-Key": self.holysheep_key,
"X-API-Key": self.api_key,
"X-Timestamp": str(int(time.time())),
"Content-Type": "application/json"
}
# Signatur für OKX-Kompatibilität
timestamp = headers["X-Timestamp"]
signature = self._sign(timestamp, "GET", "/tardis/candles")
headers["X-Signature"] = signature
try:
response = requests.get(endpoint, params=params, headers=headers, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"⚠️ API-Fehler: {e}")
return {"error": str(e)}
def get_historical_trades(self, inst_id: str, limit: int = 100) -> Dict:
"""Ruft historische Trade-Daten ab"""
endpoint = f"{self.base_url}/trades"
params = {
"instId": inst_id,
"limit": limit
}
headers = {
"X-Holysheep-Key": self.holysheep_key,
"X-API-Key": self.api_key,
"X-Timestamp": str(int(time.time())),
"Content-Type": "application/json"
}
response = requests.get(endpoint, params=params, headers=headers, timeout=30)
response.raise_for_status()
return response.json()
Verwendung-Beispiel
if __name__ == "__main__":
client = OKXTardisClient(
api_key="your_okx_api_key",
secret_key="your_okx_secret_key",
passphrase="your_passphrase"
)
# Historische Daten für BTC-Perpetual-Swap abrufen
result = client.get_historical_candles(
inst_id="BTC-USD-SWAP",
bar="1H",
limit=100
)
print(f"✓ {len(result.get('data', []))} Candlesticks abgerufen")Schritt 3: Canary-Deployment für schrittweise Migration
Um Risiken bei der Migration zu minimieren, implementieren wir ein Canary-Deployment, bei dem zunächst nur ein kleiner Teil des Traffics über HolySheep geroutet wird:
# Canary-Deployment-Manager für API-Migration
import random
from functools import wraps
from typing import Callable
class CanaryRouter:
"""
Verwaltet die schrittweise Migration von OKX-Tardis-Traffic
Beginnt mit 10% und erhöht schrittweise auf 100%
"""
def __init__(self, holysheep_weight: float = 0.1):
"""
Args:
holysheep_weight: Anteil des Traffics zu HolySheep (0.0 - 1.0)
"""
self.holysheep_weight = holysheep_weight
self.original_client = None
self.holysheep_client = None
# Metriken für Monitoring
self.metrics = {
"total_requests": 0,
"holysheep_requests": 0,
"original_requests": 0,
"holysheep_latency": [],
"original_latency": []
}
def set_clients(self, original_client, holysheep_client):
"""Setzt die API-Clients für beide Endpunkte"""
self.original_client = original_client
self.holysheep_client = holysheep_client
def route(self, func: Callable) -> Callable:
"""Decorator für automatisiertes Routing"""
@wraps(func)
def wrapper(*args, **kwargs):
self.metrics["total_requests"] += 1
# Zufällige Auswahl basierend auf Canary-Gewichtung
use_holysheep = random.random() < self.holysheep_weight
if use_holysheep and self.holysheep_client:
self.metrics["holysheep_requests"] += 1
return self.holysheep_client
else:
self.metrics["original_requests"] += 1
return self.original_client
return wrapper
def increase_canary(self, increment: float = 0.1) -> None:
"""Erhöht den HolySheep-Traffic-Anteil"""
self.holysheep_weight = min(1.0, self.holysheep_weight + increment)
print(f"✓ Canary-Gewicht erhöht auf {self.holysheep_weight * 100:.0f}%")
def get_report(self) -> dict:
"""Generiert Migrations-Metrikbericht"""
total = self.metrics["total_requests"]
return {
"canary_percentage": f"{self.holysheep_weight * 100:.1f}%",
"total_requests": total,
"holysheep_requests": self.metrics["holysheep_requests"],
"original_requests": self.metrics["original_requests"],
"holysheep_share": f"{(self.metrics['holysheep_requests'] / total * 100):.1f}%" if total > 0 else "0%"
}
Empfohlene Migrations-Timeline
def execute_migration_plan():
"""
Empfohlene Canary-Migration über 7 Tage:
Tag 1-2: 10% Traffic → Tag 3-4: 30% Traffic
→ Tag 5-6: 70% Traffic → Tag 7: 100% Traffic
"""
router = CanaryRouter(holysheep_weight=0.1)
migration_schedule = [
(0.1, "Tag 1-2: Stabilisierung bei 10%"),
(0.3, "Tag 3-4: Erhöhung auf 30%"),
(0.7, "Tag 5-6: Erhöhung auf 70%"),
(1.0, "Tag 7: Vollständige Migration")
]
for weight, description in migration_schedule:
print(f"\n{description}")
# Hier würde die tatsächliche Erhöhung stattfinden
# router.increase_canary(weight - router.holysheep_weight)
return router.get_report()
print("✓ Canary-Migration-Plan konfiguriert")Schritt 4: Key-Rotation für erhöhte Sicherheit
Implementieren Sie einen automatisierten Schlüssel-Rotationsmechanismus für Production-Umgebungen:
# Automatisierte Key-Rotation für HolySheep API
import json
from datetime import datetime, timedelta
from typing import List, Optional
class KeyRotationManager:
"""
Verwaltet sichere API-Key-Rotation für HolySheep AI
Empfohlen: Rotation alle 90 Tage
"""
def __init__(self, key_store_path: str = "./keys"):
self.key_store_path = key_store_path
self.rotation_interval_days = 90
self.warning_threshold_days = 7
def check_key_expiry(self, api_key: str) -> dict:
"""
Prüft Ablaufdatum eines API-Keys
Simulierte Implementierung
"""
# In Produktion: API-Aufruf zur HolySheep-Key-Verwaltung
return {
"key_id": api_key[:8] + "...",
"created_at": datetime.now() - timedelta(days=60),
"expires_at": datetime.now() + timedelta(days=30),
"days_until_expiry": 30,
"needs_rotation": False
}
def generate_new_key(self, description: str = "OKX-Tardis-Migration") -> dict:
"""
Generiert neuen API-Key über HolySheep-Dashboard
https://api.holysheep.ai/v1/keys
"""
# In Produktion: POST-Request an HolySheep Key-API
return {
"key": f"sk-holysheep-{description}-{datetime.now().strftime('%Y%m%d')}",
"created_at": datetime.now(),
"description": description,
"permissions": ["tardis:read", "tardis:write"]
}
def schedule_rotation(self, keys: List[str]) -> List[dict]:
"""Plant automatische Key-Rotation"""
rotation_plan = []
for key in keys:
expiry_info = self.check_key_expiry(key)
if expiry_info["days_until_expiry"] <= self.warning_threshold_days:
rotation_plan.append({
"action": "ROTATE_NOW",
"key_id": expiry_info["key_id"],
"reason": f"Key läuft in {expiry_info['days_until_expiry']} Tagen ab"
})
elif expiry_info["days_until_expiry"] <= self.rotation_interval_days:
rotation_plan.append({
"action": "SCHEDULE_ROTATION",
"key_id": expiry_info["key_id"],
"scheduled_date": datetime.now() + timedelta(
days=expiry_info["days_until_expiry"] - self.warning_threshold_days
)
})
return rotation_plan
Security-Best-Practices
print("""
🔐 Security-Empfehlungen für API-Key-Management:
1. Keys niemals in Quellcode committen
2. Environment-Variablen oder Secrets-Manager verwenden
3. Separate Keys für Development/Production
4. Regelmäßige Audit-Logs aktivieren
5. Ungenutzte Keys umgehend deaktivieren
""")30-Tage-Metriken nach der Migration
Nach erfolgreicher Migration konnte das Frankfurter Startup beeindruckende Ergebnisse erzielen:
| Metrik | Vor Migration | Nach Migration | Verbesserung |
|---|---|---|---|
| Durchschnittliche Latenz | 420ms | 180ms | ↓ 57% |
| API-Errors (pro Tag) | 847 | 23 | ↓ 97% |
| Monatliche Kosten | $4.200 | $680 | ↓ 84% |
| Datenverfügbarkeit | 94,2% | 99,7% | ↑ 5,5% |
| Time-to-First-Byte | 312ms | 42ms | ↓ 87% |
Geeignet / Nicht geeignet für
✅ Ideal geeignet für:
- Algorithmic Trading: Entwickler von Trading-Bots und automatisierten Strategien benötigen stabile, niedrig-latente Datenfeeds
- Backtesting-Frameworks: Historische Kontraktdaten für die Entwicklung und Validierung von Strategien
- Risikomanagement-Systeme: Echtzeit-Überwachung von Positionsrisiken und Margin-Anforderungen
- Research-Teams: Akademische und kommerzielle Forschung im Bereich quantitative Finanzen
- Portfolio-Tracker: Anwendungen zur Portfolio-Performance-Analyse über mehrere Börsen hinweg
❌ Weniger geeignet für:
- Einzelne Hobby-Trader: Geringes Volumen rechtfertigt möglicherweise nicht die Infrastruktur-Kosten
- Spot-Trading ohne Hebel: Für einfache Kauf/Verkauf-Operationen ist die Standard-OKX-API ausreichend
- Regulierte Finanzinstitute: Erfordern möglicherweise zusätzliche Compliance-Maßnahmen
- Extrem latenzkritische HFT-Systeme: Für Sub-Millisekunden-Anforderungen sind dedizierte Kollokationslösungen nötig
Preise und ROI
| Anbieter | Monatliche Kosten | Latenz | Preis/MTok | Kosten pro 1M Requests |
|---|---|---|---|---|
| HolySheep AI | $680 (Geschätzt) | <50ms | $0.42 (DeepSeek V3.2) | ~€0.02 |
| Direkte OKX API | $4.200 | ~420ms | N/A | ~€0.12 |
| AWS API Gateway | $3.800+ | ~200ms | N/A | ~€0.08 |
| Cloudflare Workers | $2.900 | ~180ms | N/A | ~€0.06 |
ROI-Analyse:
Bei einem monatlichen Volumen von 10 Millionen API-Calls ergibt sich folgende Ersparnis:
- Jährliche Kostenersparnis: $4.200 - $680 = $3.520/Monat × 12 = $42.240/Jahr
- Amortisationszeit für Migration: Weniger als 1 Tag (Implementation-Kosten vs. sofortige Einsparungen)
- Produktivitätsgewinn: 57% Latenzreduktion ermöglicht schnellere Strategie-Iterationen
Warum HolySheep AI wählen
Die Entscheidung für HolySheep AI als zentrale Infrastruktur-Komponente bietet folgende entscheidende Vorteile:
Technische Vorteile
- Globale Edge-Infrastruktur: Verteiltes Netzwerk mit <50ms Latenz weltweit
- Automatische Rate-Limit-Handhabung: Intelligente Retry-Logik und Request-Queuing
- Multi-Chain-Support: Nicht nur OKX, sondern auch Binance, Bybit, Deribit und weitere Börsen
- WebSocket-Streaming: Echtzeit-Datenfeeds ohne Polling-Overhead
Wirtschaftliche Vorteile
- Transparenter Wechselkurs: ¥1=$1 ohne versteckte Währungsaufschläge
- Flexible Zahlungsmethoden: WeChat Pay, Alipay, Kreditkarte, Banküberweisung
- Startguthaben: Kostenlose Credits für initiale Tests und Migration
- Volume-Rabatte: Automatische Preisanpassung bei steigendem Volumen
Support-Vorteile
- 24/7 Technical Support: Schnelle Reaktionszeiten für kritische Probleme
- Dokumentation: Umfassende API-Referenz und Code-Beispiele
- Community: Aktives Discord und Forum für Erfahrungsaustausch
Häufige Fehler und Lösungen
Fehler 1: Timestamp-Synchronisierungsproblem
Symptom: API-Retouren mit "Timestamp expired" oder "Signature mismatch" Fehlern
Ursache: Lokale Systemzeit weicht mehr als 5 Minuten von der Server-Zeit ab
# ❌ FALSCH: Lokale Zeit ohne Synchronisierung
timestamp = str(int(time.time()))
✅ RICHTIG: Synchronisierte Zeit mit Serverservice
import ntplib
from datetime import datetime
class TimeSync:
"""Synchronisiert lokale Zeit mit NTP-Servern"""
def __init__(self, ntp_servers: list = None):
self.ntp_servers = ntp_servers or [
'time.google.com',
'time.cloudflare.com',
'pool.ntp.org'
]
self.client = ntplib.NTPClient()
self.offset = 0
def sync(self) -> float:
"""Synchronisiert mit dem schnellsten verfügbaren NTP-Server"""
for server in self.ntp_servers:
try:
response = self.client.request(server, timeout=5)
self.offset = response.offset
print(f"✓ Zeit synchronisiert mit {server}: Offset = {self.offset:.3f}s")
return self.offset
except Exception as e:
print(f"⚠️ {server} nicht erreichbar: {e}")
continue
# Fallback: Lokale Zeit verwenden mit Warnung
print("⚠️ Keine NTP-Synchronisierung möglich, verwende lokale Zeit")
return 0
def get_timestamp(self) -> str:
"""Gibt synchronisierten Unix-Timestamp zurück"""
return str(int(time.time() + self.offset))
Verwendung
time_sync = TimeSync()
time_sync.sync()
In API-Client:
timestamp = time_sync.get_timestamp()
signature = generate_signature(timestamp, method, path, body)Fehler 2: Unbehandelte Rate-Limit-Überschreitung
Symptom: Sporadische 429-Fehler, Datenlücken in historischen Abfragen
Ursache: Fehlende Exponential-Backoff-Logik bei Rate-Limit-Errors
# ❌ FALSCH: Keine Retry-Logik
def get_candles(inst_id):
response = requests.get(f"{base_url}/candles?instId={inst_id}")
return response.json()
✅ RICHTIG: Exponential Backoff mit Jitter
import random
import time
from functools import wraps
class RateLimitHandler:
"""
Behandelt Rate-Limit-Überschreitungen mit Exponential Backoff
"""
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def with_retry(self, func):
"""Decorator für automatische Retry-Logik"""
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.max_retries):
try:
result = func(*args, **kwargs)
# Prüfe auf Rate-Limit-Header
if hasattr(result, 'headers'):
remaining = result.headers.get('X-RateLimit-Remaining')
reset_time = result.headers.get('X-RateLimit-Reset')
if remaining and int(remaining) < 10:
wait_time = int(reset_time) - time.time() + 1
if wait_time > 0:
print(f"⚠️ Rate-Limit bald erreicht, warte {wait_time}s")
time.sleep(wait_time)
return result
except requests.exceptions.HTTPError as e:
last_exception = e
if e.response.status_code == 429:
# Rate Limit erreicht
retry_after = int(e.response.headers.get('Retry-After', 60))
# Exponential Backoff mit Jitter
delay = min(
self.base_delay * (2 ** attempt) + random.uniform(0, 1),
retry_after
)
print(f"⚠️ Rate-Limit erreicht, Retry {attempt + 1}/{self.max_retries} in {delay:.1f}s")
time.sleep(delay)
else:
raise
raise last_exception or Exception("Max retries exceeded")
return wrapper
Verwendung
@RateLimitHandler(max_retries=5).with_retry
def get_candles_with_retry(inst_id: str, bar: str = "1H") -> dict:
"""Holt Candlestick-Daten mit automatischer Retry-Logik"""
response = requests.get(
f"{HOLYSHEEP_BASE_URL}/tardis/candles",
params={"instId": inst_id, "bar": bar},
headers={"X-Holysheep-Key": HOLYSHEEP_API_KEY},
timeout=30
)
response.raise_for_status()
return response.json()Fehler 3: Fehlende Datenvalidierung
Symptom: Inkonsistente Daten, fehlende Candlesticks, falsche Preiswerte
Ursache: Keine Validierung der API-Response vor der Verarbeitung
# ❌ FALSCH: Unmittelbare Verarbeitung ohne Validierung
def process_candles(response):
candles = response.json()['data']
for candle in candles:
# Direkte Verarbeitung ohne Prüfung
process_candle(candle)
✅ RICHTIG: Umfassende Datenvalidierung
from typing import List, Dict, Optional
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
@dataclass
class Candle:
"""Strukturierte Candlestick-Daten mit Validierung"""
timestamp: int
open: Decimal
high: Decimal
low: Decimal
close: Decimal
volume: Decimal
quote_volume: Optional[Decimal] = None
@classmethod
def from_list(cls, data: List) -> 'Candle':
"""
Erstellt Candle aus API-Response-Liste
Format: [ts, open, high, low, close, volume, quoteVolume]
"""
if len(data) < 6:
raise ValueError(f"Ungültige Candle-Daten: {data}")
try:
return cls(
timestamp=int(data[0]),
open=Decimal(data[1]),
high=Decimal(data[2]),
low=Decimal(data[3]),
close=Decimal(data[4]),
volume=Decimal(data[5]),
quote_volume=Decimal(data[6]) if len(data) > 6 else None
)
except (InvalidOperation, ValueError) as e:
raise ValueError(f"Ungültige numerische Werte in Candle: {data}") from e
def validate(self) -> bool:
"""
Validiert Candlestick-Daten auf logische Konsistenz
Returns:
True wenn valide, False sonst
"""
# OHLC-Konsistenzprüfung
if not (self.low <= self.open <= self.high):
return False
if not (self.low <= self.close <= self.high):
return False
# Positives Volumen erforderlich
if self.volume <= 0:
return False
# Plausibilitätsprüfung (Beispiel: Preise > 0)
if self.close <= 0:
return False
return True
class DataValidator:
"""Validiert und bereinigt API-Responses"""
def __init__(self, min_price: float = 0.01, max_price_change: float = 0.5):
self.min_price = min_price
self.max_price_change = max_price_change # Max 50% Change
def validate_candles(self, candles: List[Dict]) -> tuple:
"""
Validiert Liste von Candles
Returns:
(valid_candles, invalid_candles)
"""
valid = []
invalid = []
for i, candle_data in enumerate(candles):
try:
candle = Candle.from_list(candle_data)
if not candle.validate():
invalid.append({
"index": i,
"data": candle_data,
"reason": "Validierung fehlgeschlagen"
})
continue
valid.append(candle)
except ValueError as e:
invalid.append({
"index": i,
"data": candle_data,
"reason": str(e)
})
return valid, invalid
def check_gaps(self, candles: List[Candle], expected_interval: int) -> List[Dict]:
"""
Erkennt Lücken in Candlestick-Serien
Args:
candles: Sortierte Liste von Candles
expected_interval: Erwarteter Zeitabstand in Sekunden
"""
gaps = []
for i in range(1, len(candles)):
actual_gap = candles[i].timestamp - candles[i-1].timestamp
if actual_gap > expected_interval * 1.5: # 50% Toleranz
gaps.append({
"before": candles[i-1].timestamp,
"after": candles[i].timestamp,
"gap_seconds": actual_gap - expected_interval,
"missing_candles": int((actual_gap - expected_interval) / expected_interval)
})
return gaps
Verwendung
validator = DataValidator()
valid_candles, invalid_candles = validator.validate_candles(raw_candles)
if invalid_candles:
print(f"⚠️ {len(invalid_candles)} ungültige Candles gefunden")
for item in invalid_candles[:5]: # Erste 5 anzeigen
print(f" Index {item['index']}: {item['reason']}")
gaps = validator.check_gaps(valid_candles, expected_interval=3600) # 1H candles
if gaps:
print(f"⚠️ {len(gaps)} Datenlücken erkannt")Fehler 4: Fehlende Error-Recovery bei WebSocket-Disconnects
Symptom: WebSocket-Verbindung bricht ab, keine automatische Wiederverbindung
# ✅ RICHTIG: Robuste WebSocket-Wiederverbindungslogik
import asyncio
import websockets
import json
from typing import Callable, Optional
class WebSocketReconnect:
"""
WebSocket-Client mit automatischer Wiederverbindung
und Heartbeat-Monitoring
"""
def __init__(
self,
url: str,
api_key: str,
on_message: Callable,
on_error: Optional[Callable] = None,
max_reconnect_attempts: int = 10,
heartbeat_interval: int = 30
):
self.url = url
self.api_key = api_key
self.on_message = on_message
self.on_error = on_error or (lambda e: print(f"WebSocket Error: {e}"))
self.max_reconnect_attempts = max_reconnect_attempts
self.heartbeat_interval = heartbeat_interval
self.ws = None
self.running = False
self.reconnect_delay = 1
async def connect(self):
"""Stellt WebSocket-Verbindung her"""
headers = {
"X-Holysheep-Key": self.api_key
}
try:
self.ws = await websockets.connect(self.url, extra_headers=headers)
self.reconnect_delay = 1 # Reset bei erfolgreicher Verbindung
print("✓ WebSocket verbunden")
return True
except Exception as e:
print(f"✗ Verbindungsfehler: {e}")
return False
async