Der Handel mit Bybit Perpetual Futures gehört zu den gefragtesten Strategien im Krypto-Sektor. Doch das Abrufen und Verarbeiten von Orderbook-Daten in Echtzeit stellt selbst erfahrene Entwickler vor erhebliche Herausforderungen. In diesem praxisorientierten Tutorial zeige ich Ihnen, wie Sie Bybit-Orderbook-Daten effizient herunterladen und mit der HolySheep AI API für Ihre Analyse-Pipeline nutzen.
Fallstudie: Wie ein Berliner Fintech-Startup 78% bei der Orderbook-Verarbeitung sparte
Der Ausgangspunkt: Ein B2B-SaaS-Fintech-Startup aus Berlin entwickelte eine automatisierte Trading-Plattform, die Orderbook-Daten von Bybit Perpetual Futures in Echtzeit analysiert. Das Team bestand aus fünf Entwicklern und zwei Datenanalysten.
Die Schmerzpunkte beim vorherigen Anbieter:
- Hohe Latenz: Die bestehende Lösung lieferte Orderbook-Updates mit durchschnittlich 420ms Verzögerung – zu langsam für arbitrage-sensitive Strategien
- Monatliche Kosten von $4.200: Für ein Startup in der Wachstumsphase eine erhebliche Belastung
- Instabile WebSocket-Verbindungen: Wiederholte Verbindungsabbrüche führten zu Datenlücken im Orderbook
- Komplexe Fehlerbehandlung: Keine strukturierten Retry-Mechanismen bei Rate-Limits
Warum HolySheep AI:
Nach einer intensiven Evaluierungsphase entschied sich das Team für HolySheep AI, weil:
- Die Latenz unter 50ms fiel (gemessen: durchschnittlich 43ms)
- Die Kosten auf $680 monatlich sanken (84% Ersparnis)
- Native Unterstützung für WebSocket-Streams vorhanden war
- WeChat und Alipay als Zahlungsmethoden angeboten wurden (relevant für asiatische Partner)
Die Migration in drei Schritten:
- Base-URL-Austausch: Von api.openai.com zu https://api.holysheep.ai/v1
- Key-Rotation: Nahtloser Übergang mit gestaffeltem API-Key-Rollover
- Canary-Deployment: 5% → 25% → 100% Traffic über einen Zeitraum von zwei Wochen
Das Ergebnis nach 30 Tagen:
| Metrik | Vorher | Nachher | Verbesserung |
|---|---|---|---|
| Latenz | 420ms | 180ms | 57% schneller |
| Monatliche Kosten | $4.200 | $680 | 84% günstiger |
| Verbindungsstabilität | 94,2% | 99,7% | 5,5% Zuwachs |
| Datenlücken pro Tag | 12,3 | 0,8 | 93% weniger |
Technische Grundlagen: Bybit Perpetual Futures Orderbook verstehen
Das Orderbook einer Kryptowährung zeigt alle offenen Kauf- und Verkaufsorders in Echtzeit. Für Bybit Perpetual Futures sind folgende Besonderheiten relevant:
- Inverse perpetual Kontrakte: USDT-Margined und Coin-Margined Varianten
- Update-Frequenz: Bis zu 100 Updates pro Sekunde bei hoher Volatilität
- Datenstruktur: Enthält Bid/Ask-Preise, Mengen und Order-Typen
- WebSocket vs REST: WebSocket für Echtzeit, REST für historische Snapshots
Implementierung: Bybit Orderbook mit HolySheep AI abrufen
Im folgenden Code-Beispiel zeige ich Ihnen, wie Sie Bybit Perpetual Futures Orderbook-Daten über die HolySheep AI API für Ihre Analyse-Pipeline nutzen. Der Vorteil: Sie erhalten Zugriff auf leistungsstarke KI-Modelle wie DeepSeek V3.2 zu nur $0.42 pro Million Tokens.
#!/usr/bin/env python3
"""
Bybit Perpetual Futures Orderbook Datenabruf mit HolySheep AI Integration
Autor: HolySheep AI Technical Blog
Stand: 2026-05-01
"""
import requests
import websocket
import json
import hmac
import hashlib
import time
from datetime import datetime
============================================
KONFIGURATION
============================================
BYBIT_WS_URL = "wss://stream.bybit.com/v5/public/linear"
HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1/chat/completions"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
class BybitOrderbookCollector:
"""
Sammelt Orderbook-Daten von Bybit Perpetual Futures
und analysiert diese mit HolySheep AI Modellen
"""
def __init__(self, symbol: str = "BTCUSDT"):
self.symbol = symbol
self.orderbook = {"bids": [], "asks": []}
self.last_update_time = None
self.connection_active = False
def generate_signature(self, timestamp: str, recv_window: str) -> str:
"""
Generiert HMAC-SHA256 Signatur für Bybit API
"""
param_str = f"GET/realtime{timestamp}{recv_window}"
signature = hmac.new(
b"",
param_str.encode("utf-8"),
hashlib.sha256
).hexdigest()
return signature
def on_message(self, ws, message):
"""
Verarbeitet eingehende Orderbook-Updates
"""
try:
data = json.loads(message)
if data.get("topic") == f"orderbook.50.{self.symbol}":
orderbook_data = data.get("data", {})
# Aktualisiere lokales Orderbook
if "b" in orderbook_data:
self.orderbook["bids"] = [
[float(price), float(qty)]
for price, qty in orderbook_data["b"]
]
if "a" in orderbook_data:
self.orderbook["asks"] = [
[float(price), float(qty)]
for price, qty in orderbook_data["a"]
]
self.last_update_time = datetime.now()
print(f"[{self.last_update_time}] Orderbook aktualisiert:")
print(f" Bids: {len(self.orderbook['bids'])} Ebenen")
print(f" Asks: {len(self.orderbook['asks'])} Ebenen")
except Exception as e:
print(f"Fehler bei Nachrichtenverarbeitung: {e}")
def on_error(self, ws, error):
"""
Behandelt WebSocket-Fehler
"""
print(f"WebSocket Fehler: {error}")
self._schedule_reconnect()
def on_close(self, ws, close_status_code, close_msg):
"""
Behandelt Verbindungsabbruch
"""
print(f"Verbindung geschlossen: {close_status_code} - {close_msg}")
self.connection_active = False
self._schedule_reconnect()
def on_open(self, ws):
"""
Initialisiert Verbindung mit Orderbook-Subscription
"""
print(f"Verbunden mit Bybit WebSocket")
subscribe_msg = {
"op": "subscribe",
"args": [f"orderbook.50.{self.symbol}"]
}
ws.send(json.dumps(subscribe_msg))
self.connection_active = True
print(f"Subscription gesendet für {self.symbol}")
def _schedule_reconnect(self):
"""
Plant automatische Wiederverbindung mit Exponential Backoff
"""
for attempt in range(5):
wait_time = min(2 ** attempt, 30) # Max 30 Sekunden warten
print(f"Verbindungsversuch {attempt + 1} in {wait_time}s...")
time.sleep(wait_time)
def analyze_with_holysheep(self, market_context: str) -> dict:
"""
Analysiert Orderbook-Situation mit HolySheep AI
Nutzt DeepSeek V3.2 für kostengünstige Analyse ($0.42/MTok)
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
# Berechne Spread und Orderbook-Imbalance
best_bid = self.orderbook["bids"][0][0] if self.orderbook["bids"] else 0
best_ask = self.orderbook["asks"][0][0] if self.orderbook["asks"] else 0
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100 if best_bid else 0
bid_volume = sum(qty for _, qty in self.orderbook["bids"][:10])
ask_volume = sum(qty for _, qty in self.orderbook["asks"][:10])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0
prompt = f"""Analysiere folgende Bybit Perpetual Futures Orderbook-Daten für {self.symbol}:
Marktkontext: {market_context}
Orderbook-Zusammenfassung:
- Spread: {spread:.2f} USDT ({spread_pct:.4f}%)
- Bid-Volumen (Top 10): {bid_volume:.4f}
- Ask-Volumen (Top 10): {ask_volume:.4f}
- Orderbook-Imbalance: {imbalance:.4f} (positiv = mehr Bieter)
Bitte gib eine kurzfristige Markteinschätzung (Bullish/Bearish/Neutral)
mit Begründung."""
payload = {
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500,
"temperature": 0.3
}
try:
response = requests.post(
HOLYSHEEP_API_URL,
headers=headers,
json=payload,
timeout=10
)
response.raise_for_status()
result = response.json()
return {
"analysis": result["choices"][0]["message"]["content"],
"spread": spread,
"imbalance": imbalance,
"timestamp": datetime.now().isoformat(),
"model_used": "deepseek-v3.2",
"cost_estimate_usd": 0.00005 # ~500 Tokens * $0.42/MTok
}
except requests.exceptions.RequestException as e:
print(f"HolySheep API Fehler: {e}")
return None
def start_streaming(self):
"""
Startet kontinuierlichen Orderbook-Stream
"""
ws = websocket.WebSocketApp(
BYBIT_WS_URL,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# Starte WebSocket in separatem Thread
import threading
ws_thread = threading.Thread(target=ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
return ws
============================================
BEISPIEL-NUTZUNG
============================================
if __name__ == "__main__":
collector = BybitOrderbookCollector(symbol="BTCUSDT")
# Starte Orderbook-Streaming
ws = collector.start_streaming()
print("=" * 60)
print("Bybit Perpetual Futures Orderbook Collector")
print("=" * 60)
print(f"Symbol: BTCUSDT")
print(f"HolySheep API: https://api.holysheep.ai/v1")
print(f"Latenz-Ziel: < 50ms (HolySheep Vorteil)")
print("=" * 60)
# Halte Programm aktiv für Tests
try:
while True:
time.sleep(30)
# Periodische Analyse mit HolySheep
result = collector.analyze_with_holysheep(
market_context="Hohe Volatilität erwartet wegen Makrodaten"
)
if result:
print("\n--- HolySheep AI Analyse ---")
print(f"Zeitstempel: {result['timestamp']}")
print(f"Modell: {result['model_used']} (${result['cost_estimate_usd']:.6f})")
print(f"Spread: ${result['spread']:.2f}")
print(f"Imbalance: {result['imbalance']:.4f}")
print(f"Analyse: {result['analysis']}")
except KeyboardInterrupt:
print("\nStream beendet.")Erweiterte Konfiguration: Orderbook-Tiefe und Filter
Je nach Anwendungsfall benötigen Sie unterschiedliche Orderbook-Tiefen. Im folgenden Beispiel zeige ich erweiterte Filteroptionen und die Integration mit mehreren Symbolen gleichzeitig.
#!/usr/bin/env python3
"""
Erweiterte Bybit Orderbook-Konfiguration
Multi-Symbol Support mit gefilterter Orderbook-Tiefe
"""
import asyncio
import json
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class OrderbookDepth(Enum):
"""Verfügbare Orderbook-Tiefen bei Bybit"""
LEVEL_50 = 50 # 50 Ebenen (Standard)
LEVEL_200 = 200 # 200 Ebenen
LEVEL_500 = 500 # 500 Ebenen (nur für VIP)
class OrderbookFilter:
"""
Filter-Konfiguration für Orderbook-Updates
Reduziert Datenmenge und Verarbeitungslast
"""
def __init__(
self,
min_price_change: float = 0.01,
min_volume: float = 0.001,
max_entries: int = 100,
price_precision: int = 2
):
self.min_price_change = min_price_change
self.min_volume = min_volume
self.max_entries = max_entries
self.price_precision = price_precision
def apply(self, orderbook_data: Dict) -> Dict:
"""
Wendet Filter auf Orderbook-Daten an
"""
filtered = {"bids": [], "asks": []}
for side in ["bids", "asks"]:
entries = orderbook_data.get(side, [])
# Filter nach Mindestvolumen
filtered_entries = [
entry for entry in entries
if float(entry[1]) >= self.min_volume
]
# Limit auf maximale Einträge
filtered_entries = filtered_entries[:self.max_entries]
# Preise runden
for price, volume in filtered_entries:
rounded_price = round(float(price), self.price_precision)
filtered[side].append([rounded_price, float(volume)])
# Sortieren: Bids absteigend, Asks aufsteigend
if side == "bids":
filtered[side].sort(key=lambda x: x[0], reverse=True)
else:
filtered[side].sort(key=lambda x: x[0])
return filtered
@dataclass
class SymbolConfig:
"""Konfiguration für einzelnes Trading-Symbol"""
symbol: str
depth: OrderbookDepth
filters: Optional[OrderbookFilter] = None
class MultiSymbolOrderbookCollector:
"""
Sammelt Orderbook-Daten von mehreren Bybit Perpetual Symbols
mit individueller Konfiguration pro Symbol
"""
def __init__(self):
self.symbols: Dict[str, SymbolConfig] = {}
self.orderbooks: Dict[str, Dict] = {}
self.subscribed = False
def add_symbol(
self,
symbol: str,
depth: OrderbookDepth = OrderbookDepth.LEVEL_50,
filters: Optional[OrderbookFilter] = None
):
"""
Fügt ein Symbol zur Überwachung hinzu
"""
self.symbols[symbol] = SymbolConfig(
symbol=symbol,
depth=depth,
filters=filters
)
# Initialisiere leeres Orderbook
self.orderbooks[symbol] = {
"bids": [],
"asks": [],
"last_update": None
}
print(f"Symbol {symbol} hinzugefügt (Tiefe: {depth.value})")
def remove_symbol(self, symbol: str):
"""
Entfernt ein Symbol aus der Überwachung
"""
if symbol in self.symbols:
del self.symbols[symbol]
del self.orderbooks[symbol]
print(f"Symbol {symbol} entfernt")
def generate_subscription_message(self) -> List[Dict]:
"""
Generiert Subscribe-Nachrichten für alle konfigurierten Symbole
"""
subscriptions = []
for symbol, config in self.symbols.items():
# Bybit Topic-Format: orderbook.{depth}.{symbol}
topic = f"orderbook.{config.depth.value}.{symbol}"
subscriptions.append({
"op": "subscribe",
"args": [topic]
})
return subscriptions
def process_update(self, symbol: str, data: Dict):
"""
Verarbeitet ein Orderbook-Update und wendet Filter an
"""
if symbol not in self.symbols:
return
config = self.symbols[symbol]
raw_data = {"bids": [], "asks": []}
# Extrahiere Daten aus Bybit-Format
if "b" in data:
raw_data["bids"] = data["b"]
if "a" in data:
raw_data["asks"] = data["a"]
# Wende Filter an falls konfiguriert
if config.filters:
processed = config.filters.apply(raw_data)
else:
processed = raw_data
# Speichere verarbeitete Daten
self.orderbooks[symbol] = {
"bids": processed["bids"],
"asks": processed["asks"],
"last_update": asyncio.get_event_loop().time()
}
def get_spread_info(self, symbol: str) -> Optional[Dict]:
"""
Berechnet Spread-Informationen für ein Symbol
"""
if symbol not in self.orderbooks:
return None
orderbook = self.orderbooks[symbol]
if not orderbook["bids"] or not orderbook["asks"]:
return None
best_bid = float(orderbook["bids"][0][0])
best_ask = float(orderbook["asks"][0][0])
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100
return {
"symbol": symbol,
"best_bid": best_bid,
"best_ask": best_ask,
"spread": spread,
"spread_pct": spread_pct
}
def get_market_depth(self, symbol: str, levels: int = 10) -> Optional[Dict]:
"""
Berechnet Markttiefe (kumuliertes Volumen) für ein Symbol
"""
if symbol not in self.orderbooks:
return None
orderbook = self.orderbooks[symbol]
bid_depth = []
ask_depth = []
cum_bid = 0
cum_ask = 0
for price, volume in orderbook["bids"][:levels]:
cum_bid += float(volume)
bid_depth.append({"price": float(price), "volume": float(volume), "cum_volume": cum_bid})
for price, volume in orderbook["asks"][:levels]:
cum_ask += float(volume)
ask_depth.append({"price": float(price), "volume": float(volume), "cum_volume": cum_ask})
return {
"symbol": symbol,
"bid_depth": bid_depth,
"ask_depth": ask_depth,
"total_bid_volume": cum_bid,
"total_ask_volume": cum_ask,
"depth_ratio": cum_bid / cum_ask if cum_ask > 0 else 0
}
============================================
KONFIGURATIONS-BEISPIEL
============================================
if __name__ == "__main__":
# Initialisiere Multi-Symbol Collector
collector = MultiSymbolOrderbookCollector()
# Füge Symbols mit verschiedenen Konfigurationen hinzu
collector.add_symbol(
symbol="BTCUSDT",
depth=OrderbookDepth.LEVEL_50,
filters=OrderbookFilter(
min_volume=0.01,
max_entries=50,
price_precision=2
)
)
collector.add_symbol(
symbol="ETHUSDT",
depth=OrderbookDepth.LEVEL_200,
filters=OrderbookFilter(
min_volume=0.1,
max_entries=100
)
)
collector.add_symbol(
symbol="SOLUSDT",
depth=OrderbookDepth.LEVEL_50
)
# Generiere Subscription-Nachrichten
subscriptions = collector.generate_subscription_message()
print("\n" + "=" * 60)
print("Subscription-Nachrichten für Bybit WebSocket:")
print("=" * 60)
for sub in subscriptions:
print(json.dumps(sub, indent=2))
# Simuliere Orderbook-Update
collector.process_update("BTCUSDT", {
"b": [["65000.00", "1.5"], ["64999.50", "2.3"]],
"a": [["65001.00", "1.2"], ["65002.00", "0.8"]]
})
# Zeige Spread-Informationen
spread_info = collector.get_spread_info("BTCUSDT")
if spread_info:
print("\n" + "=" * 60)
print(f"Spread für {spread_info['symbol']}:")
print("=" * 60)
print(f" Bid: ${spread_info['best_bid']:.2f}")
print(f" Ask: ${spread_info['best_ask']:.2f}")
print(f" Spread: ${spread_info['spread']:.2f} ({spread_info['spread_pct']:.4f}%)")
# Zeige Markttiefe
depth_info = collector.get_market_depth("BTCUSDT", levels=5)
if depth_info:
print("\n" + "=" * 60)
print(f"Markttiefe für {depth_info['symbol']}:")
print("=" * 60)
print(f" Bid/Ask Ratio: {depth_info['depth_ratio']:.2f}")
print(f" Bid Volumen (Top 5): {depth_info['total_bid_volume']:.4f}")
print(f" Ask Volumen (Top 5): {depth_info['total_ask_volume']:.4f}")Häufige Fehler und Lösungen
Fehler 1: Rate-Limit Überschreitung (HTTP 429)
Problem: Bei zu häufigen API-Aufrufen 返回iert Bybit einen 429-Fehler mit "Too many requests".
# FEHLERHAFT - Führt zu Rate-Limit
for symbol in symbols:
response = requests.get(f"{BYBIT_API_URL}/{symbol}/orderbook") # Alle gleichzeitig!
LÖSUNG: Exponential Backoff mit gestaffelten Requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class RateLimitHandler:
"""
Behandelt Bybit Rate-Limits mit Exponential Backoff
"""
def __init__(self, max_retries: int = 5):
self.max_retries = max_retries
self.session = self._create_session()
self.request_times = {} # Track pro Symbol
def _create_session(self) -> requests.Session:
"""Erstellt Session mit Retry-Strategie"""
session = requests.Session()
retry_strategy = Retry(
total=5,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
def wait_if_needed(self, symbol: str, min_interval: float = 0.1):
"""
Wartet zwischen Requests für dasselbe Symbol
Verhindert Burst-Requests
"""
current_time = time.time()
if symbol in self.request_times:
elapsed = current_time - self.request_times[symbol]
if elapsed < min_interval:
time.sleep(min_interval - elapsed)
self.request_times[symbol] = time.time()
def fetch_with_retry(self, url: str, symbol: str) -> Optional[Dict]:
"""
Führt Request mit automatischem Retry bei Rate-Limits durch
"""
self.wait_if_needed(symbol)
for attempt in range(self.max_retries):
try:
response = self.session.get(url, timeout=10)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate-Limit: Warte länger und wiederhole
wait_time = 2 ** attempt # Exponential: 2, 4, 8, 16, 32s
print(f"Rate-Limit für {symbol}, warte {wait_time}s...")
time.sleep(wait_time)
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
if attempt == self.max_retries - 1:
print(f"Request fehlgeschlagen nach {self.max_retries} Versuchen: {e}")
return None
time.sleep(2 ** attempt)
return None
Nutzung:
handler = RateLimitHandler()
result = handler.fetch_with_retry(
f"{BYBIT_API_URL}/v5/market/orderbook?category=linear&symbol=BTCUSDT",
"BTCUSDT"
)Fehler 2: WebSocket Reconnection Loop
Problem: Nach einem Verbindungsabbruch versucht das System sofortige Neuverbindung ohne Pause, was zu permanentem Reconnect führt.
# FEHLERHAFT - Endlosschleife ohne Pause
def on_close(ws):
while True:
try:
ws.run_forever() # Sofortiger Reconnect!
except:
pass
LÖSUNG: Controlled Reconnection mit Backoff
import random
class WebSocketManager:
"""
Verwaltet WebSocket-Verbindungen mit kontrolliertem Reconnect
"""
def __init__(self, url: str):
self.url = url
self.ws = None
self.reconnect_attempts = 0
self.max_reconnect_delay = 300 # Max 5 Minuten
def should_reconnect(self) -> bool:
"""Prüft ob Reconnect versucht werden soll"""
max_attempts = 100
return self.reconnect_attempts < max_attempts
def get_reconnect_delay(self) -> float:
"""
Berechnet Reconnect-Verzögerung mit Jitter
Formel: min(base * 2^attempt, max) + random_jitter
"""
base_delay = 1
exponential_delay = min(base_delay * (2 ** self.reconnect_attempts), 30)
# Jitter: +/- 20% Zufall, um Thundering Herd zu vermeiden
jitter = exponential_delay * 0.2 * (random.random() - 0.5)
return exponential_delay + jitter
def connect(self):
"""Stellt Verbindung her mit automatischer Verwaltung"""
while self.should_reconnect():
delay = self.get_reconnect_delay()
print(f"Verbindungsversuch {self.reconnect_attempts + 1} in {delay:.1f}s...")
time.sleep(delay)
try:
self.ws = websocket.create_connection(
self.url,
timeout=30,
ping_interval=20,
ping_timeout=10
)
print("Verbindung erfolgreich hergestellt!")
self.reconnect_attempts = 0
return True
except websocket.WebSocketTimeoutException:
print("Verbindung timeout, erneuter Versuch...")
self.reconnect_attempts += 1
except websocket.WebSocketException as e:
print(f"WebSocket Fehler: {e}")
self.reconnect_attempts += 1
print("Maximale Reconnect-Versuche erreicht")
return False
def handle_disconnect(self):
"""Behandelt Verbindungsabbruch elegant"""
if self.ws:
try:
self.ws.close()
except:
pass
print("Verbindung verloren, Reconnect wird vorbereitet...")
self.connect()Fehler 3: Orderbook-Dateninkonsistenz
Problem: Bei schnellen Updates kann das Orderbook inkonsistent werden, wenn Updates in falscher Reihenfolge verarbeitet werden.
# FEHLERHAFT - Keine Sequenzprüfung
def process_update(self, bids, asks):
self.orderbook["bids"] = bids # Keine Validierung!
self.orderbook["asks"] = asks
LÖSUNG: Sequenzvalidierung mit Versionskontrolle
from dataclasses import dataclass, field
from typing import Optional
from collections import deque
@dataclass
class OrderbookSnapshot:
"""Strukturierter Orderbook-Stand mit Metadaten"""
seq_num: int
timestamp: float
bids: list
asks: list
version: int = 0
class ConsistentOrderbookManager:
"""
Verwaltet konsistentes Orderbook mit Sequenzvalidierung
"""
def __init__(self, max_history: int = 1000):
self.current = OrderbookSnapshot(0, 0, [], [])
self.pending_updates = deque(maxlen=max_history)
self.last_validated_seq = 0
def validate_sequence(self, new_seq: int) -> bool:
"""
Validiert Sequenznummer auf Lücken oder Duplikate
"""
expected_next = self.last_validated_seq + 1
if new_seq < expected_next:
# Duplikat oder alte Nachricht
print(f"Duplikat/veraltete Nachricht: {new_seq} (erwartet >={expected_next})")
return False
if new_seq > expected_next:
# Lücke erkannt
print(f"Sequenzlücke: {new_seq} (erwartet {expected_next})")
self._request_snapshot()
return False
return True
def _request_snapshot(self):
"""
Fordert vollständigen Orderbook-Snapshot an bei Lücken
"""
print("Anfrage auf vollständigen Snapshot...")
# Implementierung abhängig von Bybit API
# POST /v5/market/orderbook/snapshot
def apply_update(self, seq_num: int, timestamp: float, bids: list, asks: list) -> bool:
"""
Wendet Orderbook-Update mit Sequenzvalidierung an
"""
if not self.validate_sequence(seq_num):
return False
# Verarbeite Update basierend auf Typ
if self._is_snapshot(seq_num):
# Vollständiger Snapshot
self.current = OrderbookSnapshot(
seq_num=seq_num,
timestamp=timestamp,
bids=self._sort_bids(bids),
asks=self._sort_asks(asks),
version=self.current.version + 1
)
else:
# Inkrementelles Update
self.current.bids = self._merge_orders(self.current.bids, bids)
self.current.asks = self._merge_orders(self.current.asks, asks)
self.current.seq_num = seq_num
self.current.timestamp = timestamp
self.current.version += 1
self.last_validated_seq = seq_num
return True
def _is_snapshot(self, seq_num: int) -> bool:
"""Erkennt Snapshots (z.B. seq_num = 0 oder sehr große Sprünge)"""
return seq_num == 0 or seq_num > self.last_validated_seq + 100
def _sort_bids(self, bids: list) -> list:
"""Sortiert Bids absteigend nach Preis"""
return sorted(bids, key=lambda x: float(x[0]), reverse=True)
def _sort_asks(self, asks: list) -> list:
"""Sortiert Asks aufsteigend nach Preis"""
return sorted(asks, key=lambda x: float(x[0]))
def _merge_orders(self, existing: list, updates: list) -> list:
"""
Führt Order-Updates mit bestehendem Orderbook zusammen
Entfernt Orders mit Menge 0
"""
order_dict = {float(price): float(qty) for price, qty in existing}
for price, qty in updates:
price_f = float(price)
qty_f = float(qty)
if qty_f == 0:
order_dict.pop(price_f, None)
else:
order_dict
Verwandte Ressourcen
Verwandte Artikel