Der Handel mit Kryptowährungen über mehrere Börsen hinweg gehört zu den anspruchsvollsten Aufgaben im quantitativen Trading. Die Vielfalt der APIs, unterschiedliche Rate-Limits und inkonsistente Datenformate machen die Datenaufbereitung zum kritischen Engpass. In diesem Tutorial zeige ich Ihnen, wie Sie mit Tardis eine robuste Datenpipeline aufbauen und dabei die häufigsten API-Probleme meistern.

Das Problem: Warum Börsen-Daten so schwer zu handhaben sind

Nach meiner Praxiserfahrung in der Entwicklung von Trading-Systemen für Hedgefonds kann ich bestätigen: Die größten Herausforderungen liegen nicht im Algorithmus selbst, sondern in der Datenqualität. Ich habe Systeme gesehen, die theoretical Renditen von 300% p.a. versprachen, aber an der Realität der Börsen-API scheiterten.

Die Kernprobleme im Überblick:

HolySheep vs. Offizielle API vs. Andere Relay-Dienste

Feature HolySheep AI Offizielle Börsen-API Andere Relay-Dienste
Latenz <50ms 20-200ms 80-300ms
Kosten ¥1/$1 (85%+ Ersparnis) Kostenlos (mit Limits) $5-50/Monat
Zahlungsmethoden WeChat, Alipay, Kreditkarte Nur Krypto Kreditkarte/PayPal
Historische Daten Bis zu 5 Jahre verfügbar Max. 30 Tage Variabel (max. 1 Jahr)
Rate-Limit-Handling Automatisch optimiert Manuelle Implementierung Teilweise automatisch
Multi-Exchange Support 15+ Börsen vereinheitlicht 1 Börse pro API 5-10 Börsen
Kostenlose Credits Ja, bei Registrierung Nein Selten

Geeignet / Nicht geeignet für

Geeignet für:

Nicht geeignet für:

Preise und ROI

Die HolySheep AI Preisstruktur (Stand 2026) bietet außergewöhnliche Kosteneffizienz:

ROI-Analyse: Bei einem typischen Trading-System, das 10M Tokens/Monat für Datenanalyse verbraucht, sparen Sie mit HolySheep gegenüber OpenAI Direct ca. $720/Monat (85% Ersparnis). Die kostenlosen Credits bei der Registrierung ermöglichen sofortige Tests ohne initiale Investition.

Warum HolySheep wählen

Nach Jahren der Arbeit mit verschiedenen Datenanbietern hat sich HolySheep AI als optimale Lösung für unsere Bedürfnisse herauskristallisiert. Die Kombination aus <50ms Latenz, 85%+ Kostenersparnis und multinationalen Zahlungsoptionen (WeChat, Alipay) macht es zum idealen Partner für asiatische und globale Trading-Operationen.

Die automatische Retry-Logik, das intelligente Rate-Limit-Management und die einheitliche Datenstruktur über alle unterstützten Börsen reduzieren die Entwicklungszeit um schätzungsweise 60% im Vergleich zur Eigenentwicklung.

Tardis安装与基础配置

# Installation via npm
npm install tardis-dev


Oder via Python (für Data Science Setups)

pip install tardis-client


Grundlegende TypeScript-Konfiguration

import { TardisClient, RealtimeSubscription } from 'tardis-dev';

const client = new TardisClient({
    exchange: 'binance',
    apiKey: process.env.TARDIS_API_KEY,
    // Für HolySheep AI Integration:
    // baseUrl: 'https://api.holysheep.ai/v1'
});


Python-Beispiel für HolySheep AI Datenanalyse

from openai import OpenAI

client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

response = client.chat.completions.create(
    model="deepseek-v3",
    messages=[
        {"role": "system", "content": "Analysiere Börsen-Daten auf Anomalien"},
        {"role": "user", "content": "Prüfe folgende Trades auf ungewöhnliche Muster..."}
    ]
)
print(response.choices[0].message.content)

处理交易所API异常：完整数据清洗流程

# tardis-data-cleaner.ts - Vollständige Datenreinigung mit Tardis

import { TardisClient } from 'tardis-dev';
import { OpenAI } from 'openai';

interface Trade {
    id: string;
    timestamp: number;
    price: number;
    amount: number;
    side: 'buy' | 'sell';
    exchange: string;
}

interface CleanedTrade extends Trade {
    normalizedTimestamp: number;
    isAnomaly: boolean;
    anomalyReason?: string;
}

class TardisDataCleaner {
    private client: TardisClient;
    private aiClient: OpenAI;
    private buffer: Map = new Map();
    private anomalyThreshold = 0.05; // 5% Preisabweichung

    constructor(tardisKey: string, holysheepKey: string) {
        this.client = new TardisClient({
            apiKey: tardisKey,
            exchange: ['binance', 'coinbase', 'kraken']
        });
        
        this.aiClient = new OpenAI({
            api_key: holysheepKey,
            base_url: 'https://api.holysheep.ai/v1'
        });
    }

    async processTrades(symbol: string): Promise<CleanedTrade[]> {
        const trades: Trade[] = [];
        
        // 1. Historische Daten abrufen mit Retry-Logik
        const historicalTrades = await this.fetchWithRetry(
            () => this.client.getHistoricalTrades({ symbol }),
            3, // Max retries
            1000 // Initial delay ms
        );
        
        trades.push(...historicalTrades);

        // 2. Real-time Stream mit auto-reconnect
        await this.subscribeRealtime(symbol, (trade) => {
            this.buffer.set(symbol, [...(this.buffer.get(symbol) || []), trade]);
        });

        // 3. Batch-Verarbeitung
        const bufferedTrades = this.buffer.get(symbol) || [];
        const allTrades = [...trades, ...bufferedTrades];

        // 4. Normalisierung
        const normalizedTrades = this.normalizeTimestamps(allTrades);

        // 5. Anomalie-Erkennung mit KI
        const cleanedTrades = await this.detectAnomalies(normalizedTrades);

        return cleanedTrades;
    }

    private async fetchWithRetry<T>(
        fn: () => Promise<T>,
        retries: number,
        delay: number
    ): Promise<T> {
        for (let i = 0; i < retries; i++) {
            try {
                return await fn();
            } catch (error: any) {
                if (error.status === 429) { // Rate limit
                    await this.sleep(delay * Math.pow(2, i));
                    continue;
                }
                if (error.status === 500 || error.status === 502) { // Server error
                    await this.sleep(delay * Math.pow(2, i));
                    continue;
                }
                throw error;
            }
        }
        throw new Error(Failed after ${retries} retries);
    }

    private async subscribeRealtime(
        symbol: string,
        onTrade: (trade: Trade) => void
    ): Promise<void> {
        let reconnectAttempts = 0;
        const maxReconnects = 10;

        const subscribe = async () => {
            try {
                await this.client.subscribe({
                    channel: 'trades',
                    symbol
                }, (trade) => {
                    reconnectAttempts = 0; // Reset counter on success
                    onTrade(trade);
                });
            } catch (error: any) {
                if (reconnectAttempts < maxReconnects) {
                    reconnectAttempts++;
                    await this.sleep(1000 * reconnectAttempts);
                    await subscribe();
                } else {
                    console.error(Max reconnects reached for ${symbol});
                }
            }
        };

        await subscribe();
    }

    private normalizeTimestamps(trades: Trade[]): Trade[] {
        return trades.map(trade => ({
            ...trade,
            // Konvertiere zu Unix Millisekunden (UTC)
            timestamp: trade.timestamp < 1e12 
                ? trade.timestamp * 1000 
                : trade.timestamp
        }));
    }

    private async detectAnomalies(trades: Trade[]): Promise<CleanedTrade[]> {
        // Gruppiere nach Zeitfenster (1 Minute)
        const priceMap = new Map<number, number[]>();
        
        trades.forEach(trade => {
            const window = Math.floor(trade.timestamp / 60000) * 60000;
            if (!priceMap.has(window)) priceMap.set(window, []);
            priceMap.get(window)!.push(trade.price);
        });

        // Berechne Statistiken
        const stats = new Map<number, { mean: number; std: number }>();
        priceMap.forEach((prices, window) => {
            const mean = prices.reduce((a, b) => a + b) / prices.length;
            const variance = prices.reduce((a, b) => a + Math.pow(b - mean, 2), 0) / prices.length;
            stats.set(window, { mean, std: Math.sqrt(variance) });
        });

        // Markiere Anomalien
        return trades.map(trade => {
            const window = Math.floor(trade.timestamp / 60000) * 60000;
            const stat = stats.get(window);
            
            if (!stat) return { ...trade, normalizedTimestamp: trade.timestamp, isAnomaly: false };

            const deviation = Math.abs(trade.price - stat.mean) / stat.mean;
            const isAnomaly = deviation > this.anomalyThreshold;

            return {
                ...trade,
                normalizedTimestamp: trade.timestamp,
                isAnomaly,
                anomalyReason: isAnomaly 
                    ? Preisabweichung ${(deviation * 100).toFixed(2)}% vom Durchschnitt
                    : undefined
            };
        });
    }

    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

// Verwendung
const cleaner = new TardisDataCleaner(
    process.env.TARDIS_API_KEY!,
    "YOUR_HOLYSHEEP_API_KEY" // HolySheep AI Key
);

const cleanedData = await cleaner.processTrades('BTC-USDT');
console.log(Verarbeitet: ${cleanedData.length} Trades);
console.log(Anomalien: ${cleanedData.filter(t => t.isAnomaly).length});

处理缺失数据：Interpolation und Forward-Fill Strategien

# data-interpolation.py - Fortgeschrittene Datenlückenschließung

import pandas as pd
import numpy as np
from typing import List, Dict, Optional
from dataclasses import dataclass
import asyncio

@dataclass
class OHLCV:
    timestamp: int
    open: float
    high: float
    low: float
    close: float
    volume: float

class DataGapHandler:
    """
    Behandelt fehlende Datenpunkte in Börsen-Ohlcv-Daten
    mit intelligenten Interpolationsstrategien
    """
    
    def __init__(self, max_gap_ms: int = 3600000):  # Max 1 Stunde
        self.max_gap = max_gap_ms
    
    def process_ohlcv_series(self, data: List[OHLCV]) -> List[OHLCV]:
        """Hauptmethode zur Lückenschließung"""
        if not data:
            return []
        
        df = pd.DataFrame([{
            'timestamp': d.timestamp,
            'open': d.open,
            'high': d.high,
            'low': d.low,
            'close': d.close,
            'volume': d.volume
        } for d in data])
        
        # 1. Identifiziere Lücken
        gaps = self.identify_gaps(df)
        
        # 2. Kleinere Lücken: Interpolation
        small_gaps = gaps[gaps['gap_ms'] <= self.max_gap / 10]
        df = self.interpolate_small_gaps(df, small_gaps)
        
        # 3. Mittlere Lücken: Forward-Fill mit Validierung
        medium_gaps = gaps[(gaps['gap_ms'] > self.max_gap / 10) & 
                          (gaps['gap_ms'] <= self.max_gap)]
        df = self.forward_fill_medium_gaps(df, medium_gaps)
        
        # 4. Große Lücken: Markieren für manuelle Prüfung
        large_gaps = gaps[gaps['gap_ms'] > self.max_gap]
        df = self.mark_large_gaps(df, large_gaps)
        
        return [OHLCV(**row) for row in df.to_dict('records')]
    
    def identify_gaps(self, df: pd.DataFrame) -> pd.DataFrame:
        """Identifiziert alle Lücken im Datensatz"""
        df = df.sort_values('timestamp')
        timestamps = df['timestamp'].values
        
        gaps = []
        for i in range(1, len(timestamps)):
            gap_ms = timestamps[i] - timestamps[i-1]
            if gap_ms > 0:  # Nur positive Gaps (keine Duplikate)
                gaps.append({
                    'start_idx': i-1,
                    'end_idx': i,
                    'gap_ms': gap_ms,
                    'start_ts': timestamps[i-1],
                    'end_ts': timestamps[i]
                })
        
        return pd.DataFrame(gaps)
    
    def interpolate_small_gaps(
        self, 
        df: pd.DataFrame, 
        gaps: pd.DataFrame
    ) -> pd.DataFrame:
        """Lineare Interpolation für kleine Lücken (<6 Minuten)"""
        if gaps.empty:
            return df
        
        for _, gap in gaps.iterrows():
            start_idx = gap['start_idx']
            end_idx = gap['end_idx']
            
            start_row = df.iloc[start_idx]
            end_row = df.iloc[end_idx]
            
            # Berechne Interpolationsfaktor
            total_gap = gap['gap_ms']
            
            # Füge interpolierte Zeilen ein
            num_new_rows = max(1, int(total_gap / 60000) - 1)
            
            new_rows = []
            for j in range(1, num_new_rows + 1):
                factor = j / (num_new_rows + 1)
                
                interpolated = {
                    'timestamp': int(start_row['timestamp'] + total_gap * factor),
                    'open': self._interpolate(start_row['open'], end_row['open'], factor),
                    'high': self._interpolate(start_row['high'], end_row['high'], factor),
                    'low': self._interpolate(start_row['low'], end_row['low'], factor),
                    'close': self._interpolate(start_row['close'], end_row['close'], factor),
                    'volume': 0,  # Volume = 0 bei interpolierten Punkten
                    'is_interpolated': True
                }
                new_rows.append(interpolated)
            
            # Füge neue Zeilen ein
            df = pd.concat([
                df.iloc[:end_idx],
                pd.DataFrame(new_rows),
                df.iloc[end_idx:]
            ], ignore_index=True)
        
        return df
    
    def _interpolate(self, start: float, end: float, factor: float) -> float:
        """Lineare Interpolation zwischen zwei Werten"""
        return start + (end - start) * factor
    
    def forward_fill_medium_gaps(
        self,
        df: pd.DataFrame,
        gaps: pd.DataFrame
    ) -> pd.DataFrame:
        """
        Forward-Fill für mittlere Lücken mit Volumen-Reduzierung
        um die Datenlücke zu markieren
        """
        for _, gap in gaps.iterrows():
            start_idx = gap['start_idx']
            end_idx = gap['end_idx']
            
            # Volume der Fill-Zeilen reduzieren
            for idx in range(start_idx + 1, end_idx):
                if f'is_interpolated' not in df.columns or \
                   not df.iloc[idx].get('is_interpolated', False):
                    df.at[df.index[idx], 'volume'] *= 0.3  # 70% Reduzierung
                    df.at[df.index[idx], 'is_forward_fill'] = True
        
        return df
    
    def mark_large_gaps(
        self,
        df: pd.DataFrame,
        gaps: pd.DataFrame
    ) -> pd.DataFrame:
        """Markiert große Lücken für spätere Analyse/Manuelle Korrektur"""
        df['has_large_gap'] = False
        df['gap_marker'] = None
        
        for _, gap in gaps.iterrows():
            # Markiere den Bereich zwischen den Lücken
            df.loc[
                (df['timestamp'] >= gap['start_ts']) & 
                (df['timestamp'] <= gap['end_ts']),
                'has_large_gap'
            ] = True
            
            df.loc[
                (df['timestamp'] >= gap['start_ts']) & 
                (df['timestamp'] <= gap['end_ts']),
                'gap_marker'
            ] = f"Gap: {gap['gap_ms']/3600000:.1f}h"
        
        return df



Async Wrapper für HolySheep AI Anomalie-Erkennung

class AIEnhancedGapHandler(DataGapHandler):
    """Erweitert den Gap Handler mit KI-gestützter Anomalie-Erkennung"""
    
    def __init__(self, holysheep_api_key: str):
        super().__init__()
        self.client = OpenAI(
            api_key=holysheep_api_key,
            base_url="https://api.holysheep.ai/v1"
        )
    
    async def analyze_large_gaps_ai(
        self, 
        df: pd.DataFrame,
        symbol: str
    ) -> Dict[str, any]:
        """Nutzt HolySheep AI um große Datenlücken zu analysieren"""
        
        large_gap_data = df[df['has_large_gap'] == True]
        if large_gap_data.empty:
            return {"status": "no_gaps", "recommendations": []}
        
        # Erstelle Zusammenfassung für die KI
        summary = self._create_gap_summary(large_gap_data)
        
        prompt = f"""
Analysiere folgende Datenlücken für {symbol}:

{summary}

Identifiziere mögliche Ursachen:
1. Börsen-spezifische Outages
2. Netzwerkprobleme
3. Api-Limit-Erschöpfung
4. Vorbörsliche/vorfeiertägliche Perioden

Gib JSON zurück mit:
- "likely_cause": Wahrscheinlichste Ursache
- "data_quality_score": 0-100
- "recommendations": Liste von Handlungsempfehlungen
"""
        
        response = self.client.chat.completions.create(
            model="deepseek-v3",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    def _create_gap_summary(self, gap_data: pd.DataFrame) -> str:
        """Erstellt eine文本 Zusammenfassung der Lückendaten"""
        return f"""
Anzahl der betroffenen Zeilen: {len(gap_data)}
Zeitstempel-Bereich: {gap_data['timestamp'].min()} - {gap_data['timestamp'].max()}
Durchschnittlicher Volumen-Rückgang: {gap_data['volume'].mean():.2f}
Preis-Range: {gap_data['low'].min():.2f} - {gap_data['high'].max():.2f}
"""



Hauptprogramm

async def main():
    # Initialisiere Handler
    gap_handler = AIEnhancedGapHandler("YOUR_HOLYSHEEP_API_KEY")
    
    # Lade Daten (Beispiel aus Tardis)
    sample_data = [
        OHLCV(1704067200000, 42000, 42100, 41900, 42050, 150),
        OHLCV(1704067260000, 42050, 42150, 42000, 42100, 160),  # 1 Min später (normal)
        OHLCV(1704067380000, 42100, 42200, 42050, 42150, 180),  # 2 Min später (normal)
        # Simulierte Lücke: 30 Min später
        OHLCV(1704069000000, 42500, 42600, 42450, 42550, 170),
    ]
    
    # Verarbeite Daten
    cleaned = gap_handler.process_ohlcv_series(sample_data)
    
    print(f"Verarbeitete Kerzen: {len(cleaned)}")
    print(f"Interpolierte Punkte: {sum(1 for c in cleaned if hasattr(c, 'is_interpolated') and c.is_interpolated)}")
    
    # KI-Analyse für große Lücken
    df = pd.DataFrame([{
        'timestamp': c.timestamp,
        'open': c.open,
        'high': c.high,
        'low': c.low,
        'close': c.close,
        'volume': c.volume
    } for c in cleaned])
    df['has_large_gap'] = False
    
    analysis = await gap_handler.analyze_large_gaps_ai(df, "BTC-USDT")
    print(f"KI-Analyse: {analysis}")


if __name__ == "__main__":
    asyncio.run(main())

Häufige Fehler und Lösungen

Fehler 1: Rate-Limit-Erschöpfung bei Burst-Requests

Symptom: HTTP 429 Fehler, Datenlücken in den Exporten, Timeouts

# FEHLERHAFT - Sofort alle Daten anfordern
async function fetchAllData() {
    const symbols = ['BTC-USDT', 'ETH-USDT', 'SOL-USDT', 'XRP-USDT'];
    
    // Diese Methode führt zu Rate-Limit-Überschreitung
    const results = await Promise.all(
        symbols.map(s => tardis.getHistoricalTrades({ symbol: s, from: startDate }))
    );
    return results;
}


LÖSUNG - Rate-Limited Batch-Verarbeitung

class RateLimitedClient {
    private requestQueue: Promise<any> = Promise.resolve();
    private minInterval = 100; // Min 100ms zwischen Requests
    
    async fetchWithCooldown<T>(fn: () => Promise<T>): Promise<T> {
        return new Promise((resolve, reject) => {
            this.requestQueue = this.requestQueue.then(async () => {
                try {
                    // Prüfe Rate-Limit-Header
                    const result = await fn();
                    const remaining = parseInt(result.headers['x-ratelimit-remaining'] || '10');
                    
                    if (remaining < 5) {
                        // Warte länger wenn Limit fast erreicht
                        const resetTime = parseInt(result.headers['x-ratelimit-reset'] || '0');
                        const waitMs = Math.max(0, resetTime - Date.now()) + 1000;
                        await this.sleep(waitMs);
                    }
                    
                    await this.sleep(this.minInterval);
                    resolve(result);
                } catch (error: any) {
                    if (error.status === 429) {
                        // Exponential Backoff
                        const retryAfter = parseInt(error.headers['retry-after'] || '1');
                        await this.sleep(retryAfter * 1000);
                        // Rekursiver Retry
                        resolve(this.fetchWithCooldown(fn));
                    } else {
                        reject(error);
                    }
                }
            });
        });
    }
    
    private sleep(ms: number): Promise<void> {
        return new Promise(r => setTimeout(r, ms));
    }
}

// Verwendung
const rateLimited = new RateLimitedClient();
const results = await rateLimited.fetchWithCooldown(() => 
    tardis.getHistoricalTrades({ symbol: 'BTC-USDT' })
);

Fehler 2: Zeitzonen-Probleme bei Multi-Exchange-Daten

Symptom: Trades erscheinen in falscher zeitlicher Reihenfolge, Overlapping candles, doppelte Einträge

# FEHLERHAFT - Annahme einheitlicher Zeitstempel
function mergeTrades(binance: any[], coinbase: any[]) {
    // Führt zu falschen Zuordnungen!
    return [...binance, ...coinbase].sort((a, b) => a.timestamp - b.timestamp);
}


LÖSUNG - Explizite Normalisierung zu UTC

class TimestampNormalizer {
    // Bekannte Exchange-Zeitzonen
    private static readonly EXCHANGE_TZ: Record<string, string> = {
        'binance': 'UTC',      // Bereits UTC
        'coinbase': 'America/New_York',  // Lokale Zeit
        'kraken': 'UTC',       // UTC
        'bitstamp': 'Europe/London',     // London Zeit
        'okex': 'Asia/Shanghai',         // China Zeit
    };
    
    static normalize(trades: any[], exchange: string): any[] {
        const tz = this.EXCHANGE_TZ[exchange] || 'UTC';
        
        return trades.map(trade => {
            let ts = trade.timestamp;
            
            // Fall 1: Unix-Sekunden (vor 1970)
            if (ts < 1e9) {
                ts *= 1000;
            }
            
            // Fall 2: String-Zeitstempel
            if (typeof ts === 'string') {
                ts = new Date(ts).getTime();
            }
            
            // Fall 3: Explizite Zeitzone
            if (trade.timestamp && trade.timestamp.$date) {
                // MongoDB-Format
                ts = new Date(trade.timestamp.$date).getTime();
            }
            
            return {
                ...trade,
                timestamp: ts,
                timestampUtc: new Date(ts).toISOString(),
                exchange: exchange,
                normalized: true
            };
        });
    }
    
    static mergeAndDeduplicate(exchanges: Record<string, any[]>): any[] {
        const allTrades: any[] = [];
        
        // Normalisiere jeden Exchange
        for (const [exchange, trades] of Object.entries(exchanges)) {
            const normalized = this.normalize(trades, exchange);
            allTrades.push(...normalized);
        }
        
        // Sortiere nach Zeitstempel
        allTrades.sort((a, b) => a.timestamp - b.timestamp);
        
        // Entferne Duplikate (gleicher Trade-ID und Exchange)
        const seen = new Set<string>();
        return allTrades.filter(trade => {
            const key = ${trade.exchange}-${trade.id};
            if (seen.has(key)) return false;
            seen.add(key);
            return true;
        });
    }
}

// Verwendung
const normalized = TimestampNormalizer.mergeAndDeduplicate({
    binance: binanceTrades,
    coinbase: coinbaseTrades,
    kraken: krakenTrades
});

Fehler 3: WebSocket Reconnection-Desaster

Symptom: Doppelte Nachrichten, Memory-Leaks, Crash nach längerer Laufzeit

# FEHLERHAFT - Keine Reconnection-Logik
async function subscribe(symbol: string) {
    const ws = new WebSocket('wss://api.tardis.dev/realtime');
    
    ws.on('message', (data) => {
        processTrade(JSON.parse(data));
    });
    
    // Keine Fehlerbehandlung!
    // Verbindung bleibt offen auch nach Server-Close
}


LÖSUNG - Robuste WebSocket-Verwaltung

class RobustWebSocket {
    private ws: WebSocket | null = null;
    private reconnectAttempts = 0;
    private readonly maxAttempts = 20;
    private messageBuffer: any[] = [];
    private lastProcessedId: string | null = null;
    
    constructor(
        private url: string,
        private onMessage: (msg: any) => void,
        private onError?: (err: any) => void
    ) {}
    
    async connect(): Promise<void> {
        return new Promise((resolve, reject) => {
            this.ws = new WebSocket(this.url);
            
            this.ws.on('open', () => {
                console.log('WebSocket verbunden');
                this.reconnectAttempts = 0;
                this.sendBufferedMessages();
                resolve();
            });
            
            this.ws.on('message', (raw) => {
                try {
                    const msg = JSON.parse(raw.toString());
                    this.handleMessage(msg);
                } catch (e) {
                    console.error('Parse-Fehler:', e);
                }
            });
            
            this.ws.on('close', (code, reason) => {
                console.log(WebSocket geschlossen: ${code} - ${reason});
                this.scheduleReconnect();
            });
            
            this.ws.on('error', (err) => {
                console.error('WebSocket-Fehler:', err);
                this.onError?.(err);
            });
        });
    }
    
    private handleMessage(msg: any): void {
        // Deduplizierung via Sequenz-Nummer
        if (msg.id && msg.id <= this.lastProcessedId) {
            return; // Duplikat, überspringen
        }
        
        // Replay-Protection: Falls Verbindung verloren ging
        if (msg.sequence && this.lastProcessedId) {
            const lastSeq = parseInt(this.lastProcessedId);
            const newSeq = parseInt(msg.sequence);
            
            if (newSeq > lastSeq + 1) {
                console.warn(Lücke erkannt: ${lastSeq + 1} bis ${newSeq - 1});
                this.onGapDetected(lastSeq + 1, newSeq - 1);
            }
        }
        
        this.lastProcessedId = msg.id || msg.sequence;
        this.onMessage(msg);
    }
    
    private onGapDetected(from: number, to: number): void {
        // Hole fehlende Nachrichten aus dem Backend
        this.fetchMissedMessages(from, to);
    }
    
    private async fetchMissedMessages(from: number, to: number): Promise<void> {
        console.log(Hole fehlende Nachrichten ${from} bis ${to});
        //

Verwandte Ressourcen
Verwandte Artikel