OKX L2增量快照历史数据处理：Tardis CSV-Format-Parsing und Orderbook-Rekonstruktion

Orderbook-Rekonstruktion: Delta-Updates effizient verarbeiten

Die Kernherausforderung bei L2-Inkremental-Snapshots ist die korrekte Rekonstruktion des Orderbook-Zustands:

from sortedcontainers import SortedDict
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import time

@dataclass
class OrderbookLevel:
    """Eine Preisstufe im Orderbook."""
    price: float
    quantity: float
    orders: int = 1  # Anzahl der Orders auf diesem Level

class ReconstructedOrderbook:
    """
    Rekonstruiert vollständige Orderbooks aus inkrementellen L2-Updates.
    Verwendet SortedDict für effiziente O(log n) Operationen.
    """
    
    def __init__(self, max_depth: int = 400):
        self.max_depth = max_depth
        self.bids: SortedDict = SortedDict()  # price -> OrderbookLevel
        self.asks: SortedDict = SortedDict()
        self.last_update_id: int = 0
        self.sequence_gaps: List[Tuple[int, int]] = []
    
    def apply_snapshot(self, snapshot: Dict):
        """
        Wendet einen vollständigen Snapshot an.
        Setzt den Orderbook-State komplett zurück.
        """
        self.bids.clear()
        self.asks.clear()
        
        for price, qty in snapshot.get('bids', []):
            self.bids[float(price)] = OrderbookLevel(float(price), float(qty))
        
        for price, qty in snapshot.get('asks', []):
            self.asks[float(price)] = OrderbookLevel(float(price), float(qty))
        
        self.last_update_id = snapshot.get('update_id', 0)
    
    def apply_delta(self, delta: Dict) -> bool:
        """
        Wendet inkrementelle Updates an.
        
        Returns:
            True wenn Update angewendet, False bei Sequence-Error
        """
        update_id = delta.get('update_id', 0)
        
        # Prüfe auf Sequence-Lücken
        if update_id <= self.last_update_id:
            return False  # Veraltetes Update
        
        if update_id > self.last_update_id + 1:
            # Sequence-Lücke erkannt
            self.sequence_gaps.append((self.last_update_id, update_id))
            print(f"Warning: Sequence-Lücke erkannt: {self.last_update_id} -> {update_id}")
        
        # Verarbeite Bids
        for bid in delta.get('bids', []):
            price, qty = float(bid[0]), float(bid[1])
            if qty == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = OrderbookLevel(price, qty)
        
        # Verarbeite Asks
        for ask in delta.get('asks', []):
            price, qty = float(ask[0]), float(ask[1])
            if qty == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = OrderbookLevel(price, qty)
        
        self.last_update_id = update_id
        return True
    
    def get_best_bid_ask(self) -> Tuple[Optional[float], Optional[float]]:
        """Gibt aktuellen Bid/Ask zurück."""
        bid_price = self.bids.peekitem(0)[0] if self.bids else None
        ask_price = self.asks.peekitem(0)[0] if self.asks else None
        return bid_price, ask_price
    
    def get_spread(self) -> Optional[float]:
        """Berechnet aktuellen Spread."""
        bid, ask = self.get_best_bid_ask()
        if bid and ask:
            return ask - bid
        return None
    
    def get_mid_price(self) -> Optional[float]:
        """Berechnet Mittelkurs."""
        bid, ask = self.get_best_bid_ask()
        if bid and ask:
            return (bid + ask) / 2
        return None
    
    def get_depth(self, levels: int = 10) -> Dict:
        """Gibt Top-N-Levels zurück."""
        top_bids = [
            {'price': price, 'qty': level.quantity}
            for price, level in list(self.bids.items())[:levels]
        ]
        top_asks = [
            {'price': price, 'qty': level.quantity}
            for price, level in list(self.asks.items())[:levels]
        ]
        return {'bids': top_bids, 'asks': top_asks}
    
    def calculate_imbalance(self) -> float:
        """
        Berechnet Orderbook-Imbalance.
        Positiv = mehr Bid-Volume, Negativ = mehr Ask-Volume
        """
        bid_volume = sum(level.quantity for level in self.bids.values())
        ask_volume = sum(level.quantity for level in self.asks.values())
        
        if bid_volume + ask_volume == 0:
            return 0
        
        return (bid_volume - ask_volume) / (bid_volume + ask_volume)


HolySheep: Echtzeit-Updates empfangen
class HolySheepWebSocketHandler:
    """
    Empfängt Echtzeit-Orderbook-Updates über HolySheep WebSocket.
    Latenz: <50ms durch optimierte Infrastruktur.
    """
    
    WS_URL = "wss://api.holysheep.ai/v1/market/okx/ws"
    
    def __init__(self, api_key: str, symbols: List[str]):
        self.api_key = api_key
        self.symbols = symbols
        self.orderbooks: Dict[str, ReconstructedOrderbook] = {
            sym: ReconstructedOrderbook() for sym in symbols
        }
        self.callbacks: List[callable] = []
        self.reconnect_attempts = 0
        self.max_reconnect = 5
    
    async def connect(self):
        """Stellt WebSocket-Verbindung her."""
        import websockets
        
        while self.reconnect_attempts < self.max_reconnect:
            try:
                async with websockets.connect(
                    self.WS_URL,
                    extra_headers={"Authorization": f"Bearer {self.api_key}"}
                ) as ws:
                    # Subscribe zu Orderbook-Feeds
                    subscribe_msg = {
                        "type": "subscribe",
                        "channels": ["orderbook"],
                        "symbols": self.symbols
                    }
                    await ws.send(json.dumps(subscribe_msg))
                    
                    # Nachrichtenschleife
                    async for message in ws:
                        await self._process_message(message)
            
            except websockets.ConnectionClosed:
                self.reconnect_attempts += 1
                wait_time = 2 ** self.reconnect_attempts
                print(f"Verbindung verloren. Retry in {wait_time}s...")
                await asyncio.sleep(wait_time)
    
    async def _process_message(self, message: str):
        """Verarbeitet eingehende WebSocket-Nachrichten."""
        import json
        data = json.loads(message)
        
        if data.get('type') == 'snapshot':
            for sym in self.symbols:
                self.orderbooks[sym].apply_snapshot(data[sym])
        
        elif data.get('type') == 'delta':
            for sym, delta in data.get('updates', {}).items():
                self.orderbooks[sym].apply_delta(delta)
        
        # Benachrichtige Callbacks
        for callback in self.callbacks:
            await callback(data)

# ❌ FALSCH: Updates ohne Prüfung anwenden
async def wrong_handler(message):
    delta = json.loads(message)
    orderbook.apply_delta(delta)  # Kann zu Inkonsistenzen führen!

✅ RICHTIG: Full-Snapshot nach Reconnect anfordern
class RobustWebSocketClient:
    async def on_reconnect(self):
        """Fordert Full-Snapshot nach Reconnect an."""
        await self.send({
            "type": "request_snapshot",
            "symbol": self.symbol
        })
        # Warte auf Snapshot, bevor Deltas verarbeitet werden
        await self.wait_for_snapshot()
    
    async def wait_for_snapshot(self, timeout: int = 5):
        """Wartet auf Full-Snapshot oder Timeout."""
        start = time.time()
        while time.time() - start < timeout:
            if self.last_snapshot_id > 0:
                return True
            await asyncio.sleep(0.1)
        raise TimeoutError("Snapshot nicht innerhalb des Timeouts erhalten")

# ❌ FALSCH: Annahme fester Timestamp-Format
def parse_timestamp_wrong(ts: str) -> int:
    return int(ts)  # Funktioniert nur bei einem Format!

✅ RICHTIG: Automatische Format-Erkennung
def parse_timestamp_auto(ts: str) -> int:
    """Erkennt Timestamp-Format automatisch."""
    try:
        value = float(ts)
        
        if value > 1e15:  # Nanosekunden
            return int(value / 1_000_000)
        elif value > 1e12:  # Millisekunden
            return int(value)
        elif value > 1e9:  # Sekunden
            return int(value * 1000)
        else:
            raise ValueError(f"Unbekanntes Timestamp-Format: {ts}")
    except ValueError as e:
        # Fallback: Versuche ISO-Format
        try:
            from datetime import datetime
            dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
            return int(dt.timestamp() * 1000)
        except:
            raise ValueError(f"Konnte Timestamp nicht parsen: {ts}")

# ❌ FALSCH: Keine Fehlerbehandlung
async def fetch_data_naive(url):
    async with session.get(url) as response:
        return await response.json()  # Wirft bei 429!

✅ RICHTIG: Exponential Backoff mit Retry
import random

class RateLimitHandler:
    def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def fetch_with_retry(self, session, url: str) -> dict:
        """Holt Daten mit exponentiellem Backoff bei Rate-Limits."""
        for attempt in range(self.max_retries):
            try:
                async with session.get(url) as response:
                    if response.status == 200:
                        return await response.json()
                    elif response.status == 429:
                        # Rate-Limit erreicht: Warte mit Jitter
                        retry_after = response.headers.get('Retry-After', '60')
                        delay = float(retry_after) + random.uniform(0, 1)
                        
                        print(f"Rate-Limit. Warte {delay:.1f}s...")
                        await asyncio.sleep(delay)
                    else:
                        raise ConnectionError(f"HTTP {response.status}")
            
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == self.max_retries - 1:
                    raise
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Verbindungsfehler. Retry in {delay:.1f}s: {e}")
                await asyncio.sleep(delay)
        
        raise RuntimeError(f"Max retries ({self.max_retries}) erreicht")

# ❌ FALSCH: Annahme, dass Preisstufe existiert
def update_level_unsafe(ob, price, qty):
    if qty == 0:
        del ob.bids[price]  # KeyError möglich!

✅ RICHTIG: Safe Delete mit Graceful Handling
def update_level_safe(ob, price: float, qty: float, side: str):
    """Aktualisiert Orderbook-Level sicher."""
    book = ob.bids if side == 'bid' else ob.asks
    
    if qty == 0:
        # Entferne Level mit pop() statt del
        removed = book.pop(price, None)
        if removed is None:
            print(f"Warning: Level {price} auf {side} existierte nicht")
    else:
        # Füge hinzu oder aktualisiere
        book[price] = OrderbookLevel(price, qty)