Einleitung
Der Zugriff auf Echtzeit-Marktdaten ist für algorithmischen Handel und quantitative Analyse unerlässlich. Die **OKX API** bietet eine der robustesten Order-Book- und Trade-Detail-Schnittstellen im Kryptomarkt. In diesem Tutorial zeige ich Ihnen, wie Sie深度(Die Tiefe) des Order Books und detaillierte Handelsdaten effizient abrufen und für Ihre Trading-Strategien nutzen können.
*Erfahrungsbericht des Autors: Seit über drei Jahren nutze ich aktiv verschiedene Krypto-Börsen-APIs für meine automatisierten Handelssysteme. Die OKX API sticht durch ihre Zuverlässigkeit und Geschwindigkeit hervor – Latenzzeiten von unter 20ms sind hier keine Seltenheit.*
**Kostenanalyse für 10M Token/Monat (2026 Benchmark):**
| Modell | Preis/MTok | 10M Token Kosten | Latenz |
|--------|------------|------------------|--------|
| GPT-4.1 | $8,00 | $80,00 | ~800ms |
| Claude Sonnet 4.5 | $15,00 | $150,00 | ~700ms |
| Gemini 2.5 Flash | $2,50 | $25,00 | ~400ms |
| **DeepSeek V3.2** | **$0,42** | **$4,20** | ~350ms |
Mit **HolySheep AI** profitieren Sie von identischen Modellen zu dramatisch niedrigeren Preisen – bis zu 85% Ersparnis durch den Yuan-Dollar-Kurs von ¥1=$1.
---
Was ist die OKX Market Data API?
Die OKX Open Platform bietet RESTful APIs für den Zugriff auf:
- **K线数据 (K-Line/Candlestick-Daten)**
- **深度数据 (Order Book Depth)**
- **成交明细 (Trade/Transaction Details)**
- **行情数据 (Ticker/Market Data)**
Für Order Book und Trade Details sind insbesondere zwei Endpunkte relevant:
1.
GET /api/v5/market/books-l2 – Order Book mit 400 Stufen
2.
GET /api/v5/market/trades – Letzte 600 Trades
---
Voraussetzungen
Bevor Sie beginnen, benötigen Sie:
- Ein OKX-Konto (Registration unter [OKX](https://www.okx.com/account.html))
- Einen API-Schlüssel (erstellt im OKX Dashboard)
- Python 3.8+ mit
requests-Bibliothek
- Optional: HolySheep AI für die Analyse der Daten
Python-Umgebung einrichten
Virtuelle Umgebung erstellen
python -m venv okx_env
source okx_env/bin/activate # Windows: okx_env\Scripts\activate
Notwendige Pakete installieren
pip install requests websocket-client pandas numpy
---
Order Book (深度) abrufen
Grundlegendes Beispiel
import requests
import time
import hashlib
import hmac
import base64
from typing import Dict, List, Optional
class OKXMarketData:
"""OKX Market Data API Client für Order Book und Trades"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = "", secret_key: str = "", passphrase: str = ""):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
def _get_timestamp(self) -> str:
"""Aktuellen Timestamp im ISO-Format abrufen"""
return requests.get(f"{self.BASE_URL}/api/v5/public/time").json()["data"][0]["ts"]
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""HMAC-SHA256 Signatur für authentifizierte Anfragen"""
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
def get_order_book(self, inst_id: str = "BTC-USDT", sz: int = 400) -> Dict:
"""
Order Book mit Tiefe abrufen
Args:
inst_id: Instrument-ID (z.B. "BTC-USDT")
sz: Anzahl der Preisstufen (max. 400)
Returns:
Dictionary mit bids und asks
"""
endpoint = f"/api/v5/market/books-l2?instId={inst_id}&sz={sz}"
try:
response = requests.get(f"{self.BASE_URL}{endpoint}")
response.raise_for_status()
data = response.json()
if data["code"] == "0":
return data["data"][0]
else:
print(f"API Fehler: {data['msg']}")
return None
except requests.exceptions.RequestException as e:
print(f"Netzwerkfehler: {e}")
return None
def get_order_book_depth_snapshot(self, inst_id: str = "BTC-USDT") -> Dict:
"""
Order Book Depth Snapshot für Trading-Strategien
Zeigt:
- Bids: Kauforders (Preis, Menge, Liquidität)
- Asks: Verkaufsorders (Preis, Menge, Liquidität)
- Spread: Differenz zwischen bestem Bid und Ask
- Mid Price: Mittlerer Preis
"""
order_book = self.get_order_book(inst_id)
if not order_book:
return None
bids = order_book.get("bids", [])
asks = order_book.get("asks", [])
# Verarbeite Daten
best_bid = float(bids[0][0]) if bids else 0
best_ask = float(asks[0][0]) if asks else 0
spread = best_ask - best_bid
mid_price = (best_bid + best_ask) / 2
spread_pct = (spread / mid_price) * 100 if mid_price > 0 else 0
# Berechne kumulierte Tiefe
bid_volume = sum(float(b[1]) for b in bids[:20])
ask_volume = sum(float(a[1]) for a in asks[:20])
return {
"inst_id": inst_id,
"timestamp": order_book.get("ts"),
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": mid_price,
"spread": spread,
"spread_pct": spread_pct,
"bid_volume_20": bid_volume,
"ask_volume_20": ask_volume,
"imbalance": (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0,
"bids": bids[:20], # Top 20 Bids
"asks": asks[:20] # Top 20 Asks
}
Verwendung
client = OKXMarketData()
depth_data = client.get_order_book_depth_snapshot("BTC-USDT")
if depth_data:
print(f"📊 BTC-USDT Order Book Analyse")
print(f" Mid Price: ${depth_data['mid_price']:,.2f}")
print(f" Spread: ${depth_data['spread']:.2f} ({depth_data['spread_pct']:.4f}%)")
print(f" Bid Volume (Top 20): {depth_data['bid_volume_20']:.4f} BTC")
print(f" Ask Volume (Top 20): {depth_data['ask_volume_20']:.4f} BTC")
print(f" Order Imbalance: {depth_data['imbalance']:.2%}")
---
Trade Details (成交明细) abrufen
Implementation für Echtzeit-Trades
import requests
import json
from datetime import datetime
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class Trade:
"""Struktur für einzelne Trades"""
inst_id: str
trade_id: str
price: float
size: float
side: str # buy oder sell
timestamp: int
timestamp_str: str
@classmethod
def from_api_response(cls, data: List) -> 'Trade':
return cls(
inst_id=data[0],
trade_id=data[1],
price=float(data[2]),
size=float(data[3]),
side=data[4],
timestamp=int(data[5]),
timestamp_str=data[8]
)
class OKXTradeData:
"""OKX Trade Data Client"""
BASE_URL = "https://www.okx.com"
def get_recent_trades(self, inst_id: str = "BTC-USDT", limit: int = 100) -> List[Trade]:
"""
Letzte Trades abrufen
Args:
inst_id: Instrument-ID
limit: Anzahl der Trades (max. 600)
Returns:
Liste von Trade-Objekten
"""
endpoint = f"/api/v5/market/trades?instId={inst_id}&limit={limit}"
try:
response = requests.get(f"{self.BASE_URL}{endpoint}", timeout=10)
response.raise_for_status()
data = response.json()
if data["code"] == "0":
trades = [Trade.from_api_response(t) for t in data["data"]]
return trades
else:
print(f"API Fehler: {data['msg']}")
return []
except requests.exceptions.RequestException as e:
print(f"Netzwerkfehler: {e}")
return []
def analyze_trade_flow(self, trades: List[Trade]) -> Dict:
"""
Trade Flow Analyse für Sentiment-Erkennung
Berechnet:
- Buy/Sell Ratio
- Durchschnittliche Trade-Größe
- Volumen-gewichteter Durchschnittspreis (VWAP)
- Große Trades (>100k USDT)
"""
if not trades:
return {}
buy_volume = sum(t.size * t.price for t in trades if t.side == "buy")
sell_volume = sum(t.size * t.price for t in trades if t.side == "sell")
total_volume = buy_volume + sell_volume
buy_count = sum(1 for t in trades if t.side == "buy")
sell_count = sum(1 for t in trades if t.side == "sell")
# VWAP berechnen
vwap = total_volume / sum(t.size for t in trades) if trades else 0
# Große Trades identifizieren (>100k USDT)
large_trades = [
t for t in trades
if t.size * t.price > 100000
]
return {
"total_trades": len(trades),
"buy_count": buy_count,
"sell_count": sell_count,
"buy_ratio": buy_count / len(trades) if trades else 0,
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"total_volume": total_volume,
"volume_imbalance": (buy_volume - sell_volume) / total_volume if total_volume > 0 else 0,
"vwap": vwap,
"large_trades_count": len(large_trades),
"large_trades": large_trades[:10] # Top 10 größte Trades
}
Verwendung
trade_client = OKXTradeData()
trades = trade_client.get_recent_trades("BTC-USDT", limit=200)
if trades:
analysis = trade_client.analyze_trade_flow(trades)
print(f"📈 BTC-USDT Trade Flow Analyse")
print(f" Gesamt Trades: {analysis['total_trades']}")
print(f" Buy/Sell Ratio: {analysis['buy_ratio']:.2%}")
print(f" Volumen: ${analysis['total_volume']:,.2f}")
print(f" VWAP: ${analysis['vwap']:,.2f}")
print(f" Volume Imbalance: {analysis['volume_imbalance']:.2%}")
print(f" Große Trades (>100k): {analysis['large_trades_count']}")
---
Kombination: Order Book + Trade Flow für Trading-Strategien
import time
from threading import Thread
from collections import deque
class MarketAnalyzer:
"""
Kombiniert Order Book und Trade Daten für Trading-Entscheidungen
"""
def __init__(self, inst_id: str = "BTC-USDT"):
self.inst_id = inst_id
self.order_book_client = OKXMarketData()
self.trade_client = OKXTradeData()
# Historische Daten für Trends
self.price_history = deque(maxlen=100)
self.volume_history = deque(maxlen=100)
self.imbalance_history = deque(maxlen=50)
def get_market_snapshot(self) -> Dict:
"""Vollständiger Marktsnapshot"""
# Parallel abrufen für Geschwindigkeit
order_book = self.order_book_client.get_order_book_depth_snapshot(self.inst_id)
trades = self.trade_client.get_recent_trades(self.inst_id, limit=50)
trade_analysis = self.trade_client.analyze_trade_flow(trades)
if not order_book or not trades:
return {}
# Zusammenfassung erstellen
snapshot = {
"timestamp": datetime.now().isoformat(),
"inst_id": self.inst_id,
"price": {
"mid": order_book["mid_price"],
"best_bid": order_book["best_bid"],
"best_ask": order_book["best_ask"]
},
"order_book": {
"spread_pct": order_book["spread_pct"],
"bid_volume": order_book["bid_volume_20"],
"ask_volume": order_book["ask_volume_20"],
"imbalance": order_book["imbalance"]
},
"trades": {
"count": trade_analysis["total_trades"],
"volume": trade_analysis["total_volume"],
"buy_ratio": trade_analysis["buy_ratio"],
"vwap": trade_analysis["vwap"],
"large_trades": trade_analysis["large_trades_count"]
}
}
# Signale generieren
snapshot["signals"] = self._generate_signals(snapshot)
return snapshot
def _generate_signals(self, snapshot: Dict) -> Dict:
"""Trading-Signale basierend auf Marktdaten"""
signals = {
"order_imbalance": None,
"trade_sentiment": None,
"spread_signal": None,
"combined": "NEUTRAL"
}
# Order Book Imbalance Signal
imb = snapshot["order_book"]["imbalance"]
if imb > 0.2:
signals["order_imbalance"] = "BULLISH"
elif imb < -0.2:
signals["order_imbalance"] = "BEARISH"
else:
signals["order_imbalance"] = "NEUTRAL"
# Trade Sentiment
buy_ratio = snapshot["trades"]["buy_ratio"]
if buy_ratio > 0.6:
signals["trade_sentiment"] = "BULLISH"
elif buy_ratio < 0.4:
signals["trade_sentiment"] = "BEARISH"
else:
signals["trade_sentiment"] = "NEUTRAL"
# Spread Signal (niedriger Spread = gesunder Markt)
spread_pct = snapshot["order_book"]["spread_pct"]
if spread_pct < 0.05:
signals["spread_signal"] = "LOW_SPREAD"
elif spread_pct > 0.1:
signals["spread_signal"] = "HIGH_SPREAD"
else:
signals["spread_signal"] = "NORMAL"
# Kombiniertes Signal
bullish_count = sum(1 for v in [signals["order_imbalance"], signals["trade_sentiment"]]
if v == "BULLISH")
if bullish_count >= 2:
signals["combined"] = "BULLISH"
elif bullish_count == 0:
signals["combined"] = "BEARISH"
return signals
def run_analysis_loop(self, interval: int = 5):
"""Kontinuierliche Analyse-Schleife"""
print(f"🔄 Starte Analyse für {self.inst_id} (Intervall: {interval}s)")
print("Drücken Sie Ctrl+C zum Beenden\n")
try:
while True:
snapshot = self.get_market_snapshot()
if snapshot:
print(f"\n⏰ {snapshot['timestamp']}")
print(f"💰 Preis: ${snapshot['price']['mid']:,.2f}")
print(f"📊 Order Imbalance: {snapshot['order_book']['imbalance']:.2%}")
print(f"📈 Trade Buy Ratio: {snapshot['trades']['buy_ratio']:.2%}")
print(f"🎯 Signal: {snapshot['signals']['combined']}")
time.sleep(interval)
except KeyboardInterrupt:
print("\n\n🛑 Analyse beendet")
Starte kontinuierliche Analyse
analyzer = MarketAnalyzer("BTC-USDT")
analyzer.run_analysis_loop(interval=5)
---
WebSocket für Echtzeit-Daten (Optional)
Für noch schnellere Daten können Sie WebSockets nutzen:
import websocket
import json
import threading
import time
class OKXWebSocketClient:
"""WebSocket Client für Echtzeit-Marktdaten"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(self):
self.ws = None
self.running = False
self.callbacks = []
def on_message(self, ws, message):
"""Nachrichten-Handler"""
data = json.loads(message)
# Trade-Daten
if "data" in data:
for item in data["data"]:
if item.get("instType") == "SPOT":
trade_info = {
"inst_id": item["instId"],
"price": float(item["px"]),
"size": float(item["sz"]),
"side": item["side"],
"timestamp": item["ts"]
}
for callback in self.callbacks:
callback(trade_info)
def on_error(self, ws, error):
print(f"WebSocket Fehler: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("WebSocket geschlossen")
if self.running:
time.sleep(5)
self.connect()
def on_open(self, ws):
"""Subscription senden"""
subscribe_msg = {
"op": "subscribe",
"args": [
{
"channel": "trades",
"instId": "BTC-USDT"
},
{
"channel": "books-l2-tbt", # TBT = Top of Book Tick
"instId": "BTC-USDT"
}
]
}
ws.send(json.dumps(subscribe_msg))
print("✅ WebSocket Subscriptions aktiv")
def add_callback(self, callback):
"""Callback für neue Daten hinzufügen"""
self.callbacks.append(callback)
def connect(self):
"""WebSocket Verbindung herstellen"""
self.running = True
self.ws = websocket.WebSocketApp(
self.WS_URL,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
thread = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
def disconnect(self):
"""Verbindung trennen"""
self.running = False
if self.ws:
self.ws.close()
Beispiel-Callback
def handle_trade(trade):
print(f"🆕 Trade: {trade['inst_id']} @ ${trade['price']:,.2f} ({trade['side']})")
WebSocket starten
ws_client = OKXWebSocketClient()
ws_client.add_callback(handle_trade)
ws_client.connect()
---
Datenanalyse mit HolySheep AI
Nachdem Sie die Order-Book- und Trade-Daten gesammelt haben, können Sie **HolySheep AI** für fortgeschrittene Analysen nutzen. Mit der API von HolyShepe erhalten Sie Zugang zu leistungsstarken KI-Modellen zu einem Bruchteil der Kosten.
Kostenvergleich: 10 Millionen Token pro Monat
| Anbieter | Modell | Preis/MTok | Monatliche Kosten | Latenz |
|----------|--------|------------|-------------------|--------|
| OpenAI | GPT-4.1 | $8,00 | $80,00 | ~800ms |
| Anthropic | Claude Sonnet 4.5 | $15,00 | $150,00 | ~700ms |
| Google | Gemini 2.5 Flash | $2,50 | $25,00 | ~400ms |
| **HolySheep** | **DeepSeek V3.2** | **$0,42** | **$4,20** | **<50ms** |
Mit HolySheep sparen Sie **95%** gegenüber Claude Sonnet 4.5 für dieselbe Token-Menge!
import requests
import json
from typing import List, Dict
class HolySheepAIAnalyzer:
"""
HolySheep AI für Marktdaten-Analyse nutzen
base_url: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def analyze_order_book_with_ai(self, order_book_data: Dict) -> str:
"""
Order Book von KI analysieren lassen
Nutzt DeepSeek V3.2 für schnelle und genaue Analyse
Kosten: nur $0,42 pro Million Token!
"""
prompt = f"""Analysiere das folgende Order Book für Trading-Signale:
Best Bid: ${order_book_data['best_bid']:,.2f}
Best Ask: ${order_book_data['best_ask']:,.2f}
Mid Price: ${order_book_data['mid_price']:,.2f}
Spread: {order_book_data['spread_pct']:.4f}%
Bid Volume (Top 20): {order_book_data['bid_volume_20']:.4f}
Ask Volume (Top 20): {order_book_data['ask_volume_20']:.4f}
Order Imbalance: {order_book_data['imbalance']:.2%}
Gib eine kurze Einschätzung mit:
1. Kurzfristiger Bias (Bullish/Bearish/Neutral)
2. Support-Level
3. Resistance-Level
4. Risiko-Einschätzung"""
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500,
"temperature": 0.3
},
timeout=10
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
return f"Fehler: {response.status_code}"
def analyze_trade_flow_with_ai(self, trade_data: Dict) -> str:
"""
Trade Flow von KI analysieren lassen
"""
prompt = f"""Analysiere den folgenden Trade Flow:
Gesamt Trades: {trade_data['total_trades']}
Buy/Sell Ratio: {trade_data['buy_ratio']:.2%}
Gesamt Volumen: ${trade_data['total_volume']:,.2f}
VWAP: ${trade_data['vwap']:,.2f}
Volume Imbalance: {trade_data['volume_imbalance']:.2%}
Große Trades (>100k): {trade_data['large_trades_count']}
Erkläre:
1. Sentiment (kauft oder verkauft das Volumen?)
2. Whales aktiv? (basierend auf großen Trades)
3. Kurzfristige Prognose"""
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500,
"temperature": 0.3
},
timeout=10
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
return f"Fehler: {response.status_code}"
Verwendung mit HolySheep
💡 Registrieren Sie sich hier: https://www.holysheep.ai/register
analyzer = HolySheepAIAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
Order Book analysieren
order_analysis = analyzer.analyze_order_book_with_ai(depth_data)
print("🤖 KI Analyse des Order Books:")
print(order_analysis)
Trade Flow analysieren
trade_analysis = analyzer.analyze_trade_flow_with_ai(analysis)
print("\n🤖 KI Analyse des Trade Flows:")
print(trade_analysis)
---
Häufige Fehler und Lösungen
1. **Rate Limit überschritten (HTTP 429)**
**Problem:** Zu viele Anfragen in kurzer Zeit.
**Lösung:**
import time
from functools import wraps
import requests
def rate_limit(max_calls: int = 20, period: int = 2):
"""
Rate Limiting Decorator
max_calls: Maximale Anfragen
period: Zeitraum in Sekunden
"""
calls = []
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
# Entferne alte Timestamps
calls[:] = [t for t in calls if now - t < period]
if len(calls) >= max_calls:
sleep_time = period - (now - calls[0])
if sleep_time > 0:
print(f"⏳ Rate Limit erreicht. Warte {sleep_time:.2f}s...")
time.sleep(sleep_time)
calls.append(time.time())
return func(*args, **kwargs)
return wrapper
return decorator
@rate_limit(max_calls=20, period=2)
def safe_get_order_book(inst_id: str):
"""Order Book mit automatischem Rate Limiting"""
endpoint = f"https://www.okx.com/api/v5/market/books-l2?instId={inst_id}&sz=400"
response = requests.get(endpoint, timeout=10)
return response.json()
Oder mit Exponential Backoff für robuste Fehlerbehandlung
def robust_api_call(func, max_retries: int = 3):
"""API-Aufruf mit Exponential Backoff"""
for attempt in range(max_retries):
try:
return func()
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"⚠️ Versuch {attempt + 1} fehlgeschlagen. Warte {wait_time}s...")
time.sleep(wait_time)
else:
raise e
2. **Ungültige Timestamp-Signatur**
**Problem:** Authentifizierungsfehler bei signierten Anfragen.
**Lösung:**
def fixed_sign(timestamp: str, method: str, path: str, body: str = "") -> str:
"""
Korrigierte Signatur-Generierung für OKX API
WICHTIG:
- Timestamp muss mit API synchronisiert sein
- Body muss EXAKT so wie im Request verwendet werden
"""
import json
# Stelle sicher, dass Timestamp im richtigen Format ist
# OKX erwartet: YYYY-MM-DDTHH:MM:SS.SSSZ
if not timestamp.endswith('Z'):
# Timestamp von OKX API verwenden (nicht lokale Zeit!)
api_time_response = requests.get("https://www.okx.com/api/v5/public/time")
timestamp = api_time_response.json()["data"][0]["ts"]
# Body muss leerer String sein wenn keine Payload
body_str = body if body else ""
# Nachricht erstellen
message = timestamp + method + path + body_str
# Signatur mit korrektem Key-Format
mac = hmac.new(
base64.b64decode(secret_key), # Key muss Base64-dekodiert werden!
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
Verwendung
timestamp = client._get_timestamp() # Von OKX API holen!
signature = fixed_sign(timestamp, "GET", "/api/v5/account/balance")
3. **Daten-Inkonsistenzen bei WebSocket**
**Problem:** Doppelte oder fehlende Daten bei WebSocket-Verbindung.
**Lösung:**
import json
from collections import defaultdict
class WebSocketDataCleaner:
"""Bereinigt und synchronisiert WebSocket-Daten"""
def __init__(self):
self.last_trade_ids = set()
self.order_book_state = defaultdict(dict)
self.checksum_errors = 0
def process_trade(self, trade_data: dict) -> dict:
"""
Trade-Daten bereinigen:
- Duplikate entfernen
- Sortierung nach Timestamp
"""
trade_id = trade_data.get("tradeId")
# Duplikat-Prüfung
if trade_id in self.last_trade_ids:
return None # Duplikat, ignorieren
self.last_trade_ids.add(trade_id)
# Nur die letzten 1000 IDs behalten
if len(self.last_trade_ids) > 1000:
self.last_trade_ids = set(list(self.last_trade_ids)[-1000:])
return trade_data
def process_order_book_update(self, update: dict) -> dict:
"""
Order Book Updates verarbeiten:
- Checksum verifizieren (falls vorhanden)
- Inkrementelle Updates anwenden
"""
# Checksum-Verifikation
checksum = update.get("chk")
if checksum:
calculated = self._calculate_checksum(update)
if calculated != checksum:
self.checksum_errors += 1
print(f"⚠️ Checksum Fehler #{self.checksum_errors}")
# Neuen Snapshot anfordern
return {"action": "RESET", "data": update}
# Inkrementelles Update anwenden
inst_id = update.get("instId")
for bid in update.get("bids", []):
price, size = float(bid[0]), float(bid[1])
if size == 0:
self.order_book_state[inst_id].pop(price, None)
else:
self.order_book_state[inst_id][price] = size
for ask in update.get("asks", []):
price, size = float(ask[0]), float(ask[1])
if size == 0:
self.order_book_state[inst_id].pop(price, None)
else:
self.order_book_state[inst_id][price] = size
return {"action": "UPDATE", "state": dict(self.order_book_state[inst_id])}
def _calculate_checksum(self, data: dict) -> str:
"""Checksum für Order Book verifizieren"""
# Vereinfachte Checksum-Berechnung
bids = data.get("bids", [])[:25]
asks = data.get("asks", [])[:25]
bid_str = "|".join([f"{b[0]}:{b[1]}" for b in bids])
ask_str = "|".join([f"{a[0]}:{a[1]}" for a in asks])
return f"{bid_str}|{ask_str}"
def get_current_state(self, inst_id: str) -> dict:
"""Aktuellen Order Book State abrufen"""
return dict(self.order_book_state.get(inst_id, {}))
WebSocket mit Cleaner integrieren
cleaner = WebSocketDataCleaner()
def on_trade_message(message):
trade = json.loads(message)
cleaned = cleaner.process_trade(trade)
if cleaned:
print(f"Neuer Trade: {cleaned}")
4. **Verbindungsunterbrechungen**
**Problem:** WebSocket trennt unerwartet, Daten gehen verloren.
**Lösung:**
import websocket
import threading
import time
import random
class ResilientWebSocket:
"""WebSocket mit automatischer Wiederverbindung"""
def __init__(self, url, on_message, on_error, subscriptions):
self.url = url
self.on_message = on_message
self.on_error = on_error
self.subscriptions = subscriptions
self.ws = None
self.running = False
self.reconnect_delay = 1
self.max_reconnect_delay = 60
def connect(self):
"""Verbindung herstellen mit automatischer Wiederverbindung"""
self.running = True
thread = threading.Thread(target=self._run_forever)
thread.daemon = True
thread.start()
def _run_forever(self):
"""Endlosschleife mit exponentieller Wiederverbindung
Verwandte Ressourcen
Verwandte Artikel