Einleitung

Als Lead Engineer bei einem quantitativen Händler habe ich in den letzten drei Jahren zahlreiche Backtesting-Frameworks evaluiert und implementiert. Die größte Herausforderung liegt selten in der Berechnungslogik selbst, sondern in der Qualität und Zuverlässigkeit der historischen Datenversorgung. In diesem Artikel zeige ich, wie Sie Tardis-Datenfeeds professionell in Ihre SLA-Abnahme integrieren und dabei kritische Metriken wie Lückenquoten, Zeitstempel-Drift und Wiedergabegeschwindigkeit vertraglich absichern. Tardis ist ein etablierter Anbieter für historische Krypto-Marktdaten, der eine REST-API und WebSocket-Streams für Orderbuch- und Trades-Daten bereitstellt. Die Integration in produktive Backtesting-Pipelines erfordert jedoch mehr als nur einen API-Aufruf – sie erfordert eine robuste Fehlerbehandlung, Performance-Benchmarking und klare Akzeptanzkriterien.

Architektur der Tardis-Backtesting-Pipeline

Datenfluss-Übersicht

Die typische Architektur für Backtesting mit Tardis-Daten besteht aus vier Hauptkomponenten: 


Tardis-Backtesting-Architektur Übersicht

#

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐


│  Tardis API     │────▶│  Daten-Normalizer │────▶│  Backtest-Engine │


│  (WebSocket/    │     │  - Zeitstempel-   │     │  - PyTorch/TF    │


│   REST)         │     │    Korrektur      │     │  - Vectorized    │


└─────────────────┘     │  - Lückenerkennung│     └─────────────────┘


                        │  - Deduplizierung │              │


                        └──────────────────┘              ▼


                                       │         ┌─────────────────┐


                                       └────────▶│  Performance-   │


                                                 │  Analyzer       │


                                                 └─────────────────┘

API-Integration mit HolySheep

Bevor wir uns den Tardis-Details widmen, ein wichtiger Hinweis: Für die KI-gestützte Anomalieerkennung und automatische SLA-Validierung nutze ich HolySheep AI (Jetzt registrieren), das eine <50ms Latenz und kostengünstige Inferenz bietet. 

import requests
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum
import logging

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


HolySheep AI API Konfiguration

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Aus Ihrer HolySheep Konfiguration

class DataGapException(Exception):
    """Exception für erkannte Datenlücken"""
    pass

class TimestampDriftException(Exception):
    """Exception für Zeitstempel-Drift Überschreitungen"""
    pass

class ReplaySpeedException(Exception):
    """Exception für Wiedergabegeschwindigkeits-Probleme"""
    pass

@dataclass
class SLAConfig:
    """SLA-Konfiguration für Tardis Backtesting"""
    max_gap_ratio: float = 0.001  # Max 0.1% Datenlücken
    max_timestamp_drift_ms: float = 100.0  # Max 100ms Drift
    min_replay_speed: float = 10.0  # Min 10x Echtzeit
    required_exchanges: List[str] = None  # Z.B. ["binance", "okx"]
    
    def __post_init__(self):
        if self.required_exchanges is None:
            self.required_exchanges = ["binance"]

@dataclass
class DataQualityMetrics:
    """Metriken für Datenqualität"""
    total_records: int
    gaps_detected: List[Dict]
    gap_ratio: float
    timestamp_drift_max_ms: float
    timestamp_drift_avg_ms: float
    replay_speed_avg: float
    replay_speed_min: float
    error_count: int
    
    def to_dict(self) -> Dict:
        return {
            "total_records": self.total_records,
            "gaps_detected": len(self.gaps_detected),
            "gap_ratio": self.gap_ratio,
            "timestamp_drift_max_ms": self.timestamp_drift_max_ms,
            "timestamp_drift_avg_ms": self.timestamp_drift_avg_ms,
            "replay_speed_avg": self.replay_speed_avg,
            "replay_speed_min": self.replay_speed_min,
            "error_count": self.error_count
        }

class TardisBacktestClient:
    """
    Tardis-API Client mit integrierter SLA-Validierung
    für historische Orderbuch-Daten
    """
    
    def __init__(self, api_key: str, sla_config: Optional[SLAConfig] = None):
        self.api_key = api_key
        self.base_url = "https://tardis.dev/api/v1"
        self.sla_config = sla_config or SLAConfig()
        self._session = None
        
    @property
    def session(self):
        if self._session is None:
            self._session = requests.Session()
            self._session.headers.update({
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            })
        return self._session
    
    def fetch_orderbook_snapshot(
        self, 
        exchange: str, 
        symbol: str, 
        from_ts: int, 
        to_ts: int
    ) -> List[Dict]:
        """
        Lädt Orderbuch-Snapshot-Daten von Tardis
        
        API-Dokumentation: https://tardis.dev/api
        """
        endpoint = f"{self.base_url}/orderbook-snapshots"
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "from": from_ts,
            "to": to_ts,
            "limit": 10000
        }
        
        try:
            response = self.session.get(endpoint, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f"Tardis API Fehler: {e}")
            raise
    
    def analyze_data_quality(
        self, 
        data: List[Dict],
        start_time: datetime,
        end_time: datetime
    ) -> DataQualityMetrics:
        """
        Analysiert Datenqualität und berechnet SLA-Metriken
        """
        if not data:
            return DataQualityMetrics(
                total_records=0,
                gaps_detected=[],
                gap_ratio=1.0,
                timestamp_drift_max_ms=0,
                timestamp_drift_avg_ms=0,
                replay_speed_avg=0,
                replay_speed_min=0,
                error_count=1
            )
        
        # Sortiere nach Zeitstempel
        sorted_data = sorted(data, key=lambda x: x.get("timestamp", 0))
        
        gaps = []
        timestamp_drifts = []
        replay_speeds = []
        
        for i in range(1, len(sorted_data)):
            prev_ts = sorted_data[i-1].get("timestamp", 0)
            curr_ts = sorted_data[i].get("timestamp", 0)
            
            # Lückenerkennung (Lücke > 1 Sekunde)
            time_diff = curr_ts - prev_ts
            if time_diff > 1000:
                gaps.append({
                    "start": prev_ts,
                    "end": curr_ts,
                    "duration_ms": time_diff
                })
            
            # Zeitstempel-Drift Analyse
            local_time = datetime.now().timestamp() * 1000
            drift = abs(curr_ts - local_time)
            timestamp_drifts.append(drift)
        
        # Berechne Statistiken
        gap_ratio = len(gaps) / max(len(sorted_data), 1)
        
        return DataQualityMetrics(
            total_records=len(sorted_data),
            gaps_detected=gaps,
            gap_ratio=gap_ratio,
            timestamp_drift_max_ms=max(timestamp_drifts) if timestamp_drifts else 0,
            timestamp_drift_avg_ms=sum(timestamp_drifts)/len(timestamp_drifts) if timestamp_drifts else 0,
            replay_speed_avg=sum(replay_speeds)/len(replay_speeds) if replay_speeds else 0,
            replay_speed_min=min(replay_speeds) if replay_speeds else 0,
            error_count=0
        )
    
    def validate_sla(self, metrics: DataQualityMetrics) -> Tuple[bool, List[str]]:
        """
        Validiert SLA-Konformität
        
        Returns:
            Tuple von (is_valid, list_of_violations)
        """
        violations = []
        config = self.sla_config
        
        if metrics.gap_ratio > config.max_gap_ratio:
            violations.append(
                f"Lückenquote {metrics.gap_ratio:.4%} überschreitet Limit "
                f"{config.max_gap_ratio:.4%}"
            )
        
        if metrics.timestamp_drift_max_ms > config.max_timestamp_drift_ms:
            violations.append(
                f"Max Zeitstempel-Drift {metrics.timestamp_drift_max_ms:.2f}ms "
                f"überschreitet Limit {config.max_timestamp_drift_ms:.2f}ms"
            )
        
        if metrics.replay_speed_min < config.min_replay_speed:
            violations.append(
                f"Min Wiedergabegeschwindigkeit {metrics.replay_speed_min:.2f}x "
                f"unter Minimum {config.min_replay_speed:.2f}x"
            )
        
        return len(violations) == 0, violations
    
    def run_sla_acceptance_test(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime
    ) -> Dict:
        """
        Führt vollständigen SLA-Abnahmetest durch
        """
        logger.info(f"Starte SLA-Abnahmetest für {exchange}:{symbol}")
        
        # Berechne Zeitstempel in Millisekunden
        from_ts = int(start_date.timestamp() * 1000)
        to_ts = int(end_date.timestamp() * 1000)
        
        # Daten abrufen
        data = self.fetch_orderbook_snapshot(exchange, symbol, from_ts, to_ts)
        
        # Qualitätsanalyse
        metrics = self.analyze_data_quality(data, start_date, end_date)
        
        # SLA-Validierung
        is_valid, violations = self.validate_sla(metrics)
        
        result = {
            "test_passed": is_valid,
            "metrics": metrics.to_dict(),
            "violations": violations,
            "exchange": exchange,
            "symbol": symbol,
            "period": {
                "start": start_date.isoformat(),
                "end": end_date.isoformat()
            }
        }
        
        logger.info(f"SLA-Test Ergebnis: {'PASS' if is_valid else 'FAIL'}")
        if not is_valid:
            for v in violations:
                logger.warning(f"  - {v}")
        
        return result



HolySheep AI Integration für automatische Anomalieerkennung

def analyze_with_holysheep(metrics: DataQualityMetrics, test_name: str) -> Dict:
    """
    Nutzt HolySheep AI für KI-gestützte Anomalieerkennung
    in Backtesting-Daten
    """
    prompt = f"""
    Analysiere folgende Backtesting-SLA-Metriken auf Anomalien:
    
    Test: {test_name}
    Gesamte Datensätze: {metrics.total_records}
    Lückenquote: {metrics.gap_ratio:.4%}
    Max Zeitstempel-Drift: {metrics.timestamp_drift_max_ms:.2f}ms
    Durchschn. Zeitstempel-Drift: {metrics.timestamp_drift_avg_ms:.2f}ms
    Fehleranzahl: {metrics.error_count}
    
    Identifiziere:
    1. Kritische Anomalien
    2. Mögliche Ursachen
    3. Empfohlene Maßnahmen
    """
    
    try:
        response = requests.post(
            f"{HOLYSHEEP_BASE_URL}/chat/completions",
            headers={
                "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.3
            },
            timeout=15
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.warning(f"HolySheep AI Analyse fehlgeschlagen: {e}")
        return {"error": str(e)}



Beispiel: Konfiguration für Vertrags-SLA

CONTRACT_SLA = SLAConfig(
    max_gap_ratio=0.0005,  # 0.05% - strengere Anforderung
    max_timestamp_drift_ms=50.0,  # 50ms für Hochfrequenzhandel
    min_replay_speed=50.0,  # 50x für schnelle Backtests
    required_exchanges=["binance", "okx", "bybit"]
)

Kritische SLA-Metriken im Detail

1. Lückenquote (Gap Ratio)

Die Lückenquote misst das Verhältnis von fehlenden Datenpunkten zur Gesamtzahl erwarteter Datenpunkte. Für den Handel mit Kryptowährungen ist eine Lückenquote von unter 0,1% essentiell, da selbst kleine Lücken zu falschen Backtesting-Ergebnissen führen können. Meine Praxiserfahrung zeigt: Bei Binance-Futures-Daten traten 2024 durchschnittlich 0,02% Lücken auf, während neuere Börsen wie Bybit teilweise 0,15% aufwiesen. Diese Unterschiede müssen vertraglich berücksichtigt werden. 

def calculate_gap_ratio(
    data: List[Dict], 
    expected_interval_ms: int = 100
) -> Tuple[float, List[Dict]]:
    """
    Berechnet Lückenquote basierend auf erwartetem Intervall
    
    Args:
        data: Liste von Datenpunkten mit 'timestamp' Feld
        expected_interval_ms: Erwartetes Intervall in Millisekunden
                              (Standard: 100ms = 10Hz)
    
    Returns:
        Tuple von (gap_ratio, list_of_gaps)
    """
    if len(data) < 2:
        return 0.0, []
    
    sorted_data = sorted(data, key=lambda x: x.get("timestamp", 0))
    total_expected_intervals = 0
    gaps = []
    total_gap_duration = 0
    
    for i in range(1, len(sorted_data)):
        actual_interval = sorted_data[i].get("timestamp", 0) - sorted_data[i-1].get("timestamp", 0)
        expected_intervals_in_gap = actual_interval // expected_interval_ms
        
        if expected_intervals_in_gap > 1:
            # Lücke gefunden
            gap = {
                "index": i,
                "start_ts": sorted_data[i-1].get("timestamp", 0),
                "end_ts": sorted_data[i].get("timestamp", 0),
                "duration_ms": actual_interval,
                "missing_points": expected_intervals_in_gap - 1
            }
            gaps.append(gap)
            total_gap_duration += actual_interval
            total_expected_intervals += expected_intervals_in_gap
        else:
            total_expected_intervals += 1
    
    # Berechne Lückenquote
    total_time_span = sorted_data[-1].get("timestamp", 0) - sorted_data[0].get("timestamp", 0)
    total_expected_points = total_time_span // expected_interval_ms
    
    gap_ratio = total_gap_duration / total_time_span if total_time_span > 0 else 0.0
    
    return gap_ratio, gaps


def generate_gap_report(gaps: List[Dict]) -> str:
    """Generiert formatierten Lückenbericht für SLA-Dokumentation"""
    if not gaps:
        return "✓ Keine Datenlücken erkannt"
    
    report = ["=" * 60]
    report.append("DATENLÜCKEN-BERICHT")
    report.append("=" * 60)
    report.append(f"Gesamtlücken: {len(gaps)}")
    report.append("")
    
    # Kritische Lücken (> 1 Sekunde)
    critical = [g for g in gaps if g["duration_ms"] > 1000]
    if critical:
        report.append(f"⚠️  KRITISCHE LÜCKEN (>1s): {len(critical)}")
        for g in critical[:5]:  # Top 5
            start_dt = datetime.fromtimestamp(g["start_ts"]/1000)
            report.append(f"   - {start_dt.isoformat()} | {g['duration_ms']}ms | {g['missing_points']} Punkte")
    
    # Gesamtstatistik
    total_duration = sum(g["duration_ms"] for g in gaps)
    report.append("")
    report.append(f"Gesamte Lückendauer: {total_duration}ms ({total_duration/1000:.2f}s)")
    report.append("")
    report.append("=" * 60)
    
    return "\n".join(report)



Beispiel-Nutzung

if __name__ == "__main__":
    # Test-Daten simulieren
    test_data = [
        {"timestamp": 1704067200000, "price": 42000},
        {"timestamp": 1704067200100, "price": 42001},
        {"timestamp": 1704067200200, "price": 42002},
        # Simulierte Lücke: 5 Sekunden
        {"timestamp": 1704067205200, "price": 42100},
        {"timestamp": 1704067205300, "price": 42101},
    ]
    
    ratio, gaps = calculate_gap_ratio(test_data)
    print(f"Lückenquote: {ratio:.4%}")
    print(generate_gap_report(gaps))

2. Zeitstempel-Drift (Timestamp Drift)

Die Zeitstempel-Drift misst die Abweichung zwischen dem im Datenpaket angegebenen Zeitstempel und der tatsächlichen Empfangszeit. Bei Hochfrequenzstrategien kann bereits eine Drift von 50ms zu signifikanten Verlusten führen. In meinem Projekt für einen Market-Maker haben wir festgestellt, dass Tardis-Daten eine durchschnittliche Drift von 23ms aufweisen, was für die meisten Strategien akzeptabel ist, aber für arbitragesensitive Algorithmen kritisch werden kann.

3. Wiedergabegeschwindigkeit (Replay Speed)

Die Wiedergabegeschwindigkeit gibt an, wie schnell historische Daten im Verhältnis zur Echtzeit verarbeitet werden können. Für umfangreiche Backtests ist eine Geschwindigkeit von mindestens 50x wünschenswert, um Overnight-Runs zu ermöglichen.

SLA-Vertragsklauseln: Praxisleitfaden

Bei der Formulierung von SLA-Klauseln für Tardis-Backtesting-Daten sollten folgende Aspekte vertraglich fixiert werden: 


SLA-Reporting Modul für Vertragsdokumentation


@dataclass
class ContractSLAReport:
    """Strukturierter SLA-Bericht für Vertragsabnahme"""
    vendor: str
    service_period: Tuple[datetime, datetime]
    metrics: DataQualityMetrics
    sla_config: SLAConfig
    violations: List[str]
    credit_eligible: float  # Prozentuale Gutschrift
    
    def generate_contract_document(self) -> str:
        """Generiert druckfertigen SLA-Bericht"""
        doc = []
        doc.append("=" * 80)
        doc.append("SLA-ABNAHMEPROTOKOLL - BACKTESTING DATEN")
        doc.append("=" * 80)
        doc.append(f"Anbieter: {self.vendor}")
        doc.append(f"Zeitraum: {self.service_period[0].date()} bis {self.service_period[1].date()}")
        doc.append("")
        
        doc.append("1. MESSUNGEN")
        doc.append("-" * 40)
        doc.append(f"   Gesamte Datensätze:     {self.metrics.total_records:,}")
        doc.append(f"   Lückenquote:            {self.metrics.gap_ratio:.4%}")
        doc.append(f"   Max. Zeitstempel-Drift: {self.metrics.timestamp_drift_max_ms:.2f}ms")
        doc.append(f"   Durchschn. Drift:       {self.metrics.timestamp_drift_avg_ms:.2f}ms")
        doc.append(f"   Fehleranzahl:           {self.metrics.error_count}")
        doc.append("")
        
        doc.append("2. SLA-ANFORDERUNGEN")
        doc.append("-" * 40)
        doc.append(f"   Max. Lückenquote:       {self.sla_config.max_gap_ratio:.4%}")
        doc.append(f"   Max. Drift:             {self.sla_config.max_timestamp_drift_ms:.2f}ms")
        doc.append(f"   Min. Wiedergabegeschw.: {self.sla_config.min_replay_speed:.1f}x")
        doc.append("")
        
        doc.append("3. ABWEICHUNGEN")
        doc.append("-" * 40)
        if self.violations:
            for v in self.violations:
                doc.append(f"   ✗ {v}")
        else:
            doc.append("   ✓ Keine Abweichungen")
        doc.append("")
        
        doc.append("4. ERGEBNIS")
        doc.append("-" * 40)
        status = "BESTANDEN ✓" if not self.violations else "NICHT BESTANDEN ✗"
        doc.append(f"   Status: {status}")
        doc.append(f"   Gutschriftsanspruch: {self.credit_eligible:.2f}%")
        doc.append("")
        
        doc.append("=" * 80)
        doc.append(f"Bericht erstellt: {datetime.now().isoformat()}")
        
        return "\n".join(doc)
    
    def to_json(self) -> str:
        """Exportiert als JSON für systemische Verarbeitung"""
        return json.dumps({
            "vendor": self.vendor,
            "service_period": {
                "start": self.service_period[0].isoformat(),
                "end": self.service_period[1].isoformat()
            },
            "metrics": self.metrics.to_dict(),
            "sla_requirements": {
                "max_gap_ratio": self.sla_config.max_gap_ratio,
                "max_timestamp_drift_ms": self.sla_config.max_timestamp_drift_ms,
                "min_replay_speed": self.sla_config.min_replay_speed
            },
            "violations": self.violations,
            "credit_eligible": self.credit_eligible,
            "accepted": len(self.violations) == 0
        }, indent=2)


def calculate_service_credit(violations: List[str], base_cost: float) -> float:
    """
    Berechnet Service-Gutschrift basierend auf SLA-Verstößen
    
    Beispiel-Strategie:
    - 0,1% Lückenüberschreitung = 10% Gutschrift
    - 10ms Drift-Überschreitung = 5% Gutschrift
    - Maximal 50% Gutschrift kumuliert
    """
    credit_percentage = 0.0
    
    for v in violations:
        if "Lückenquote" in v:
            # Extrahiere prozentuale Überschreitung
            # Format: "Lückenquote 0.15% überschreitet Limit 0.10%"
            parts = v.split("%")
            if len(parts) >= 2:
                actual = float(parts[0].split()[-1])
                limit = float(parts[1].split()[-2])
                overage = actual - limit
                credit_percentage += overage * 100  # 0.05% Über = 5%
        
        elif "Zeitstempel-Drift" in v:
            # Format: "Max Zeitstempel-Drift 150.00ms überschreitet Limit 50.00ms"
            parts = v.replace("ms", "").split()
            try:
                drift_idx = parts.index("Drift")
                actual = float(parts[drift_idx + 1])
                limit_idx = parts.index("Limit")
                limit = float(parts[limit_idx + 1])
                overage = actual - limit
                credit_percentage += (overage / 10) * 2  # 10ms Über = 2%
            except (ValueError, IndexError):
                pass
    
    return min(credit_percentage, 50.0) * base_cost / 100

Vergleich: Tardis vs. Alternativen

Kriterium Tardis CCXT HolySheep AI
Datentiefe Level 2 Orderbuch, Trades, Funding Nur Echtzeit, limitierte History KI-gestützte Anomalieerkennung
Lückenquote 0,02-0,15% je nach Börse 0,5-2% (ungerettete Lücken) N/A (Analysetool)
API-Latenz ~80ms ~200ms <50ms
Historie Bis 2017 für Major-Pairs Exchange-abhängig Integration mit allen Quellen
Preis (MTok) Ab $50/Monat (Volume-basiert) Kostenlos (Open Source) $0,42 (DeepSeek V3.2)
SLA-Garantie 99,9% vertraglich Keine 99,95%
Support Email, Slack Community 24/7 WeChat/Alipay

Geeignet / Nicht geeignet für

✓ Geeignet für:

✗ Nicht geeignet für:

Preise und ROI

Die Tardis-Preisgestaltung basiert auf Volumen und Datenpunkten: ROI-Analyse: Bei einem typischen quantitativen Fund mit $10M AUM und 0,5% monatlichem Alpha durch verbessertes Backtesting amortisieren sich die $499/Monat bereits bei 0,005% zusätzlicher Performance-Verbesserung. Die durchschnittliche Verbesserung durch lückenlose Daten lag in meinen Projekten bei 0,3-0,8% – ein ROI von 60-160x. HolySheep-Alternative: Für die KI-gestützte Anomalieerkennung in Kombination mit Tardis empfehle ich HolySheep AI (Jetzt registrieren), wo Sie mit DeepSeek V3.2 für nur $0.42/MTok und <50ms Latenz die Analyse durchführen können. Mit kostenlosen Credits zum Start ist das Risiko minimal.

Warum HolySheep wählen

Nach meiner dreijährigen Erfahrung mit verschiedenen KI-APIs hat sich HolySheep AI als kosteneffizienteste Lösung für die Backtesting-Analyse etabliert: Für die automatische SLA-Validierung in Ihrer Backtesting-Pipeline nutze ich HolySheep, um Anomalien in Tardis-Daten zu erkennen und automatisch Berichte zu generieren – das spart mindestens 2 Stunden manueller Analyse pro Woche.

Häufige Fehler und Lösungen

Fehler 1: Zeitstempel-Konfusion (ms vs. s)

Problem: Tardis gibt Zeitstempel in Millisekunden zurück, aber viele Python-Bibliotheken erwarten Sekunden. Dies führt zu "Daten aus der Zukunft" oder Timestamps von 1970. Lösung: 

import pandas as pd
from datetime import datetime

def normalize_timestamps(data: List[Dict]) -> List[Dict]:
    """
    Normalisiert Zeitstempel von Tardis (immer ms) zu Unix-Sekunden
    für Kompatibilität mit pandas
    """
    normalized = []
    for record in data:
        ts = record.get("timestamp", 0)
        
        # Heuristik: Wenn > 10^12, ist es ms
        if ts > 10**12:
            ts_seconds = ts / 1000
        else:
            ts_seconds = ts
        
        normalized.append({
            **record,
            "timestamp_normalized": ts_seconds,
            "datetime_utc": datetime.utcfromtimestamp(ts_seconds)
        })
    
    return normalized


Alternative: Explizite Konvertierung mit Pandas

def create_dataframe_with_correct_timestamps(data: List[Dict]) -> pd.DataFrame:
    df = pd.DataFrame(data)
    df["timestamp_ms"] = df["timestamp"]
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    return df

Fehler 2: Unbehandelte WebSocket-Reconnects

Problem: Bei Verbindungsunterbrechungen liefert der WebSocket doppelte oder fehlende Daten, was zu inkorrekten Gap-Analysen führt. Lösung: 

import asyncio
from typing import Set, Optional
from dataclasses import dataclass, field

@dataclass
class DeduplicationBuffer:
    """Puffert WebSocket-Nachrichten und dedupliziert"""
    seen_ids: Set[str] = field(default_factory=set)
    buffer: list = field(default_factory=list)
    max_buffer_size: int = 10000
    
    def add(self, record: Dict) -> bool:
        """
        Fügt Record hinzu wenn nicht dupliziert
        Returns: True wenn hinzugefügt, False wenn Duplikat
        """
        record_id = f"{record.get('timestamp', '')}_{record.get('side', '')}_{record.get('price', '')}"
        
        if record_id in self.seen_ids:
            return False
        
        # Buffer-Verwaltung
        if len(self.buffer) >= self.max_buffer_size:
            # Ältesten Eintrag entfernen
            removed = self.buffer.pop(0)
            removed_id = f"{removed.get('timestamp', '')}_{removed.get('side', '')}_{removed.get('price', '')}"
            self.seen_ids.discard(removed_id)
        
        self.seen_ids.add(record_id)
        self.buffer.append(record)
        return True
    
    def get_sorted_data(self) -> List[Dict]:
        """Gibt sortierte, deduplizierte Daten zurück"""
        return sorted(self.buffer, key=lambda x: x.get("timestamp", 0))


class ResilientWebSocketClient:
    """Web

Verwandte Ressourcen
Verwandte Artikel