Stellen Sie sich vor: Sie entwickeln einen 算法交易机器人 für Krypto-Derivate. Ihr System muss in Echtzeit die feinsten Marktdaten verarbeiten – nicht die aggregierten 1-Minute-Kerzen, sondern 逐笔成交 (jede einzelne Transaktion) und die präzise 订单簿-Struktur. Genau das habe ich vergangenen Monat für ein quantitatives Hedgefonds-Projekt umgesetzt. Die Herausforderung: Bybits offizielle Dokumentation ist lückenhaft, und die wenigsten Tutorials zeigen, wie man aus den Rohdaten ein brauchbares Orderbuch rekonstruiert.
In diesem Tutorial zeige ich Ihnen Step-by-Step, wie Sie mit der Bybit Unified Trading Account API Tick-Daten abrufen, diese für ML-Modelle aufbereiten und ein lokales Orderbuch in Echtzeit pflegen – inklusive Fehlerbehandlung und Performance-Optimierung.
1. Voraussetzungen und API-Setup
Bevor wir beginnen, benötigen Sie:
- Ein Bybit-Konto mit aktiviertem API-Key (Unified Trading Account)
- Python 3.9+ mit
websockets,requestsundpandas - Optional: HolySheep AI Jetzt registrieren für sentimentbasierte Signalanalyse via LLM
API-Anmeldedaten konfigurieren
# config.py
import os
BYBIT_API_KEY = os.getenv("BYBIT_API_KEY", "your_bybit_api_key")
BYBIT_API_SECRET = os.getenv("BYBIT_API_SECRET", "your_bybit_secret")
BYBIT_TESTNET = True # Für Production auf False setzen
HolySheep AI für erweiterte Analyse
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "your_holysheep_key")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
WebSocket-Endpunkte (Bybit)
if BYBIT_TESTNET:
WS_URL = "wss://stream-testnet.bybit.com/v5/trade"
REST_URL = "https://api-testnet.bybit.com/v5"
else:
WS_URL = "wss://stream.bybit.com/v5/trade"
REST_URL = "https://api.bybit.com/v5"2. Tick-by-Tick成交数据获取 via WebSocket
Die effizienteste Methode für Echtzeit-Transaktionsdaten ist der Bybit WebSocket-Stream. Für Futures im Unified Trading Account nutzen wir den trade-Topic.
# bybit_trade_stream.py
import json
import time
import asyncio
from websocket import WebSocketApp
from config import WS_URL
class BybitTradeStream:
def __init__(self, symbols: list, on_trade_callback=None):
self.symbols = symbols
self.on_trade_callback = on_trade_callback
self.ws = None
self.trade_buffer = []
def _on_message(self, ws, message):
data = json.loads(message)
# Nur Trade-Daten verarbeiten
if data.get("topic", "").startswith("trade."):
for trade in data.get("data", []):
tick = {
"symbol": trade["s"],
"side": trade["S"], # Buy oder Sell
"price": float(trade["p"]), # Ausführungspreis
"size": float(trade["v"]), # Anzahl Kontrakte
"timestamp": int(trade["T"]), # Transaktionszeit
"trade_id": trade["i"]
}
self.trade_buffer.append(tick)
# Callback für Echtzeit-Verarbeitung
if self.on_trade_callback:
self.on_trade_callback(tick)
def _on_error(self, ws, error):
print(f"WebSocket Fehler: {error}")
def _on_close(self, ws, close_status_code, close_msg):
print(f"Verbindung geschlossen: {close_status_code}")
# Automatischer Reconnect nach 5 Sekunden
time.sleep(5)
self.connect()
def _on_open(self, ws):
# Subscribe zu Trade-Streams für gewünschte Symbols
subscribe_msg = {
"op": "subscribe",
"args": [f"trade.{symbol}" for symbol in self.symbols]
}
ws.send(json.dumps(subscribe_msg))
print(f" subscribed to: {self.symbols}")
def connect(self):
self.ws = WebSocketApp(
WS_URL,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
def start(self):
self.connect()
# Non-blocking WebSocket-Loop
import threading
thread = threading.Thread(target=self.ws.run_forever, daemon=True)
thread.start()
return thread
Beispiel-Nutzung
def process_trade(tick):
"""Echtzeit-Verarbeitung jeder Transaktion"""
print(f"Trade: {tick['symbol']} | {tick['side']} | "
f"Preis: {tick['price']} | Größe: {tick['size']}")
stream = BybitTradeStream(
symbols=["BTCUSDT", "ETHUSDT"],
on_trade_callback=process_trade
)
stream.start()
print("Trade-Stream aktiv...")3. Orderbuch-Rekonstruktion aus Tick-Daten
Die wahre Kunst liegt in der Orderbuch-Rekonstruktion. Bybit bietet zwar einen Orderbook-WebSocket, aber für ML-Training und Backtesting brauchen Sie volle Kontrolle über die Datenstruktur.
# orderbook_reconstructor.py
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
import time
@dataclass
class OrderBookLevel:
price: float
size: float
@property
def notional(self) -> float:
return self.price * self.size
@dataclass
class OrderBook:
symbol: str
bids: Dict[float, float] = field(default_factory=dict) # price -> size
asks: Dict[float, float] = field(default_factory=dict)
last_update_time: int = 0
sequence: int = 0
def add_trade(self, side: str, price: float, size: float):
"""
Trade verarbeitet das Orderbuch:
- Sell => reduziert Bids (Kaufaufträge wurden "gehittet")
- Buy => reduziert Asks (Verkaufsaufträge wurden "gehittet")
"""
if side == "Sell":
# Verkäufer "nimmt" Bid-Seite
self.bids[price] = max(0, self.bids.get(price, 0) - size)
if self.bids[price] <= 0:
del self.bids[price]
else: # Buy
# Käufer "nimmt" Ask-Seite
self.asks[price] = max(0, self.asks.get(price, 0) - size)
if self.asks[price] <= 0:
del self.asks[price]
def add_order(self, side: str, price: float, size: float):
"""Neue Limit-Order zum Orderbuch hinzufügen"""
if side == "Buy":
if size == 0:
self.bids.pop(price, None)
else:
self.bids[price] = size
else:
if size == 0:
self.asks.pop(price, None)
else:
self.asks[price] = size
def get_spread(self) -> float:
"""Bid-Ask-Spread in Basispunkten"""
best_bid = max(self.bids.keys()) if self.bids else 0
best_ask = min(self.asks.keys()) if self.asks else float('inf')
if best_ask == float('inf'):
return 0
return (best_ask - best_bid) / best_bid * 10000 if best_bid else 0
def get_mid_price(self) -> float:
"""Mittelkurs"""
best_bid = max(self.bids.keys()) if self.bids else 0
best_ask = min(self.asks.keys()) if self.asks else 0
return (best_bid + best_ask) / 2 if best_bid and best_ask else 0
def get_top_levels(self, depth: int = 10) -> Tuple[List[OrderBookLevel], List[OrderBookLevel]]:
"""Top N Orderbuch-Ebenen für Visualisierung"""
top_bids = [
OrderBookLevel(p, s)
for p, s in sorted(self.bids.items(), reverse=True)[:depth]
]
top_asks = [
OrderBookLevel(p, s)
for p, s in sorted(self.asks.items())[:depth]
]
return top_bids, top_asks
def to_dataframe(self) -> dict:
"""Für pandas-Export"""
return {
"symbol": self.symbol,
"mid_price": self.get_mid_price(),
"spread_bps": self.get_spread(),
"bid_depth": len(self.bids),
"ask_depth": len(self.asks),
"total_bid_notional": sum(p * s for p, s in self.bids.items()),
"total_ask_notional": sum(p * s for p, s in self.asks.items()),
"timestamp": time.time()
}
class OrderBookManager:
"""Zentrales Management für mehrere Orderbücher"""
def __init__(self):
self.books: Dict[str, OrderBook] = {}
def get_or_create(self, symbol: str) -> OrderBook:
if symbol not in self.books:
self.books[symbol] = OrderBook(symbol=symbol)
return self.books[symbol]
def process_tick(self, tick: dict):
"""Einen einzelnen Trade verarbeiten"""
book = self.get_or_create(tick["symbol"])
book.add_trade(
side=tick["side"],
price=tick["price"],
size=tick["size"]
)
def process_orderbook_snapshot(self, symbol: str, data: dict):
"""Kompletten Orderbuch-Snapshot verarbeiten (Initialisierung)"""
book = self.get_or_create(symbol)
# Bids und Asks aus Snapshot
for price, size in data.get("b", []): # bids
book.bids[float(price)] = float(size)
for price, size in data.get("a", []): # asks
book.asks[float(price)] = float(size)
book.last_update_time = data.get("u", 0)
def export_state(self) -> dict:
"""Aktuellen Zustand aller Orderbücher exportieren"""
return {
symbol: book.to_dataframe()
for symbol, book in self.books.items()
}
Beispiel-Nutzung
manager = OrderBookManager()
Simulierte Tick-Daten verarbeiten
simulated_trades = [
{"symbol": "BTCUSDT", "side": "Sell", "price": 67450.50, "size": 0.5},
{"symbol": "BTCUSDT", "side": "Buy", "price": 67451.00, "size": 0.3},
{"symbol": "BTCUSDT", "side": "Sell", "price": 67450.00, "size": 1.2},
]
for trade in simulated_trades:
manager.process_tick(trade)
book = manager.get_or_create("BTCUSDT")
print(f"Mid Price: ${book.get_mid_price():.2f}")
print(f"Spread: {book.get_spread():.2f} bps")
print(f"Bid Depth: {book.bids}")
print(f"Ask Depth: {book.asks}")4. REST-API für Historische Daten
Für Backtesting benötigen Sie historische Tick-Daten. Die REST-API liefert diese in 1.000er-Chargen.
# bybit_historical.py
import requests
import time
import hmac
import hashlib
from typing import List, Dict, Optional
from config import REST_URL, BYBIT_API_KEY, BYBIT_API_SECRET
class BybitRESTClient:
def __init__(self, testnet: bool = True):
self.base_url = REST_URL
self.api_key = BYBIT_API_KEY
self.secret = BYBIT_API_SECRET
def _generate_signature(self, params: str) -> str:
"""HMAC-SHA256 Signatur erstellen"""
return hmac.new(
self.secret.encode(),
params.encode(),
hashlib.sha256
).hexdigest()
def get_recent_trades(
self,
symbol: str,
limit: int = 1000,
cursor: Optional[str] = None
) -> Dict:
"""
Historische Trades abrufen
Limit max: 1000 pro Request
"""
endpoint = "/v5/market/recent-trade"
params = {
"category": "linear", # USDT Perpetuals
"symbol": symbol,
"limit": min(limit, 1000)
}
if cursor:
params["cursor"] = cursor
url = f"{self.base_url}{endpoint}"
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
def get_orderbook(
self,
symbol: str,
depth: int = 50
) -> Dict:
"""
Aktuelles Orderbuch (Snapshot)
"""
endpoint = "/v5/market/orderbook"
params = {
"category": "linear",
"symbol": symbol,
"limit": depth
}
url = f"{self.base_url}{endpoint}"
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
def fetch_historical_ticks(
self,
symbol: str,
start_time: int = None,
end_time: int = None,
max_records: int = 10000
) -> List[Dict]:
"""
Alle Trades im Zeitraum sammeln (mit Auto-Pagination)
start_time/end_time in Millisekunden
"""
all_trades = []
cursor = None
while len(all_trades) < max_records:
if cursor:
data = self.get_recent_trades(symbol, cursor=cursor)
else:
data = self.get_recent_trades(symbol)
if data.get("retCode") != 0:
raise Exception(f"API Error: {data.get('retMsg')}")
trades = data.get("result", {}).get("list", [])
if not trades:
break
all_trades.extend(trades)
# Zeitraum-Filter anwenden
if start_time:
first_ts = int(trades[0]["T"])
if first_ts < start_time:
break
# Pagination
cursor = data.get("result", {}).get("nextPageCursor")
if not cursor:
break
# Rate Limiting: max 100 req/10s = 1 req pro 100ms
time.sleep(0.15)
return all_trades[:max_records]
Beispiel: Letzte 5.000 BTC-Trades abrufen
client = BybitRESTClient(testnet=True)
Aktuelles Orderbuch
book_snapshot = client.get_orderbook("BTCUSDT", depth=20)
print("Orderbook Snapshot:")
print(f"Bids: {book_snapshot['result']['b'][:3]}")
print(f"Asks: {book_snapshot['result']['a'][:3]}")
Historische Trades
24 Stunden zurück
end_ts = int(time.time() * 1000)
start_ts = end_ts - (24 * 60 * 60 * 1000)
trades = client.fetch_historical_ticks(
symbol="BTCUSDT",
start_time=start_ts,
end_time=end_ts,
max_records=5000
)
print(f"\n{len(trades)} Trades im Zeitraum geladen")5. Integration mit HolySheep AI für Sentiment-Analyse
Der spannendste Teil: Nutzen Sie die Trade-Daten für KI-gestützte Sentiment-Analyse. Mit HolySheep AI Jetzt registrieren können Sie每小时 Tausende von Trades automatisch analysieren – mit <50ms Latenz und 85%+ Kostenersparnis gegenüber OpenAI.
# holysheep_sentiment.py
import requests
from config import HOLYSHEEP_BASE_URL, HOLYSHEEP_API_KEY
from orderbook_reconstructor import OrderBookManager
import json
class TradeSentimentAnalyzer:
"""
Analysiert Tick-Muster für Sentiment-Signale
Nutzt HolySheep AI für erweiterte Text-/Kontextanalyse
"""
def __init__(self):
self.api_key = HOLYSHEEP_API_KEY
self.base_url = HOLYSHEEP_BASE_URL
self.orderbook_manager = OrderBookManager()
# Latenz-Tracking
self.request_count = 0
self.total_latency_ms = 0
def analyze_trade_pattern(
self,
trades: list,
use_ai: bool = True
) -> dict:
"""
Trade-Sequenz analysieren
Args:
trades: Liste von Trade-Dicts
use_ai: Whether to use HolySheep LLM for deep analysis
"""
# Statistische Basis-Analyse
buy_volume = sum(t["size"] for t in trades if t["side"] == "Buy")
sell_volume = sum(t["size"] for t in trades if t["side"] == "Sell")
buy_count = sum(1 for t in trades if t["side"] == "Buy")
sell_count = sum(1 for t in trades if t["side"] == "Sell")
prices = [t["price"] for t in trades]
price_change = (max(prices) - min(prices)) / min(prices) * 100
basic_signal = {
"buy_ratio": buy_count / (buy_count + sell_count),
"volume_imbalance": (buy_volume - sell_volume) / (buy_volume + sell_volume),
"price_volatility_pct": price_change,
"total_trades": len(trades),
"buy_volume": buy_volume,
"sell_volume": sell_volume
}
if use_ai and len(trades) >= 50:
# HolySheep AI Deep Dive
ai_analysis = self._call_holysheep(trades, basic_signal)
return {**basic_signal, "ai_analysis": ai_analysis}
return basic_signal
def _call_holysheep(self, trades: list, signal: dict) -> dict:
"""
HolySheep AI für kontextbasierte Sentiment-Analyse
"""
# Trade-Sequenz als Text formatieren
trade_summary = self._format_trades_for_llm(trades)
prompt = f"""Analysiere die folgenden Krypto-Trade-Daten für {trades[0]['symbol']}:
Statistik:
- Buy Ratio: {signal['buy_ratio']:.2%}
- Volume Imbalance: {signal['volume_imbalance']:.2%}
- Volatilität: {signal['price_volatility_pct']:.2f}%
- Gesamt-Trades: {signal['total_trades']}
Letzte 20 Trades:
{trade_summary}
Gib zurück:
1. Kurzfristiges Sentiment (Bullish/Bearish/Neutral)
2. Aggressions-Index (0-10, wer treibt den Markt?)
3. Whales-Score (0-10, Hinweis auf Großanleger)
4. Eine kurze Erklärung (2 Sätze)
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat", # Günstigste Option: $0.42/MTok
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 200
}
start_time = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=10
)
latency_ms = (time.time() - start_time) * 1000
self.request_count += 1
self.total_latency_ms += latency_ms
response.raise_for_status()
result = response.json()
return {
"sentiment": result["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"model": "deepseek-chat",
"cost_estimate_usd": self._estimate_cost(result)
}
def _format_trades_for_llm(self, trades: list) -> str:
"""Letzte 20 Trades als Text formatieren"""
last_20 = trades[-20:]
lines = []
for t in last_20:
side = "B" if t["side"] == "Buy" else "S"
lines.append(f"{side}: ${t['price']:.2f} x {t['size']}")
return "\n".join(lines)
def _estimate_cost(self, response: dict) -> float:
"""Kostenschätzung für HolySheep (DeepSeek V3.2: $0.42/MTok)"""
usage = response.get("usage", {})
tokens = usage.get("total_tokens", 0)
return tokens / 1_000_000 * 0.42 # DeepSeek-Preis
def get_stats(self) -> dict:
avg_latency = self.total_latency_ms / self.request_count if self.request_count else 0
return {
"total_requests": self.request_count,
"avg_latency_ms": round(avg_latency, 2),
"estimated_total_cost_usd": self.request_count * 0.00042
}
Beispiel-Nutzung
analyzer = TradeSentimentAnalyzer()
Simulierte Trade-Daten
simulated_trades = [
{"symbol": "BTCUSDT", "side": "Buy", "price": 67450, "size": 5.0, "timestamp": 1700000000},
{"symbol": "BTCUSDT", "side": "Sell", "price": 67449, "size": 2.0, "timestamp": 1700000001},
# ... mehr Trades
]
for i in range(100):
side = "Buy" if i % 3 != 0 else "Sell"
price = 67450 + (i % 10)
size = 0.5 + (i % 5) * 0.5
simulated_trades.append({
"symbol": "BTCUSDT",
"side": side,
"price": price,
"size": size,
"timestamp": 1700000000 + i
})
Analyse durchführen
result = analyzer.analyze_trade_pattern(simulated_trades, use_ai=True)
print("=== Trade Sentiment Report ===")
print(f"Buy Ratio: {result['buy_ratio']:.1%}")
print(f"Volume Imbalance: {result['volume_imbalance']:+.2f}")
if "ai_analysis" in result:
print(f"\nAI Sentiment: {result['ai_analysis']['sentiment']}")
print(f"Latenz: {result['ai_analysis']['latency_ms']}ms")
print(f"Kosten: ${result['ai_analysis']['cost_estimate_usd']:.6f}")
stats = analyzer.get_stats()
print(f"\n=== Gesamt-Statistik ===")
print(f"Anfragen: {stats['total_requests']}")
print(f"Durchschn. Latenz: {stats['avg_latency_ms']}ms")6. Vollständiges Beispiel: Real-Time Trading Dashboard
# trading_dashboard.py
import asyncio
import streamlit as st
import pandas as pd
import time
from bybit_trade_stream import BybitTradeStream
from orderbook_reconstructor import OrderBookManager, OrderBook
from holysheep_sentiment import TradeSentimentAnalyzer
class TradingDashboard:
"""Real-Time Trading Dashboard mit Orderbuch und Sentiment"""
def __init__(self, symbols: list):
self.symbols = symbols
self.orderbook_manager = OrderBookManager()
self.sentiment_analyzer = TradeSentimentAnalyzer()
# Trade-Speicher für laufende Analyse
self.trade_buffers = {s: [] for s in symbols}
self.buffer_size = 100
# Stream initialisieren
self.stream = BybitTradeStream(
symbols=symbols,
on_trade_callback=self._on_trade
)
def _on_trade(self, tick: dict):
"""Callback für jeden neuen Trade"""
# Orderbuch aktualisieren
self.orderbook_manager.process_tick(tick)
# Trade-Buffer pflegen
symbol = tick["symbol"]
self.trade_buffers[symbol].append(tick)
if len(self.trade_buffers[symbol]) > self.buffer_size:
self.trade_buffers[symbol] = self.trade_buffers[symbol][-self.buffer_size:]
def get_display_data(self, symbol: str) -> dict:
"""Alle Daten für ein Symbol für die Anzeige"""
book = self.orderbook_manager.get_or_create(symbol)
return {
"orderbook": book,
"trades": self.trade_buffers[symbol],
"mid_price": book.get_mid_price(),
"spread": book.get_spread(),
"sentiment": self.sentiment_analyzer.analyze_trade_pattern(
self.trade_buffers[symbol],
use_ai=True
) if len(self.trade_buffers[symbol]) >= 50 else None
}
def run_streamlit(self):
"""Streamlit UI starten"""
st.title("🚀 Real-Time Trading Dashboard")
# Live-Daten-Updates
while True:
for symbol in self.symbols:
data = self.get_display_data(symbol)
col1, col2 = st.columns(2)
with col1:
st.subheader(f"{symbol} Order Book")
bids, asks = data["orderbook"].get_top_levels(5)
bid_df = pd.DataFrame([
{"Preis": b.price, "Größe": b.size, "Notional": b.notional}
for b in bids
])
st.dataframe(bid_df, use_container_width=True)
with col2:
st.subheader(f"{symbol} Sentiment")
st.metric("Mid Price", f"${data['mid_price']:.2f}")
st.metric("Spread", f"{data['spread']:.2f} bps")
if data["sentiment"]:
st.write(f"Buy Ratio: {data['sentiment']['buy_ratio']:.1%}")
st.write(f"Volume Imbalance: {data['sentiment']['volume_imbalance']:+.2f}")
if data["sentiment"].get("ai_analysis"):
ai = data["sentiment"]["ai_analysis"]
st.info(ai["sentiment"][:200] + "...")
st.caption(f"Latenz: {ai['latency_ms']}ms | Kosten: ${ai['cost_estimate_usd']:.6f}")
time.sleep(1) # 1-Sekunden-Update
st.rerun()
Dashboard starten
if __name__ == "__main__":
dashboard = TradingDashboard(symbols=["BTCUSDT", "ETHUSDT"])
dashboard.stream.start()
dashboard.run_streamlit()Häufige Fehler und Lösungen
1. WebSocket-Verbindung bricht unerwartet ab (1006/Close Code)
Symptom: Die Verbindung wird ohne Fehlermeldung geschlossen, oft nach 5-30 Minuten.
# Fehlerursache: Server-seitiges Timeout bei Inaktivität
Lösung: Regelmäßigen Ping senden oder Subscription erneuern
class RobustWebSocket:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.last_ping = time.time()
self.ping_interval = 20 # Sekunden
def _on_ping(self, ws, data):
"""Automatischer Ping-Handler"""
ws.send(data, opcode=0x9) # Pong Frame
def _check_alive(self):
"""Verbindungs-Heartbeat"""
if time.time() - self.last_ping > self.ping_interval:
try:
self.ws.ping()
self.last_ping = time.time()
except:
self._reconnect()
def _reconnect(self):
"""Graceful Reconnect"""
print("Reconnecting...")
self.ws.close()
time.sleep(1)
self.ws = WebSocketApp(
self.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, daemon=True)
thread.start()2. Orderbuch-Drift: Preise weichen immer weiter ab
Symptom: Nach einigen Minuten stimmt das rekonstruierte Orderbuch nicht mehr mit dem echten überein.
# Fehlerursache: Fehlende Orderbuch-Updates (nur Trades reichen nicht)
Lösung: Periodisch Orderbuch-Snapshot vom REST-API holen
class SyncedOrderBook:
def __init__(self, symbol: str, rest_client):
self.symbol = symbol
self.rest_client = rest_client
self.local_book = OrderBook(symbol=symbol)
self.last_snapshot_time = 0
self.sync_interval = 60 # Sekunden
def update(self):
"""Orderbuch synchronisieren"""
current_time = time.time()
# Snapshots alle 60s holen
if current_time - self.last_snapshot_time > self.sync_interval:
snapshot = self.rest_client.get_orderbook(self.symbol, depth=50)
if snapshot.get("retCode") == 0:
data = snapshot["result"]
# Lokales Orderbuch mit Snapshot überschreiben
self.local_book.bids = {
float(p): float(s) for p, s in data["b"]
}
self.local_book.asks = {
float(p): float(s) for p, s in data["a"]
}
self.last_snapshot_time = current_time
print(f"[{self.symbol}] Orderbuch synchronisiert")
return self.local_book3. Rate Limit erreicht (10004/10005)
Symptom: API-Anfragen schlagen fehl mit "rate limit exceeded".
# Fehlerursache: Mehr als 100 Anfragen pro 10 Sekunden an REST-API
Lösung: Exponential Backoff mit Request-Queue
import threading
import time
from collections import deque
class RateLimitedClient:
def __init__(self, calls_per_10s: int = 80, burst_limit: int = 10):
self.window = 10 # Sekunden
self.max_calls = calls_per_10s
self.burst_limit = burst_limit
self.request_times = deque()
self.lock = threading.Lock()
def wait_and_call(self, func, *args, **kwargs):
"""Thread-safe Aufruf mit Rate-Limiting"""
with self.lock:
now = time.time()
# Alte Requests aus Window entfernen
cutoff = now - self.window
while self.request_times and self.request_times[0] < cutoff:
self.request_times.popleft()
# Burst-Limit prüfen
recent_count = len([t for t in self.request_times if now - t < 1])
if recent_count >= self.burst_limit:
sleep_time = 1 - (now - self.request_times[-1]) if self.request_times else 1
time.sleep(max(sleep_time, 0.1))
# Max-Calls prüfen
if len(self.request_times) >= self.max_calls:
oldest = self.request_times[0]
sleep_time = self.window - (now - oldest) + 0.1
time.sleep(sleep_time)
# Request durchführen
self.request_times.append(time.time())
return func(*args, **kwargs)
Beispiel-Nutzung
limited_client = RateLimitedClient(calls_per_10s=80)
for i in range(200):
result = limited_client.wait_and_call(
rest_client.get_recent_trades,
symbol="BTCUSDT"
)
print(f"Request {i+1}: OK")4. Falsche Timestamp-Interpretation
Symptom:Verwandte Ressourcen
Verwandte Artikel