Willkommen zu meinem Praxistest für Order Book 深度数据实时处理 mit Python. In diesem Tutorial zeige ich Ihnen, wie Sie Finanzmarktdaten in Echtzeit verarbeiten, analysieren und für Trading-Strategien nutzen können. Wir werden gemeinsam die technischen Herausforderungen der Tiefe-Datenverarbeitung meistern und dabei praktische Lösungen mit HolySheep AI als Backend für KI-gestützte Analysen implementieren.

Was ist ein Order Book?

Das Order Book (Orderbuch) ist die zentrale Datenstruktur im Börsenhandel. Es enthält alle offenen Kauf- und Verkaufsorders für ein bestimmtes Wertpapier, geordnet nach Preisniveau. Die Tiefe (Depth) eines Order Books gibt an, wie viele Preisstufen vom besten Gebot/Verkauf entfernt aggregiert werden.

Praxistest: Order Book Datenverarbeitung

Testumgebung und Kriterien

Ich habe diesen Praxistest mit folgender Konfiguration durchgeführt:

Latenzmessung

Die Latenz ist bei der Order Book Verarbeitung kritisch. Ich habe folgende durchschnittliche Latenzen gemessen:

Python实战代码：Grundstruktur

Beginnen wir mit der fundamentalen Order Book Klasse, die wir für die Echtzeitverarbeitung benötigen:

import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import defaultdict
import websockets
from datetime import datetime

@dataclass
class OrderBookEntry:
    """Einzelner Order Book Eintrag"""
    price: float
    quantity: float
    timestamp: float = field(default_factory=time.time)
    
    def __repr__(self):
        return f"Price: {self.price}, Qty: {self.quantity}"

@dataclass
class OrderBook:
    """Kompletter Order Book mit Bids und Asks"""
    symbol: str
    bids: Dict[float, float] = field(default_factory=dict)  # price -> quantity
    asks: Dict[float, float] = field(default_factory=dict)
    last_update: float = field(default_factory=time.time)
    update_count: int = 0
    
    def update_bid(self, price: float, quantity: float):
        """Aktualisiert ein Gebot"""
        if quantity == 0:
            self.bids.pop(price, None)
        else:
            self.bids[price] = quantity
        self.last_update = time.time()
        self.update_count += 1
    
    def update_ask(self, price: float, quantity: float):
        """Aktualisiert ein Verkaufsangebot"""
        if quantity == 0:
            self.asks.pop(price, None)
        else:
            self.asks[price] = quantity
        self.last_update = time.time()
        self.update_count += 1
    
    def get_spread(self) -> float:
        """Berechnet den Spread zwischen bestem Bid und Ask"""
        best_bid = max(self.bids.keys()) if self.bids else 0
        best_ask = min(self.asks.keys()) if self.asks else float('inf')
        return best_ask - best_bid
    
    def get_mid_price(self) -> float:
        """Berechnet den Mittelpreis"""
        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
    
    def get_depth(self, levels: int = 10) -> Dict:
        """Gibt die Tiefe des Order Books zurück"""
        sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:levels]
        sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
        
        return {
            'symbol': self.symbol,
            'bids': [{'price': p, 'quantity': q} for p, q in sorted_bids],
            'asks': [{'price': p, 'quantity': q} for p, q in sorted_asks],
            'spread': self.get_spread(),
            'mid_price': self.get_mid_price(),
            'total_bid_volume': sum(q for _, q in sorted_bids),
            'total_ask_volume': sum(q for _, q in sorted_asks)
        }


Beispiel-Nutzung

ob = OrderBook(symbol="BTC/USDT")
ob.update_bid(42150.0, 0.5)
ob.update_bid(42145.0, 1.2)
ob.update_ask(42155.0, 0.8)
print(ob.get_depth())

WebSocket-Verbindung für Echtzeit-Updates

Die echte Herausforderung beginnt mit der WebSocket-Verbindung zu Börsen. Hier ist meine实战implementierung:

import asyncio
import websockets
import json
import zlib
from typing import Callable, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OrderBookWebSocketClient:
    """WebSocket Client für Order Book Echtzeit-Updates"""
    
    def __init__(self, symbol: str, exchange: str = "binance"):
        self.symbol = symbol.lower().replace("/", "")
        self.exchange = exchange
        self.order_book = OrderBook(symbol=symbol)
        self.running = False
        self.message_count = 0
        self.error_count = 0
        self.latencies = []
        
        # Exchange-spezifische WebSocket URLs
        self.ws_urls = {
            "binance": "wss://stream.binance.com:9443/ws",
            "coinbase": "wss://ws-feed.exchange.coinbase.com",
            "kraken": "wss://ws.kraken.com"
        }
    
    def get_subscribe_message(self) -> dict:
        """Erstellt das passende Abonnement-Message für den Exchange"""
        if self.exchange == "binance":
            return {
                "method": "SUBSCRIBE",
                "params": [f"{self.symbol}@depth@100ms"],
                "id": 1
            }
        elif self.exchange == "coinbase":
            return {
                "type": "subscribe",
                "product_ids": [self.symbol.replace("btc", "BTC-")],
                "channels": ["level2_batch"]
            }
        elif self.exchange == "kraken":
            return {

Verwandte Ressourcen
Verwandte Artikel