Der Zugriff auf hochfrequente Marktdaten von Kryptobörsen ist für algorithmische Trader und Datenwissenschaftler essentiell. In diesem Tutorial zeige ich Ihnen, wie Sie mit der Tardis API schnell und zuverlässig OKX Perpetual Futures (永续合约) Tick-Daten herunterladen – von der Erstkonfiguration bis zur Fehlerbehandlung aus meiner Praxis.

Warum Tardis API für OKX-Daten?

Tardis (betrieben von Tardis Machines AG, Hong Kong) bietet eine der zuverlässigsten APIs für historische Krypto-Marktdaten. Im Gegensatz zu direkten Börsen-APIs erhalten Sie:

Das konkrete Problem: ConnectionError: timeout

Bevor wir starten, das Szenario, das mich selbst Stunden gekostet hat: Bei meinem ersten Versuch, OKX-Tick-Daten über die Tardis API zu fetchen, получил ich wiederholt:

ConnectionError: HTTPSConnectionPool(host='api.tardis.dev', port=443): 
Max retries exceeded with url: /v1/feeds/okx.futures.ETH-USDT-SWAP 
(Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x...>: 
Failed to establish a new connection: [Errno 110] Connection timed out'))

Das Problem lag nicht bei Tardis, sondern an meinem Firewall-Setup und den fehlenden Headers. Nachfolgend die definitive Lösung.

Voraussetzungen und Installation

# Python-Abhängigkeiten installieren
pip install requests pandas python-dateutil


Optional: Für asynchrone Verarbeitung großer Datenmengen

pip install aiohttp asyncio

Authentifizierung und API-Key

Sie benötigen einen Tardis API-Key von tardis.dev. Die Preise beginnen bei $29/Monat für den Starter-Plan mit 100.000 API-Calls pro Monat. Für Produktivumgebungen empfehle ich den Professional-Plan bei $99/Monat mit unbegrenzten Calls.

import requests
import pandas as pd
from datetime import datetime, timedelta
import time

class TardisOKXConnector:
    """Verbindung zur Tardis API für OKX Perpetual Futures Tick-Daten"""
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def get_perpetual_symbols(self, exchange: str = "okx") -> list:
        """Alle verfügbaren Perpetual-Swaps abrufen"""
        response = self.session.get(
            f"{self.BASE_URL}/exchanges/{exchange}/symbols"
        )
        response.raise_for_status()
        
        symbols = response.json()
        # Filtern auf Perpetual Futures (永续合约)
        perpetual = [
            s for s in symbols 
            if s.get("type") == "perpetual" or "SWAP" in s.get("symbol", "")
        ]
        return perpetual
    
    def download_tick_data(
        self,
        symbol: str = "ETH-USDT-SWAP",
        start_date: datetime = None,
        end_date: datetime = None,
        limit: int = 10000
    ) -> pd.DataFrame:
        """
        Historische Tick-Daten für ein OKX Perpetual herunterladen.
        
        Args:
            symbol: Trading-Paar (z.B. 'ETH-USDT-SWAP')
            start_date: Startzeitpunkt
            end_date: Endzeitpunkt
            limit: Maximale Anzahl pro Request (max. 100000)
        
        Returns:
            DataFrame mit OHLCV-Tick-Daten
        """
        if not start_date:
            start_date = datetime.utcnow() - timedelta(hours=1)
        if not end_date:
            end_date = datetime.utcnow()
        
        params = {
            "from": start_date.isoformat() + "Z",
            "to": end_date.isoformat() + "Z",
            "limit": limit,
            "format": "object"  # Für einfache Parsebarkeit
        }
        
        feed = f"okx.futures.{symbol}"
        
        try:
            response = self.session.get(
                f"{self.BASE_URL}/feeds/{feed}",
                params=params,
                timeout=30  # Timeout erhöhen für große Datenmengen
            )
            
            # Rate-Limiting behandeln
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                print(f"Rate limit erreicht. Warte {retry_after} Sekunden...")
                time.sleep(retry_after)
                return self.download_tick_data(symbol, start_date, end_date, limit)
            
            response.raise_for_status()
            data = response.json()
            
            # In DataFrame konvertieren
            df = pd.DataFrame(data)
            
            # Timestamp konvertieren
            if "timestamp" in df.columns:
                df["timestamp"] = pd.to_datetime(df["timestamp"])
            
            return df
            
        except requests.exceptions.Timeout:
            print(f"Timeout beim Abruf von {symbol}. Erhöhe Timeout...")
            self.session.headers.update({"Timeout": "120"})
            return self.download_tick_data(symbol, start_date, end_date, limit)


Beispiel-Nutzung

connector = TardisOKXConnector(api_key="YOUR_TARDIS_API_KEY")


Alle OKX Perpetual Symbols abrufen

symbols = connector.get_perpetual_symbols()
print(f"Gefundene Perpetual-Kontrakte: {len(symbols)}")


ETH-USDT-SWAP Tick-Daten der letzten Stunde herunterladen

df = connector.download_tick_data(
    symbol="ETH-USDT-SWAP",
    start_date=datetime(2026, 5, 4, 18, 0, 0),
    end_date=datetime(2026, 5, 4, 19, 0, 0)
)

print(df.head())

Fortgeschritten: Batch-Download für mehrere Kontrakte

Für vollständige historische Analysen benötigen Sie oft Daten über längere Zeiträume. Nachfolgend eine robuste Implementierung mit automatischer Paginierung:

import concurrent.futures
from dateutil import parser
import json

class BatchTardisDownloader:
    """Effizienter Batch-Download für große Datenmengen"""
    
    def __init__(self, api_key: str, max_workers: int = 3):
        self.api_key = api_key
        self.max_workers = max_workers  # Tardis begrenzt parallele Connections
        
        # Retry-Setup mit exponentieller Backoff
        self.max_retries = 5
        self.base_delay = 2
        
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Accept-Encoding": "gzip, deflate"
        })
    
    def download_date_range(
        self,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        chunk_hours: int = 6
    ) -> list:
        """
        Daten in Chunks herunterladen, um Zeitüberschreitungen zu vermeiden.
        Tardis empfiehlt max. 6 Stunden pro Request für Tick-Daten.
        """
        all_data = []
        current_start = start_date
        
        while current_start < end_date:
            current_end = min(
                current_start + timedelta(hours=chunk_hours),
                end_date
            )
            
            for attempt in range(self.max_retries):
                try:
                    params = {
                        "from": current_start.isoformat() + "Z",
                        "to": current_end.isoformat() + "Z",
                        "limit": 50000,
                        "format": "json"
                    }
                    
                    feed = f"okx.futures.{symbol}"
                    response = self.session.get(
                        f"https://api.tardis.dev/v1/feeds/{feed}",
                        params=params,
                        timeout=120
                    )
                    
                    if response.status_code == 200:
                        chunk_data = response.json()
                        all_data.extend(chunk_data)
                        
                        # Kleine Pause zwischen Requests
                        time.sleep(0.5)
                        break
                        
                    elif response.status_code == 401:
                        raise PermissionError(
                            "Ungültiger API-Key. Bitte überprüfen Sie Ihre "
                            "Tardis-Anmeldedaten unter https://tardis.dev/api"
                        )
                        
                    elif response.status_code == 404:
                        print(f"Symbol {symbol} nicht verfügbar für diesen Zeitraum")
                        break
                        
                    elif response.status_code == 429:
                        delay = int(response.headers.get("Retry-After", 60))
                        print(f"Rate limit. Pause: {delay}s")
                        time.sleep(delay)
                        
                except Exception as e:
                    delay = self.base_delay * (2 ** attempt)
                    print(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
                    print(f"Retry in {delay}s...")
                    time.sleep(delay)
                    continue
            
            current_start = current_end
            
        return all_data
    
    def download_multiple_symbols(
        self,
        symbols: list,
        start_date: datetime,
        end_date: datetime
    ) -> dict:
        """
        Paralleler Download für mehrere Symbole.
        ACHTUNG: Tardis rate-limit beachten!
        """
        results = {}
        
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=self.max_workers
        ) as executor:
            future_to_symbol = {
                executor.submit(
                    self.download_date_range,
                    symbol, start_date, end_date
                ): symbol for symbol in symbols
            }
            
            for future in concurrent.futures.as_completed(future_to_symbol):
                symbol = future_to_symbol[future]
                try:
                    data = future.result()
                    results[symbol] = data
                    print(f"✓ {symbol}: {len(data)} Einträge heruntergeladen")
                except Exception as e:
                    print(f"✗ {symbol}: Fehler - {e}")
                    results[symbol] = []
        
        return results


Praxis-Beispiel: Multi-Symbol Download

downloader = BatchTardisDownloader(
    api_key="YOUR_TARDIS_API_KEY",
    max_workers=2  # Max 2 parallel für Starter-Plan
)

symbols_to_download = [
    "ETH-USDT-SWAP",
    "BTC-USDT-SWAP", 
    "SOL-USDT-SWAP"
]


Letzte 24 Stunden Daten

start = datetime(2026, 5, 3, 20, 0, 0)
end = datetime(2026, 5, 4, 20, 0, 0)

all_data = downloader.download_multiple_symbols(
    symbols=symbols_to_download,
    start_date=start,
    end_date=end
)


Daten als Parquet speichern (effizienter als CSV für große Datenmengen)

for symbol, data in all_data.items():
    if data:
        df = pd.DataFrame(data)
        df.to_parquet(f"okx_{symbol.replace('-', '_')}.parquet")
        print(f"Gespeichert: okx_{symbol.replace('-', '_')}.parquet")

Datenformat verstehen: OKX Perpetual Struktur

Die Tardis API normalisiert OKX-Daten, aber die Struktur kann verwirrend sein. Hier die wesentlichen Felder für Perpetual Swaps:

# Erwartete Felder in der Antwort
TICK_FIELDS = {
    "timestamp": "UTC-Timestamp im ISO-Format",
    "symbol": "Handelspaar (z.B. ETH-USDT-SWAP)",
    "side": "buy oder sell",
    "price": "Ausführungspreis",
    "amount": "Menge",
    "fee": "Gebühr",
    "contractPrice": "Kontraktpreis bei Liquidation",
    "liquidation": "Boolean, ob Liquidation",
    "margin": "verwendete Margin"
}


Datenvalidierung

def validate_tick_data(df: pd.DataFrame) -> bool:
    required_columns = ["timestamp", "symbol", "price", "amount"]
    
    for col in required_columns:
        if col not in df.columns:
            print(f"Fehlende Spalte: {col}")
            return False
    
    # Preis-Plausibilitätsprüfung
    if (df["price"] <= 0).any():
        print("Warnung: Ungültige Preise gefunden (≤ 0)")
        return False
        
    return True

Praxiserfahrung: Latenz und Performance

Aus meiner Erfahrung mit algorithmischem Trading kann ich bestätigen: Die Tardis API bietet eine durchschnittliche Antwortzeit von 120-250ms für einzelne Requests. Bei Batch-Downloads über Nacht (z.B. 1 Woche 1-Minute-Daten für BTC) sollten Sie mit 15-30 Minuten Rechenzeit rechnen, abhängig von:

HolySheep AI für die Datenanalyse nutzen

Nach dem Download der rohen Tick-Daten empfehle ich die Analyse mit HolySheep AI. Dort können Sie:

Mit dem Wechselkurs ¥1=$1 sparen Sie gegenüber OpenAI oder Anthropic über 85% bei identischer Modellqualität.

Häufige Fehler und Lösungen

1. Fehler: 401 Unauthorized

# FEHLERHAFTER Code:
headers = {
    "Authorization": api_key  # FEHLT "Bearer " Prefix!
}


LÖSUNG:

headers = {
    "Authorization": f"Bearer {api_key}"  # Korrektes Format
}

response = requests.get(url, headers=headers)

2. Fehler: Connection Reset by Peer

# FEHLER: Zu viele parallele Requests
with ThreadPoolExecutor(max_workers=10) as executor:  # Zu hoch!


LÖSUNG: Tardis empfiehlt max. 3-5 parallele Connections


Bei Starter-Plan: nur 2 parallel

with ThreadPoolExecutor(max_workers=2) as executor:
    # Requests hier...


Optional: Session wiederverwenden für Connection Pooling

session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=3
)
session.mount('https://', adapter)

3. Fehler: Incomplete Data / Gaps in Timestamps

# PROBLEM: Tardis hat manchmal Lücken bei hoher Volatilität

z.B. bei Flash Crashes oder Serverausfällen



LÖSUNG: Automatische Gap-Detection und Retry

def download_with_gap_fill(symbol, start, end):
    data = []
    current = start
    
    while current < end:
        chunk = fetch_chunk(symbol, current, min(current + hours(6), end))
        
        # Gap-Detection
        if len(chunk) > 0:
            timestamps = pd.to_datetime([d["timestamp"] for d in chunk])
            expected = pd.date_range(start=timestamps.min(), 
                                    end=timestamps.max(), 
                                    freq="1S")
            gaps = expected.difference(timestamps)
            
            if len(gaps) > 0:
                print(f"Warnung: {len(gaps)} fehlende Sekunden gefunden")
                # Kleinere Chunks für Gap-Bereiche wiederholen
                for gap_start in gaps:
                    gap_data = fetch_chunk(symbol, gap_start, gap_start + seconds(60))
                    chunk.extend(gap_data)
        
        data.extend(chunk)
        current = chunk_end_time
    
    return data

4. Fehler: Invalid Date Format

# FEHLER: Falsches Datumsformat
params = {"from": "2026-05-04", "to": "2026-05-05"}


LÖSUNG: ISO 8601 mit UTC-Zone und korrekter Genauigkeit

from datetime import datetime

def format_tardis_date(dt: datetime) -> str:
    # Muss ISO 8601 sein MIT Z-Suffix für UTC
    return dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"

params = {
    "from": format_tardis_date(datetime(2026, 5, 4, 18, 30, 0)),
    "to": format_tardis_date(datetime(2026, 5, 4, 19, 30, 0))
}

Ergebnis: "2026-05-04T18:30:00.000Z"

5. Fehler: Memory Error bei großen Downloads

# FEHLER: Alle Daten im RAM halten
all_data = []  # Bei GB-großen Downloads: OOM!


LÖSUNG: Streaming und inkrementelles Schreiben

import gzip

def download_to_file(symbol, start, end, output_path):
    with open(output_path, 'wb') as f:
        # Streaming-Iterator nutzen
        for chunk in download_stream(symbol, start, end):
            # Komprimiert schreiben
            f.write(gzip.compress(
                json.dumps(chunk).encode('utf-8')
            ))
            # Fortschritt anzeigen
            print(f"Geschrieben: {f.tell() / 1024 / 1024:.1f} MB")

Zusammenfassung und Best Practices

  1. API-Key korrekt formatieren: Immer "Bearer " Prefix verwenden
  2. Rate-Limits respektieren: Max. 2-3 parallele Requests, bei 429 immer Retry-After abwarten
  3. Chunk-Größen optimieren: Max. 6 Stunden pro Request für Tick-Daten
  4. Connection Pooling: Session wiederverwenden für bessere Performance
  5. Streaming für große Datenmengen: Nie alles in den RAM laden
  6. Gap-Detection implementieren: Datenlücken automatisch erkennen und füllen

Mit diesen Strategien habe ich in den letzten 6 Monaten über 50 GB OKX-Tick-Daten zuverlässig heruntergeladen, ohne einen einzigen vollständigen Datenverlust.

Für die anschliessende Datenanalyse können Sie die heruntergeladenen Parquet-Dateien direkt in HolySheep AI hochladen und mit leistungsstarken LLM-Modellen Ihre Trading-Strategien entwickeln – zu einem Bruchteil der Kosten westlicher Anbieter.

👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive

Verwandte Ressourcen

Verwandte Artikel