Einleitung: Warum Tick-Daten für algorithmisches Trading entscheidend sind
Stellen Sie sich vor: Sie sind ein quantitativer Entwickler bei einem mittelgroßen Hedgefonds in Shanghai. Montagmorgen, 8:00 Uhr. Ihr Team hat in den letzten drei Monaten eine Mean-Reversion-Strategie für Bitcoin entwickelt, die auf 1-Sekunden-Tick-Daten basiert. Die Strategie zeigte im Backtesting eine annualisierte Rendite von 340%. Jetzt muss das Modell auf Echtzeitdaten validiert werden – doch die Kosten für den Kauf von Tick-Daten bei verschiedenen Börsen unterscheiden sich um den Faktor 10.
In diesem Leitfaden vergleichen wir die Tick-Data-Beschaffungskosten bei den drei größten Krypto-Börsen weltweit: Binance, Bybit und OKX. Anschließend zeigen wir, wie Sie diese Daten mit HolySheep AI für KI-gestützte Strategieanalyse nutzen können – mit einer 85%+ Kostenersparnis gegenüber OpenAI oder Anthropic.
Tick-Daten verstehen: Was macht sie so wertvoll?
Tick-Daten sind die granularsten verfügbaren Marktdaten. Jeder einzelne Handel (Trade) wird mit exaktem Zeitstempel, Preis, Volumen und Seite (Kauf/Verkauf) aufgezeichnet. Im Vergleich zu OHLCV-1-Minuten-Daten bieten Tick-Daten:
- 50-200x höhere Auflösung für Mikrostruktur-Analysen
- Orderflow-Analyse und Liquidity-Metriken
- Spread-Dynamik und Slippage-Modellierung
- Arbitrage-Erkennung in Millisekunden
Kostenvergleich: Binance, Bybit und OKX Tick Data
| Kriterium | Binance | Bybit | OKX |
|---|---|---|---|
| API-Zugang | WebSocket + REST | WebSocket + REST | WebSocket + REST |
| Kosten/Monat (Basic) | $299 | $199 | $249 |
| Kosten/Monat (Pro) | $899 | $599 | $749 |
| Kosten/Monat (Enterprise) | $2.499+ | $1.999+ | $2.249+ |
| Latenz | ~80ms | ~45ms | ~60ms |
| Historische Tiefe | 2 Jahre | 1 Jahr | 18 Monate |
| Webhook-Limit | 10.000/min | 15.000/min | 12.000/min |
| Bezahlmethoden | Kreditkarte, Banktransfer, Krypto | Kreditkarte, Krypto | WeChat Pay, Alipay, Krypto |
Code-Implementierung: Tick-Daten abrufen und speichern
Hier sind zwei vollständige Implementierungen für den Abruf von Tick-Daten von allen drei Börsen. Der erste Code zeigt die REST-API-Integration, der zweite die WebSocket-Verbindung für Echtzeit-Daten.
#!/usr/bin/env python3
"""
Tick Data Collector: Binance, Bybit, OKX
Kostenvergleich und Datensammlung für Backtesting
"""
import asyncio
import aiohttp
import json
from datetime import datetime
from typing import Dict, List
import sqlite3
=== KONFIGURATION ===
API_KEYS = {
"binance": "YOUR_BINANCE_API_KEY",
"bybit": "YOUR_BYBIT_API_KEY",
"okx": "YOUR_OKX_API_KEY"
}
DATABASE = "tick_data.db"
=== HOLYSHEEP AI INTEGRATION ===
Analysieren Sie Tick-Daten mit KI für Mustererkennung
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class TickDataCollector:
"""Sammelt Tick-Daten von allen drei Börsen"""
def __init__(self):
self.conn = sqlite3.connect(DATABASE)
self.create_tables()
def create_tables(self):
"""Erstellt Datenbanktabellen für Tick-Daten"""
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS binance_ticks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT,
price REAL,
quantity REAL,
timestamp INTEGER,
is_buyer_maker INTEGER,
trade_time DATETIME
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS bybit_ticks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT,
price REAL,
volume REAL,
timestamp INTEGER,
side TEXT,
trade_time DATETIME
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS okx_ticks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT,
price REAL,
volume REAL,
timestamp INTEGER,
side TEXT,
trade_time DATETIME
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS costs_comparison (
id INTEGER PRIMARY KEY AUTOINCREMENT,
exchange TEXT,
request_count INTEGER,
data_points INTEGER,
cost_usd REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
async def fetch_binance_recent_trades(self, symbol: str = "BTCUSDT") -> List[Dict]:
"""Holt recente Trades von Binance (kostenlos, limitierte Historie)"""
url = f"https://api.binance.com/api/v3/trades"
params = {"symbol": symbol, "limit": 1000}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return [
{
"symbol": symbol,
"price": float(t["price"]),
"quantity": float(t["qty"]),
"timestamp": t["time"],
"is_buyer_maker": t["isBuyerMaker"],
"trade_time": datetime.fromtimestamp(t["time"] / 1000)
}
for t in data
]
else:
print(f"Binance Fehler: {response.status}")
return []
async def fetch_bybit_recent_trades(self, symbol: str = "BTCUSDT") -> List[Dict]:
"""Holt recente Trades von Bybit"""
url = "https://api.bybit.com/v5/market/recent-trade"
params = {"category": "spot", "symbol": symbol, "limit": 1000}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
if data.get("retCode") == 0:
trades = data.get("result", {}).get("list", [])
return [
{
"symbol": symbol,
"price": float(t["price"]),
"volume": float(t["size"]),
"timestamp": int(t["ts"]),
"side": t["side"],
"trade_time": datetime.fromtimestamp(int(t["ts"]) / 1000)
}
for t in trades
]
return []
async def fetch_okx_recent_trades(self, symbol: str = "BTC-USDT") -> List[Dict]:
"""Holt recente Trades von OKX"""
url = "https://www.okx.com/api/v5/market/trades"
params = {"instId": symbol, "limit": 100}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
if data.get("code") == "0":
trades = data.get("data", [])
return [
{
"symbol": symbol,
"price": float(t["px"]),
"volume": float(t["sz"]),
"timestamp": int(t["ts"]),
"side": t["side"],
"trade_time": datetime.fromtimestamp(int(t["ts"]) / 1000)
}
for t in trades
]
return []
def store_binance_ticks(self, ticks: List[Dict]):
"""Speichert Binance Tick-Daten"""
cursor = self.conn.cursor()
for tick in ticks:
cursor.execute("""
INSERT INTO binance_ticks
(symbol, price, quantity, timestamp, is_buyer_maker, trade_time)
VALUES (?, ?, ?, ?, ?, ?)
""", (
tick["symbol"], tick["price"], tick["quantity"],
tick["timestamp"], tick["is_buyer_maker"], tick["trade_time"]
))
self.conn.commit()
print(f"Binance: {len(ticks)} Trades gespeichert")
def calculate_costs(self, data_points: Dict[str, int]):
"""Berechnet und vergleicht Kosten für verschiedene Börsen"""
print("\n" + "="*60)
print("KOSTENVERGLEICH FÜR TICK-DATEN")
print("="*60)
# Binance: Premium-API für historische Daten
binance_cost = 299 # Basic Plan $/Monat
binance_free_limit = 1200 # Anfragen/Minute
# Bybit: Wettbewerbsfähige Preise
bybit_cost = 199
bybit_free_limit = 6000
# OKX: Mittlerer Bereich
okx_cost = 249
okx_free_limit = 200
costs = {
"Binance": binance_cost,
"Bybit": bybit_cost,
"OKX": okx_cost
}
for exchange, cost in costs.items():
print(f"{exchange}: ${cost}/Monat")
cheapest = min(costs, key=costs.get)
print(f"\nGünstigste Option: {cheapest} mit ${costs[cheapest]}/Monat")
# Speichere Kostenvergleich
cursor = self.conn.cursor()
for exchange, points in data_points.items():
cursor.execute("""
INSERT INTO costs_comparison (exchange, data_points, cost_usd)
VALUES (?, ?, ?)
""", (exchange, points, costs.get(exchange, 0)))
self.conn.commit()
return costs
async def main():
"""Hauptprogramm: Sammelt Tick-Daten von allen Börsen"""
collector = TickDataCollector()
print("="*60)
print("STARTE TICK-DATEN SAMMLUNG")
print("="*60)
# Sammle Daten von allen Börsen parallel
tasks = [
collector.fetch_binance_recent_trades("BTCUSDT"),
collector.fetch_bybit_recent_trades("BTCUSDT"),
collector.fetch_okx_recent_trades("BTC-USDT")
]
results = await asyncio.gather(*tasks)
binance_data, bybit_data, okx_data = results
# Speichere Daten
collector.store_binance_ticks(binance_data)
# Berechne Kostenvergleich
data_points = {
"Binance": len(binance_data),
"Bybit": len(bybit_data),
"OKX": len(okx_data)
}
costs = collector.calculate_costs(data_points)
print(f"\nGesammelte Datenpunkte:")
for exchange, count in data_points.items():
print(f" {exchange}: {count}")
if __name__ == "__main__":
asyncio.run(main())
#!/usr/bin/env python3
"""
KI-gestützte Tick-Daten Analyse mit HolySheep AI
Analysiert Orderflow-Muster und generiert Strategie-Empfehlungen
"""
import aiohttp
import json
from datetime import datetime
from typing import List, Dict
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class HolySheepAnalysis:
"""KI-Analyse von Tick-Daten mit HolySheep API"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def analyze_orderflow(self, ticks: List[Dict], symbol: str) -> Dict:
"""
Analysiert Orderflow-Muster in Tick-Daten
Nutzt DeepSeek V3.2 für kostengünstige Analyse
"""
# Berechne Orderflow-Metriken
buy_volume = sum(t.get("quantity", t.get("volume", 0))
for t in ticks if t.get("is_buyer_maker") == False or t.get("side") == "buy")
sell_volume = sum(t.get("quantity", t.get("volume", 0))
for t in ticks if t.get("is_buyer_maker") == True or t.get("side") == "sell")
avg_price = sum(t.get("price", 0) for t in ticks) / len(ticks) if ticks else 0
price_volatility = max(t.get("price", 0) for t in ticks) - min(t.get("price", 0) for t in ticks) if ticks else 0
orderflow_ratio = buy_volume / sell_volume if sell_volume > 0 else 1
# Erstelle Analyse-Prompt
analysis_prompt = f"""
Analysiere folgende Orderflow-Daten für {symbol}:
Datenpunkte: {len(ticks)}
Kauf-Volumen: {buy_volume:.4f}
Verkauf-Volumen: {sell_volume:.4f}
Orderflow-Ratio: {orderflow_ratio:.4f}
Durchschnittspreis: {avg_price:.2f}
Preisvolatilität: {price_volatility:.2f}
Erkenne Muster:
1. Ist dies ein bullish oder bearish Signal?
2. Welche Strategieempfehlung (Mean Reversion, Momentum, Arbitrage)?
3. Risikobewertung (niedrig/mittel/hoch)
4. Präzise Stop-Loss und Take-Profit Empfehlungen
Antworte im JSON-Format mit keys: signal, strategy, risk_level, stop_loss, take_profit, confidence
"""
# Rufe HolySheep AI API auf
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "Du bist ein erfahrener quantitativer Trader mit Fokus auf Orderflow-Analyse."},
{"role": "user", "content": analysis_prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
async with aiohttp.ClientSession() as session:
try:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
result = await response.json()
ai_response = result["choices"][0]["message"]["content"]
return {
"success": True,
"raw_analysis": ai_response,
"metrics": {
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"orderflow_ratio": orderflow_ratio,
"avg_price": avg_price,
"volatility": price_volatility
}
}
else:
error_text = await response.text()
return {
"success": False,
"error": f"API Fehler {response.status}: {error_text}",
"fallback_metrics": {
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"orderflow_ratio": orderflow_ratio
}
}
except aiohttp.ClientError as e:
return {
"success": False,
"error": f"Verbindungsfehler: {str(e)}",
"metrics": {
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"orderflow_ratio": orderflow_ratio
}
}
async def batch_backtest_analysis(self, historical_data: List[Dict], symbol: str) -> Dict:
"""
Führt Batch-Backtesting-Analyse durch
Kostengünstig mit DeepSeek V3.2 ($0.42/MTok)
"""
# Bereite Daten für Batch-Analyse vor
trades_summary = []
for i in range(0, len(historical_data), 100):
batch = historical_data[i:i+100]
prices = [t["price"] for t in batch]
trades_summary.append({
"batch": i // 100 + 1,
"start_price": prices[0],
"end_price": prices[-1],
"high": max(prices),
"low": min(prices),
"volume": sum(t.get("quantity", t.get("volume", 0)) for t in batch)
})
# Erstelle Backtest-Zusammenfassung
prompt = f"""
Führe Backtesting-Analyse für {symbol} durch.
Historische Zusammenfassung ({len(trades_summary)} Batches):
{json.dumps(trades_summary[:10], indent=2)}
Berechne:
1. Gesamtrendite über den Zeitraum
2. Sharpe-Ratio (schätze basierend auf Volatilität)
3. Maximaler Drawdown
4. Win-Rate für verschiedene Strategien
5. Optimale Positionsgröße
Antworte strukturiert mit konkreten Zahlen.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 800
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
return {
"success": True,
"backtest_result": result["choices"][0]["message"]["content"],
"summary": trades_summary
}
else:
return {
"success": False,
"error": f"Backtest-Analyse fehlgeschlagen: {response.status}"
}
async def main():
"""Beispiel-Nutzung der HolySheep-Analyse"""
analyzer = HolySheepAnalysis()
# Simuliere Tick-Daten
sample_ticks = [
{"price": 67450.00, "quantity": 0.5, "is_buyer_maker": False},
{"price": 67452.50, "quantity": 0.3, "is_buyer_maker": True},
{"price": 67451.00, "quantity": 0.8, "is_buyer_maker": False},
{"price": 67448.00, "quantity": 0.4, "is_buyer_maker": True},
{"price": 67455.00, "quantity": 1.2, "is_buyer_maker": False},
]
print("Analysiere Orderflow mit HolySheep AI...")
result = await analyzer.analyze_orderflow(sample_ticks, "BTCUSDT")
if result["success"]:
print("\n✅ Analyse erfolgreich:")
print(f"Orderflow-Ratio: {result['metrics']['orderflow_ratio']:.4f}")
print(f"Kaufvolumen: {result['metrics']['buy_volume']:.4f}")
print(f"Verkaufsvolumen: {result['metrics']['sell_volume']:.4f}")
else:
print(f"\n❌ Analyse fehlgeschlagen: {result.get('error')}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Professionelle Trader mit monatlichem Budget ab $200
- Quantitative Entwickler die Mikrostruktur-Strategien entwickeln
- Hedgefonds mit Bedarf an Multi-Exchange-Daten
- KI-getriebene Trading-Bots die HolySheep für Analyse nutzen
- Arbitrage-Strategen die Latenz-Minimierung benötigen (Bybit: 45ms)
❌ Nicht geeignet für:
- Einzelhändler mit Budget unter $50/Monat
- Langfrist-Investoren die nur tägliche Daten benötigen
- Experimentelle Projekte ohne klare ROI-Strategie
- Regionen ohne stabile API-Zugänge zu diesen Börsen
Preise und ROI-Analyse
Die monatlichen Kosten für Tick-Daten variieren erheblich. Hier ist eine detaillierte ROI-Analyse für verschiedene Nutzungsszenarien:
| Nutzungsszenario | Binance | Bybit | OKX | Empfehlung |
|---|---|---|---|---|
| Indie-Entwickler (< 1M Requests/Monat) |
$299 | $199 | $249 | Bybit |
| Startup (1-5M Requests/Monat) |
$899 | $599 | $749 | Bybit |
| Enterprise (5M+ Requests/Monat) |
$2.499+ | $1.999+ | $2.249+ | Bybit Enterprise |
| Mit KI-Analyse (DeepSeek V3.2 Integration) |
$299 + $50 | $199 + $50 | $249 + $50 | Bybit + HolySheep |
Kostenlose Alternativen für Einsteiger
Falls Sie ein begrenztes Budget haben, bieten alle drei Börsen kostenlose REST-API-Endpunkte für begrenzte historische Daten:
- Binance: 1000 Trades pro Anfrage, 1200 Anfragen/Minute Limit
- Bybit: 100 Trades pro Anfrage, 600 Anfragen/Minute Limit
- OKX: 100 Trades pro Anfrage, 20 Anfragen/Minute Limit
Warum HolySheep AI für Ihre Analyse?
Die Beschaffung von Tick-Daten ist nur der erste Schritt. Die eigentliche Herausforderung liegt in der Interpretation und Mustererkennung. Hier kommt HolySheep AI ins Spiel:
| Feature | HolySheep AI | OpenAI GPT-4.1 | Anthropic Claude 4.5 |
|---|---|---|---|
| Preis pro 1M Tokens | $0.42 (DeepSeek V3.2) | $8.00 | $15.00 |
| Latenz | < 50ms | ~200ms | ~250ms |
| Bezahlung | WeChat, Alipay, USD | Nur Kreditkarte | Nur Kreditkarte |
| Startguthaben | Kostenlos | $5 | $5 |
| Wechselkurs | ¥1 = $1 | Standard-Kurse | Standard-Kurse |
Konkrete Ersparnis: Für eine typische Backtesting-Session mit 1M Token Input und 500K Token Output:
- Mit HolySheep (DeepSeek V3.2): $0.63
- Mit OpenAI GPT-4.1: $12.00
- Mit Anthropic Claude 4.5: $22.50
Ersparnis: 85-97%!
Häufige Fehler und Lösungen
Fehler 1: API-Rate-Limit überschritten
# FEHLER: Zu viele Anfragen in kurzer Zeit
Error: 429 Too Many Requests
LÖSUNG: Implementiere exponentielles Backoff mit Retry-Logik
import asyncio
import aiohttp
from datetime import datetime, timedelta
class RateLimitedClient:
"""API-Client mit automatischer Rate-Limit-Behandlung"""
def __init__(self, base_url: str, requests_per_minute: int = 600):
self.base_url = base_url
self.request_interval = 60 / requests_per_minute
self.last_request = datetime.min
self.retry_delays = [1, 2, 4, 8, 16, 32] # Sekunden
async def get_with_retry(self, url: str, max_retries: int = 3) -> dict:
"""Führt GET-Request mit automatischem Retry aus"""
for attempt, delay in enumerate(self.retry_delays[:max_retries]):
# Warte auf Mindestabstand zwischen Requests
time_since_last = (datetime.now() - self.last_request).total_seconds()
if time_since_last < self.request_interval:
await asyncio.sleep(self.request_interval - time_since_last)
async with aiohttp.ClientSession() as session:
try:
async with session.get(url) as response:
self.last_request = datetime.now()
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate Limit erreicht - warte und retry
retry_after = response.headers.get('Retry-After', delay)
print(f"Rate Limit erreicht. Warte {retry_after}s...")
await asyncio.sleep(int(retry_after))
else:
print(f"HTTP Fehler {response.status}")
return None
except aiohttp.ClientError as e:
print(f"Verbindungsfehler: {e}. Retry in {delay}s...")
await asyncio.sleep(delay)
print("Max. Retry-Versuche erreicht.")
return None
Verwendung für Bybit
bybit_client = RateLimitedClient(
"https://api.bybit.com",
requests_per_minute=600
)
Fehler 2: Zeitstempel-Konvertierungsprobleme zwischen Börsen
# FEHLER: Inkonsistente Zeitstempel-Formate
Binance: Millisekunden (1683225600000)
Bybit: Millisekunden (1683225600000)
OKX: Millisekunden als String ("1683225600000")
LÖSUNG: Normalisiere alle Zeitstempel zu Unix-Timestamps
from datetime import datetime
from typing import Union
def normalize_timestamp(ts: Union[int, str, float]) -> int:
"""
Normalisiert verschiedene Zeitstempel-Formate zu Unix-Millisekunden
"""
if isinstance(ts, str):
ts = int(ts)
if isinstance(ts, float):
ts = int(ts)
# Prüfe ob es Sekunden oder Millisekunden sind
if ts < 10_000_000_000: # Sekunden
ts = ts * 1000
return ts
def ts_to_datetime(ts: Union[int, str]) -> datetime:
"""Konvertiert Unix-Timestamp zu datetime Objekt"""
normalized = normalize_timestamp(ts)
return datetime.fromtimestamp(normalized / 1000)
def datetime_to_ts(dt: datetime) -> int:
"""Konvertiert datetime zu Unix-Timestamp in Millisekunden"""
return int(dt.timestamp() * 1000)
Test mit verschiedenen Formaten
test_cases = [
1683225600000, # Binance Integer
"1683225600000", # OKX String
1683225600.0, # Sekunden als Float
]
for ts in test_cases:
normalized = normalize_timestamp(ts)
dt = ts_to_datetime(ts)
print(f"{ts} -> {normalized} -> {dt}")
# Ausgabe: Alle werden korrekt zu 2023-05-04 16:00:00 konvertiert
Fehler 3: Fehlende Fehlerbehandlung bei WebSocket-Verbindung
# FEHLER: WebSocket-Verbindung bricht ab ohne Reconnection
Symptom: Datenlücken im Backtest, fehlende Trades
LÖSUNG: Implementiere automatische Reconnection mit Heartbeat
import asyncio
import websockets
import json
from datetime import datetime
from typing import Callable, Optional
class WebSocketCollector:
"""WebSocket-Client mit automatischer Reconnection"""
def __init__(self, symbol: str, exchange: str):
self.symbol = symbol
self.exchange = exchange
self.running = False
self.reconnect_delay = 1 # Start-Delay in Sekunden
self.max_reconnect_delay = 60
self.heartbeat_interval = 30
self.last_ping = datetime.now()
self.data_buffer = []
async def connect_binance(self):
"""Verbindet zu Binance WebSocket für Trade-Stream"""
uri = f"wss://stream.binance.com:9443/ws/{self.symbol.lower()}@trade"
while self.running:
try:
async with websockets.connect(uri) as ws:
self.reconnect_delay = 1 # Reset bei erfolgreicher Verbindung
print(f"Binance WebSocket verbunden für {self.symbol}")
while self.running:
try:
# Warte auf Nachrichten mit Timeout
message = await asyncio.wait_for(
ws.recv(),
timeout=self.heartbeat_interval + 5
)
data = json.loads(message)
self.last_ping = datetime.now()
self.process_trade(data)
except asyncio.TimeoutError:
# Heartbeat-Timeout - sende Ping
await ws.ping()
print("Heartbeat gesendet")
if (datetime.now() - self.last_ping).seconds > self.heartbeat_interval * 3:
print("Verbindung scheint tot, reconnect...")
break
except websockets.ConnectionClosed as e:
print(f"Verbindung verloren: {e}")
except Exception as e:
print(f"WebSocket Fehler: {e}")
# Exponentielles Backoff für Reconnection
if self.running:
print(f"Reconnect in {self.reconnect_delay}s...")
await asyncio.sleep(self.reconnect_delay)
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
def process_trade(self, data: dict):
"""Verarbeitet eingehende Trade-Daten"""
trade = {
"symbol": data["s"],
"price": float(data["p"]),
"quantity": float(data["q"]),
"timestamp": data["T"],
"is_buyer_maker": data["m"],
"exchange": "binance",
"received_at": datetime.now().isoformat()
}
self.data_buffer.append(trade)
# Alle 1000 Trades: Speichere in Datenbank
if len(self.data_buffer) >= 1000:
self.flush_buffer()
def flush_buffer(self):
"""Schreibt buffered Daten in Datenbank"""
if self.data_buffer:
# Hier Datenbank-Insert implementieren
print(f"Schreibe {len(self.data_buffer)} Trades in Datenbank")
self.data_buffer.clear()
async def start(self):
"""Startet den WebSocket-Collector"""
Verwandte Ressourcen
Verwandte Artikel