Stellen Sie sich vor: Es ist 3:47 Uhr morgens, der Bitcoin-Kurs zeigt eine explosive Bewegung, und Ihr Trading-Bot verliert just in diesem kritischen Moment die Verbindung zur Börsen-API. Genau das passierte mir während des letzten großen Markt-Crashs, als meine automatisierten Orders ins Leere liefen und ich innerhalb von Minuten über 12.000 Dollar an verpassten Chancen verlor. In diesem Tutorial zeige ich Ihnen, wie Sie eine robuste Reconnect-Strategie implementieren, die solche Szenarien zuverlässig abfängt.
Das Problem: API-Verbindungsabbrüche verstehen
Kryptowährungs-Börsen-APIs sind ausfallsanfälliger als klassische Finanz-APIs. Die Gründe sind vielfältig:
- Rate Limiting: Binance, Coinbase Pro und Kraken begrenzen Anfragen strikt (Binance: 1200 Requests/Minute im Werbe-Modus)
- Netzwerkinstabilität: DDoS-Angriffe auf Börsen sind an der Tagesordnung
- Wartungsfenster: Unangekündigte API-Pausen während hoher Volatilität
- WebSocket-Timeouts: Bei Inaktivität trennen Börsen automatisch Verbindungen
Die Lösung: Exponential Backoff mit Jitter
Die bewährte Strategie kombiniert exponentielle Wartezeiten mit Zufalls-Jitter, um sowohl Server zu entlasten als auch Kollisionen bei gleichzeitigem Reconnect zu vermeiden.
import asyncio
import random
import time
from typing import Callable, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
FAILED = "failed"
@dataclass
class ReconnectConfig:
"""Konfiguration für Reconnect-Strategie"""
base_delay: float = 1.0 # Basis-Verzögerung in Sekunden
max_delay: float = 60.0 # Maximale Verzögerung
max_retries: int = 10 # Maximale Wiederholungen
jitter_factor: float = 0.3 # Zufalls-Jitter (30%)
backoff_multiplier: float = 2.0 # Exponentieller Faktor
@dataclass
class ReconnectState:
"""Zustand des Reconnect-Mechanismus"""
current_attempt: int = 0
state: ConnectionState = ConnectionState.DISCONNECTED
last_error: Optional[str] = None
consecutive_failures: int = 0
total_reconnects: int = 0
total_failures: int = 0
connection_history: list = field(default_factory=list)
class CryptoAPIReconnect:
"""
Robuster Reconnect-Manager für Kryptowährungs-Börsen-APIs.
Implementiert Exponential Backoff mit Jitter.
"""
def __init__(
self,
config: Optional[ReconnectConfig] = None,
on_reconnect: Optional[Callable] = None,
on_failure: Optional[Callable] = None
):
self.config = config or ReconnectConfig()
self.state = ReconnectState()
self.on_reconnect = on_reconnect
self.on_failure = on_failure
self._running = False
self._lock = asyncio.Lock()
def calculate_delay(self, attempt: int) -> float:
"""
Berechnet Verzögerung mit Exponential Backoff und Jitter.
Formel: delay = min(max_delay, base_delay * (multiplier ^ attempt)) * (1 + jitter)
"""
# Exponentielle Berechnung
exponential_delay = self.config.base_delay * (
self.config.backoff_multiplier ** attempt
)
# Begrenzen auf Maximum
capped_delay = min(exponential_delay, self.config.max_delay)
# Jitter hinzufügen für Kollisionsvermeidung
jitter = 1.0 + random.uniform(
-self.config.jitter_factor,
self.config.jitter_factor
)
final_delay = capped_delay * jitter
print(f"[Reconnect] Attempt {attempt + 1}: Warte {final_delay:.2f}s")
return final_delay
async def connect_with_retry(
self,
connect_func: Callable,
*args,
**kwargs
) -> bool:
"""
Stellt Verbindung her mit automatischer Retry-Logik.
Args:
connect_func: Asynchrone Funktion für Verbindung
*args, **kwargs: Argumente für connect_func
Returns:
True bei erfolgreicher Verbindung
"""
async with self._lock:
self._running = True
self.state.state = ConnectionState.CONNECTING
last_exception = None
for attempt in range(self.config.max_retries):
if not self._running:
print("[Reconnect] Verbindung abgebrochen")
return False
try:
self.state.current_attempt = attempt
self.state.state = ConnectionState.CONNECTING
# Verbindung herstellen
result = await connect_func(*args, **kwargs)
# Erfolg
self.state.state = ConnectionState.CONNECTED
self.state.consecutive_failures = 0
self.state.connection_history.append({
"timestamp": time.time(),
"attempt": attempt,
"success": True
})
print(f"[Reconnect] ✓ Verbindung hergestellt nach {attempt + 1} Versuchen")
if self.on_reconnect:
await self.on_reconnect(result)
return True
except Exception as e:
last_exception = e
self.state.consecutive_failures += 1
self.state.total_failures += 1
self.state.last_error = str(e)
print(f"[Reconnect] ✗ Versuch {attempt + 1} fehlgeschlagen: {e}")
# Bei Erschöpfung der Versuche
if attempt < self.config.max_retries - 1:
delay = self.calculate_delay(attempt)
self.state.state = ConnectionState.RECONNECTING
await asyncio.sleep(delay)
else:
self.state.state = ConnectionState.FAILED
print(f"[Reconnect] ✗ Alle {self.config.max_retries} Versuche fehlgeschlagen")
# Finale Fehlerbehandlung
if self.on_failure and last_exception:
await self.on_failure(last_exception)
return False
async def auto_reconnect_loop(
self,
check_connection_func: Callable,
connect_func: Callable,
check_interval: float = 30.0
):
"""
Endlos-Schleife für automatische Verbindungspflege.
Prüft periodisch Verbindung und reconnectet bei Bedarf.
"""
print(f"[Reconnect] Starte Auto-Reconnect mit {check_interval}s Intervall")
while self._running:
try:
# Verbindung prüfen
is_connected = await check_connection_func()
if not is_connected:
print("[Reconnect] Verbindung verloren, reconnecte...")
self.state.state = ConnectionState.RECONNECTING
success = await self.connect_with_retry(connect_func)
if success:
self.state.total_reconnects += 1
else:
# Warte länger nach komplettem Fehlschlag
await asyncio.sleep(self.config.max_delay)
else:
print("[Reconnect] ✓ Verbindung aktiv")
await asyncio.sleep(check_interval)
except asyncio.CancelledError:
print("[Reconnect] Auto-Reconnect gestoppt")
break
except Exception as e:
print(f"[Reconnect] Fehler im Loop: {e}")
await asyncio.sleep(5)
self.state.state = ConnectionState.DISCONNECTED
def disconnect(self):
"""Trennt Verbindung und stoppt Auto-Reconnect"""
self._running = False
self.state.state = ConnectionState.DISCONNECTED
print("[Reconnect] Verbindung getrennt")
def get_stats(self) -> dict:
"""Gibt Statistiken zurück"""
return {
"state": self.state.state.value,
"current_attempt": self.state.current_attempt,
"total_reconnects": self.state.total_reconnects,
"total_failures": self.state.total_failures,
"consecutive_failures": self.state.consecutive_failures,
"success_rate": (
self.state.total_reconnects /
max(1, self.state.total_reconnects + self.state.total_failures)
) * 100
}
Beispiel: Binance WebSocket Integration
async def binance_connect():
"""Beispiel: Binance WebSocket Verbindung"""
import websockets
uri = "wss://stream.binance.com:9443/ws/btcusdt@trade"
try:
async with websockets.connect(uri) as ws:
print("[Binance] WebSocket verbunden")
async for message in ws:
# Hier Nachrichten verarbeiten
pass
except websockets.ConnectionClosed:
print("[Binance] Verbindung geschlossen")
raise
async def check_binance_connection():
"""Prüft ob Binance-Verbindung aktiv ist"""
# Implementierung abhängig von verwendeter Bibliothek
return True
Usage Example
async def main():
reconnect = CryptoAPIReconnect(
config=ReconnectConfig(
base_delay=2.0,
max_delay=120.0,
max_retries=15
)
)
# Manueller Verbindungsversuch mit Retry
success = await reconnect.connect_with_retry(binance_connect)
if success:
# Auto-Reconnect starten
await reconnect.auto_reconnect_loop(
check_binance_connection,
binance_connect,
check_interval=30.0
)
# Statistiken abrufen
stats = reconnect.get_stats()
print(f"Stats: {stats}")
if __name__ == "__main__":
asyncio.run(main())
WebSocket-Spezifische Strategien
Für Echtzeit-Marktdaten sind WebSocket-Verbindungen unerlässlich, aber sie bringen zusätzliche Herausforderungen mit sich:
import asyncio
import json
import websockets
from typing import Dict, Set, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class WebSocketConfig:
"""WebSocket-spezifische Konfiguration"""
ping_interval: int = 30 # Ping alle 30 Sekunden
ping_timeout: int = 10 # Timeout für Pong
reconnect_on_close: bool = True
heartbeat_enabled: bool = True
subscription_resubscribe: bool = True # Erneut subscribed nach Reconnect
class WebSocketReconnectManager:
"""
Spezialisierter Manager für WebSocket-Verbindungen zu Krypto-Börsen.
Behandelt Subscription-Verwaltung und Heartbeat.
"""
def __init__(self, config: Optional[WebSocketConfig] = None):
self.config = config or WebSocketConfig()
self.websocket = None
self.subscriptions: Set[str] = set()
self.subscription_ids: Dict[str, int] = {}
self._next_sub_id = 1
self._last_pong: Optional[float] = None
self._running = False
self._heartbeat_task: Optional[asyncio.Task] = None
self._listener_task: Optional[asyncio.Task] = None
def _generate_subscription_id(self) -> int:
"""Generiert eindeutige Subscription-ID"""
sub_id = self._next_sub_id
self._next_sub_id += 1
return sub_id
async def subscribe(
self,
symbol: str,
channel: str = "trade"
) -> str:
"""
Erstellt Subscription für einen Stream.
Returns: subscription_id
"""
sub_id = self._generate_subscription_id()
stream_name = f"{symbol.lower()}@{channel}"
self.subscriptions.add(stream_name)
self.subscription_ids[stream_name] = sub_id
if self.websocket and self.websocket.open:
await self._send_subscription(stream_name)
print(f"[WS] Subscribed: {stream_name} (ID: {sub_id})")
return f"{stream_name}@{sub_id}"
async def _send_subscription(self, stream: str):
"""Sendet Subscription-Nachricht an Börse"""
subscribe_msg = {
"method": "SUBSCRIBE",
"params": [stream],
"id": self.subscription_ids.get(stream, self._generate_subscription_id())
}
await self.websocket.send(json.dumps(subscribe_msg))
async def unsubscribe(self, stream: str):
"""Entfernt Subscription"""
if stream in self.subscriptions:
self.subscriptions.remove(stream)
if self.websocket and self.websocket.open:
unsubscribe_msg = {
"method": "UNSUBSCRIBE",
"params": [stream],
"id": self.subscription_ids.get(stream, 1)
}
await self.websocket.send(json.dumps(unsubscribe_msg))
print(f"[WS] Unsubscribed: {stream}")
async def connect(
self,
uri: str,
on_message: Optional[callable] = None,
on_error: Optional[callable] = None
) -> bool:
"""
Stellt WebSocket-Verbindung her mit automatischer Subscription.
"""
max_retries = 10
base_delay = 1.0
max_delay = 60.0
for attempt in range(max_retries):
try:
print(f"[WS] Verbindungsversuch {attempt + 1}/{max_retries}")
self.websocket = await websockets.connect(
uri,
ping_interval=self.config.ping_interval,
ping_timeout=self.config.ping_timeout
)
# Erneut subscribed setzen
if self.config.subscription_resubscribe:
for stream in self.subscriptions.copy():
await self._send_subscription(stream)
await asyncio.sleep(0.1) # Rate Limiting
self._running = True
self._last_pong = time.time()
# Heartbeat starten
if self.config.heartbeat_enabled:
self._heartbeat_task = asyncio.create_task(
self._heartbeat_loop()
)
# Listener starten
self._listener_task = asyncio.create_task(
self._message_listener(on_message, on_error)
)
print("[WS] ✓ Verbindung hergestellt")
return True
except websockets.ConnectionClosed as e:
print(f"[WS] Verbindung geschlossen: {e}")
await self._handle_disconnect(uri, on_message, on_error, attempt, max_retries, base_delay, max_delay)
except Exception as e:
print(f"[WS] Fehler: {e}")
await self._handle_disconnect(uri, on_message, on_error, attempt, max_retries, base_delay, max_delay)
print("[WS] ✗ Max. Versuche erreicht")
return False
async def _handle_disconnect(
self, uri, on_message, on_error, attempt, max_retries, base_delay, max_delay
):
"""Behandelt Trennung mit Exponential Backoff"""
if attempt < max_retries - 1:
# Exponential Backoff mit Jitter
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = delay * 0.3 * (2 * asyncio.get_event_loop().time() % 1 - 1)
wait_time = delay + jitter
print(f"[WS] Warte {wait_time:.1f}s vor nächstem Versuch...")
await asyncio.sleep(wait_time)
async def _heartbeat_loop(self):
"""Sendet periodisch Ping und prüft Pong"""
while self._running:
try:
await asyncio.sleep(self.config.ping_interval)
if not self.websocket or not self.websocket.open:
print("[WS] Heartbeat: Socket geschlossen")
break
# Prüfe ob Pong erhalten
if self._last_pong:
time_since_pong = time.time() - self._last_pong
if time_since_pong > self.config.ping_interval + self.config.ping_timeout:
print(f"[WS] Heartbeat: Kein Pong seit {time_since_pong:.1f}s")
await self.websocket.close()
break
except asyncio.CancelledError:
break
except Exception as e:
print(f"[WS] Heartbeat Fehler: {e}")
break
async def _message_listener(
self,
on_message: Optional[callable],
on_error: Optional[callable]
):
"""Lauscht auf eingehende Nachrichten"""
try:
async for message in self.websocket:
try:
data = json.loads(message)
# Pong bestätigen
if data.get("pong") or data.get("result") is None:
self._last_pong = time.time()
# Benutzerdefinierter Handler
if on_message:
await on_message(data)
except json.JSONDecodeError:
print(f"[WS] Ungültiges JSON: {message}")
except websockets.ConnectionClosed:
print("[WS] Verbindung geschlossen im Listener")
except Exception as e:
if on_error:
await on_error(e)
print(f"[WS] Listener Fehler: {e}")
finally:
self._running = False
async def send(self, message: dict):
"""Sendet Nachricht über WebSocket"""
if self.websocket and self.websocket.open:
await self.websocket.send(json.dumps(message))
else:
raise ConnectionError("WebSocket nicht verbunden")
async def close(self):
"""Schließt WebSocket-Verbindung sauber"""
print("[WS] Schließe Verbindung...")
self._running = False
if self._heartbeat_task:
self._heartbeat_task.cancel()
try:
await self._heartbeat_task
except asyncio.CancelledError:
pass
if self._listener_task:
self._listener_task.cancel()
try:
await self._listener_task
except asyncio.CancelledError:
pass
if self.websocket:
await self.websocket.close()
print("[WS] Verbindung geschlossen")
Beispiel: Binance kombinierte Trade-Daten
async def on_binance_message(data: dict):
"""Behandelt eingehende Binance-Nachrichten"""
if "e" in data: # Event vorhanden
event_type = data["e"]
if event_type == "trade":
print(f"Trade: {data['s']} @ {data['p']} (Qty: {data['q']})")
# Integration mit HolySheep AI für Sentiment-Analyse
# Siehe Integration-Beispiel unten
elif event_type == "depthUpdate":
print(f"Orderbook Update: {data['s']}")
elif event_type == "24hrMiniTicker":
print(f"24h Stats: {data['s']} - O:{data['o']} H:{data['h']} L:{data['l']} C:{data['c']}")
async def main_websocket():
"""Beispiel: Binance WebSocket mit Auto-Reconnect"""
manager = WebSocketReconnectManager(
config=WebSocketConfig(
ping_interval=30,
heartbeat_enabled=True,
subscription_resubscribe=True
)
)
# Mehrere Streams subscriben
await manager.subscribe("btcusdt", "trade")
await manager.subscribe("btcusdt", "depth@100ms")
await manager.subscribe("btcusdt", "miniTicker")
# Verbinden
uri = "wss://stream.binance.com:9443/ws"
success = await manager.connect(uri, on_message=on_binance_message)
if success:
try:
# Verbindung halten
while manager._running:
await asyncio.sleep(10)
except asyncio.CancelledError:
pass
finally:
await manager.close()
if __name__ == "__main__":
asyncio.run(main_websocket())
Integration mit HolySheep AI für Trading-Analyse
Nach dem Reconnect und dem Empfang von Marktdaten können Sie HolySheep AI für Echtzeit-Sentiment-Analyse und Trading-Entscheidungen nutzen:
import aiohttp
import asyncio
from typing import List, Dict, Optional
class HolySheepTradingAnalyzer:
"""
Integration mit HolySheep AI für Krypto-Trading-Analyse.
Nutzt günstige DeepSeek V3.2 Modelle für effiziente Sentiment-Analyse.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Erstellt oder gibt bestehende Session zurück"""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self._session
async def analyze_market_sentiment(
self,
recent_trades: List[Dict],
symbol: str
) -> Dict:
"""
Analysiert Marktsentiment basierend auf recent Trades.
Nutzt DeepSeek V3.2 ($0.42/MToken) für kostengünstige Analyse.
"""
session = await self._get_session()
# Trade-Daten für Prompt vorbereiten
trade_summary = "\n".join([
f"- {t.get('time', 'N/A')}: {t.get('side', 'BUY')} {t.get('qty', 0)} @ {t.get('price', 0)}"
for t in recent_trades[-20:]
])
prompt = f"""Analysiere das Marktsentiment für {symbol} basierend auf diesen letzten Trades:
{trade_summary}
Gib zurück:
1. Sentiment (Bullish/Bearish/Neutral)
2. Kauf-/Verkaufsdruck (0-100)
3. Kurzfristige Prognose (1h)
4. Risikoeinschätzung
Antworte im JSON-Format."""
try:
async with session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "Du bist ein erfahrener Krypto-Trading-Analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
},
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
data = await response.json()
return {
"analysis": data["choices"][0]["message"]["content"],
"model": "deepseek-v3.2",
"cost_estimate": len(prompt) / 4 * 0.00042 # ~$0.42 per MToken
}
else:
error_text = await response.text()
raise Exception(f"API Error {response.status}: {error_text}")
except asyncio.TimeoutError:
return {"error": "Timeout bei HolySheep API", "sentiment": "unknown"}
except Exception as e:
return {"error": str(e), "sentiment": "error"}
async def generate_trading_signal(
self,
price: float,
volume: float,
symbol: str,
timeframe: str = "1h"
) -> Dict:
"""
Generiert Trading-Signal basierend auf Preisdaten.
Nutzt Gemini 2.5 Flash ($2.50/MToken) für schnellere Analyse.
"""
session = await self._get_session()
prompt = f"""Generiere ein Trading-Signal für {symbol}:
- Aktueller Preis: ${price}
- Volumen (letzte Stunde): {volume}
- Timeframe: {timeframe}
Analysiere und gib zurück:
{{
"action": "BUY/SELL/HOLD",
"confidence": 0-100,
"entry_price": number,
"stop_loss": number,
"take_profit": number,
"reasoning": "..."
}}"""
try:
async with session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "gemini-2.5-flash",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 300,
"response_format": {"type": "json_object"}
},
timeout=aiohttp.ClientTimeout(total=5)
) as response:
if response.status == 200:
data = await response.json()
return data["choices"][0]["message"]["content"]
else:
return {"action": "HOLD", "error": f"API {response.status}"}
except Exception as e:
return {"action": "HOLD", "error": str(e)}
async def close(self):
"""Schließt HTTP-Session"""
if self._session and not self._session.closed:
await self._session.close()
Kombinierte Nutzung mit Reconnect-Manager
async def integrated_trading_system():
"""
Vollständiges Beispiel: Binance -> Reconnect -> HolySheep Analyse
"""
import websockets
import json
# Konfiguration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Hier Ihren Key einsetzen
analyzer = HolySheepTradingAnalyzer(HOLYSHEEP_API_KEY)
recent_trades = []
ws_manager = WebSocketReconnectManager()
async def on_trade(data: dict):
"""Behandelt eingehende Trades"""
if data.get("e") == "trade":
trade = {
"time": data["T"],
"symbol": data["s"],
"price": float(data["p"]),
"qty": float(data["q"]),
"side": "BUY" if data["m"] else "SELL"
}
recent_trades.append(trade)
# Nur letzte 50 Trades behalten
if len(recent_trades) > 50:
recent_trades = recent_trades[-50:]
# Alle 10 Trades analysieren
if len(recent_trades) % 10 == 0:
analysis = await analyzer.analyze_market_sentiment(
recent_trades,
data["s"]
)
if "error" not in analysis:
print(f"[Analyse] {analysis.get('model')}")
print(f"[Kosten] ~${analysis.get('cost_estimate', 0):.4f}")
try:
# WebSocket verbinden
await ws_manager.subscribe("btcusdt", "trade")
success = await ws_manager.connect(
"wss://stream.binance.com:9443/ws",
on_message=on_trade
)
if success:
print("[System] Trading-System aktiv")
# Endlos-Schleife
while True:
await asyncio.sleep(60)
except KeyboardInterrupt:
print("\n[System] Stoppe Trading-System...")
finally:
await ws_manager.close()
await analyzer.close()
if __name__ == "__main__":
asyncio.run(integrated_trading_system())
Häufige Fehler und Lösungen
1. Endlosschleife bei Serverausfällen
Problem: Der Reconnect-Loop startet endlos neu, ohne Pause, was zu IP-Bannen durch Rate Limiting führt.
Lösung: Implementieren Sie eine maximale Wartezeit und progressive Backoff:
# FEHLERHAFT - Endlosschleife
async def bad_reconnect():
while True:
try:
await connect()
except:
await asyncio.sleep(1) # Immer nur 1 Sekunde!
# → Serverban nach ~100 Versuchen
KORREKT - Progressiver Backoff mit Limit
async def good_reconnect():
max_total_wait = 3600 # Max 1 Stunde warten
for attempt in range(100):
try:
await connect()
break
except Exception as e:
# Progressiv länger warten: 1s, 2s, 4s, 8s... max 5min
delay = min(2 ** attempt, 300) + random.uniform(0, 1)
print(f"Warte {delay:.1f}s (Attempt {attempt + 1})")
await asyncio.sleep(delay)
total_waited = sum([min(2**i, 300) for i in range(attempt + 1)])
if total_waited > max_total_wait:
print("Maximale Wartezeit erreicht, stoppe.")
raise Exception("Reconnect fehlgeschlagen nach 1h")
2. Subscription-Datenverlust nach Reconnect
Problem: Nach Reconnect sind alle WebSocket-Subscriptions verloren, aber der Code merkt es nicht.
Lösung: Automatisches Resubscribing implementieren:
# FEHLERHAFT - Subscriptions bleiben leer
class BadWebSocket:
def __init__(self):
self.ws = None
async def connect(self):
self.ws = await websockets.connect(URI)
# Fehler: subscriptions werden nicht restored!
async def subscribe(self, stream):
# Subscription nur lokal gespeichert
self.subscriptions.add(stream)
# Wird nach reconnect NICHT gesendet!
KORREKT - Subscription-Tracking und Restore
class GoodWebSocket:
def __init__(self):
self.ws = None
self._subscriptions: Dict[str, str] = {} # stream -> sub_id
async def subscribe(self, stream: str) -> str:
sub_id = str(uuid.uuid4())
self._subscriptions[stream] = sub_id
if self.ws and self.ws.open:
await self._send_subscribe(stream, sub_id)
return sub_id
async def _send_subscribe(self, stream: str, sub_id: str):
await self.ws.send(json.dumps({
"method": "SUBSCRIBE",
"params": [stream],
"id": sub_id
}))
async def connect(self):
self.ws = await websockets.connect(URI)
# WICHTIG: Alle existierenden Subscriptions wiederherstellen
for stream, sub_id in self._subscriptions.items():
await self._send_subscribe(stream, sub_id)
await asyncio.sleep(0.05) # Rate Limiting beachten
3. Race Conditions bei parallelen Reconnects
Problem: Mehrere Instanzen reconnecten gleichzeitig, verstärken Serverlast und werden gebannt.
Lösung: Verteiltes Backoff mit Jitter und Locking:
import asyncio
import hashlib
class DistributedReconnect:
def __init__(self, instance_id: str):
self.instance_id = instance_id
self._lock = asyncio.Lock()
self._last_reconnect = 0
async def reconnect_with_coordination(self):
async with self._lock:
now = time.time()
time_since_last = now - self._last_reconnect
# Mindestabstand zwischen Reconnects
min_interval = 5.0
if time_since_last < min_interval:
wait = min_interval - time_since_last
print(f"Warne {wait:.1f}s wegen Koordination")
await asyncio.sleep(wait)
# Jitter basierend auf Instance-ID für Verteilung
jitter = int(hashlib.md5(self.instance_id.encode()).hexdigest()[:8], 16)
jitter_seconds = (jitter % 100) / 100 * 5 # 0-5 Sekunden
await asyncio.sleep(jitter_seconds)
try:
await self._do_reconnect()
self._last_reconnect = time.time()
except Exception as e:
print(f"Reconnect fehlgeschlagen: {e}")
raise
async def _do_reconnect(self):
# Eigentliche Reconnect-Logik
pass
Monitoring und Alerting
Ein robustes Reconnect-System benötigt Monitoring:
from dataclasses import dataclass
from typing import List
from datetime import datetime, timedelta
import json
@dataclass
class ConnectionMetrics:
total_connections: int = 0
total_disconnections: int = 0
total_reconnects: int = 0
failed_reconnects: int = 0
avg_reconnect_time: float = 0.0
uptime_percentage: float = 100.0
last_connection_time: Optional[datetime] = None
last_disconnection_time: Optional[datetime] = None
class ReconnectMonitor:
"""Überwacht Reconnect-Performance und löst Alerts aus"""
def __init__(self, alert_threshold: dict = None):
self.metrics = ConnectionMetrics()
self.events: List[dict] = []
self.alert_threshold = alert_threshold or {
"max_failures_per_hour": 10,
"max_reconnect_time": 300, # 5 Minuten
"min_uptime_percentage": 99.0
}
self._alerts: List[str] = []
def record_connection(self, success: bool, duration: float = 0):
"""Zeichnet Verbindungsversuch auf"""
self.metrics.total_connections += 1
self.metrics.last_connection_time = datetime.now()
if success:
self.events.append({
"type": "connection",
"success": True,
"timestamp": datetime.now().isoformat()
})
else:
self.events.append({
"type": "connection",
"success": False,
"duration": duration,
"timestamp": datetime.now().isoformat()
})
self._check_alerts()