OKX & Binance historische Trades aggregieren: Arrow Parquet als einheitlicher Speicher

Wer professionell Krypto-Marktdaten verarbeitet, kennt das Problem: OKX liefert Trades als JSON-Array mit Feldern wie px, sz, side, ts; Binance liefert price, qty, time, isBuyerMaker. Beide Exchanges sind Pflicht, will man Arbitrage-, Market-Making- oder Backtesting-Modelle füttern. Die Speicherung im einheitlichen Apache Arrow-In-Memory-Format und das Ablegen auf Festplatte als Parquet ist heute der Quasi-Standard in der Quant-Welt.

Bevor wir in die Implementierung einsteigen, ein nüchterner Blick auf das, was KI-Workflows im Jahr 2026 wirklich kosten – denn viele Leser dieses Artikels betreiben zusätzlich LLM-gestützte Datenklassifikation oder Signal-Generierung:

Modell	Output $/MTok	Kosten 10M Tok/Monat	HolySheep ¥/MTok	HolySheep 10M Tok/Monat
GPT-4.1	8,00 $	80.000 $	¥8	¥80.000
Claude Sonnet 4.5	15,00 $	150.000 $	¥15	¥150.000
Gemini 2.5 Flash	2,50 $	25.000 $	¥2,50	¥25.000
DeepSeek V3.2	0,42 $	4.200 $	¥0,42	¥4.200

Durch den Wechselkurs ¥1 = $1 bei HolySheep AI (statt real ~1 $ = 7,2 ¥) ergibt sich eine Ersparnis von über 85 % – bei gleicher Modellqualität. Gerade wer täglich Millionen Tokens durch GPT-4.1 zur Tick-Klassifikation jagt, sieht hier den größten Hebel.

Ausgangslage: Warum CSV & JSON auf Dauer nicht reichen

• OKX liefert im Endpoint /api/v5/market/trades-history paginiert maximal 500 Trades – bei aktivem BTC-USDT-Markt sind das ~1 Sekunde.
• Binance liefert im Endpoint /api/v3/trades max. 1000 Trades – öffentlich, ohne Authentifizierung.
• CSV wächst unkontrolliert: 1 Tag BTC-USDT = ~6 GB. Binäre Formate wie Parquet komprimieren mit Snappy typischerweise auf 700 MB – Faktor 8–10.
• Arrow erlaubt null-copy Sharing zwischen Python, DuckDB, Polars und Spark – kein Hin-und-her-Konvertieren.

Architektur: Bronze → Silver → Gold mit Arrow & Parquet

In Anlehnung an das Medaillon-Modell:

• Bronze – Roh-JSON-Antworten, unverändert abgelegt (Append-only, S3-kompatibel).
• Silver – normalisiert in ein einheitliches Arrow-Schema (siehe unten), partitioniert nach exchange/symbol/year/month/day.
• Gold – aggregierte Minutenkerzen, Orderbook-Snapshots, Funding-Daten – daraus entstehen Handelssignale.

Wir bauen in diesem Artikel die Silver-Schicht komplett auf.

Schritt 1 – Trades von OKX abrufen

OKX verwendet einen after-Cursor (Trade-ID absteigend) für Pagination. Wichtig: nur öffentliche Endpoints nutzen, sonst API-Key-Management.

import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime, timezone
import time, os

OKX_BASE = "https://www.okx.com"
BINANCE_BASE = "https://api.binance.com"

HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; holysheep-pipeline/1.0)"}

def fetch_okx_trades(inst_id: str, total_rows: int = 50_000):
    """OKX V5 – historische Trades paginiert via after-Cursor."""
    rows, after = [], None
    while len(rows) < total_rows:
        params = {"instId": inst_id, "limit": "500"}
        if after:
            params["after"] = after
        r = requests.get(f"{OKX_BASE}/api/v5/market/trades-history",
                         params=params, headers=HEADERS, timeout=10)
        r.raise_for_status()
        payload = r.json()
        if payload.get("code") != "0":
            raise RuntimeError(f"OKX-Fehler {payload.get('code')}: {payload.get('msg')}")
        batch = payload.get("data", [])
        if not batch:
            break
        rows.extend(batch)
        # OKX liefert strikt absteigend; neueste zuerst
        after = batch[-1]["tradeId"]
        time.sleep(0.05)  # 20 req/s – OKX Public Rate-Limit
    return rows[:total_rows]

Schritt 2 – Trades von Binance abrufen

Binance nutzt numerische fromId-Pagination. Auch hier bleiben wir auf dem öffentlichen Endpoint ohne Signatur.

def fetch_binance_trades(symbol: str, total_rows: int = 50_000):
    """Binance – historische Trades paginiert via fromId."""
    rows, from_id = [], None
    while len(rows) < total_rows:
        params = {"symbol": symbol, "limit": 1000}
        if from_id is not None:
            params["fromId"] = from_id
        r = requests.get(f"{BINANCE_BASE}/api/v3/historicalTrades",
                         params=params, headers=HEADERS, timeout=10)
        r.raise_for_status()
        batch = r.json()
        if not batch:
            break
        rows.extend(batch)
        from_id = batch[-1]["id"] + 1
        time.sleep(0.05)  # 1200 req/min öffentlich
    return rows[:total_rows]

Schritt 3 – Vereinheitlichung in ein Arrow-Schema

Der entscheidende Schritt: ein Tabellenschema, in dem beide Exchanges landen. Damit kann später jede Engine (DuckDB, Polars, Spark) ohne Mapping arbeiten.

UNIFIED_SCHEMA = pa.schema([
    ("exchange",   pa.string()),
    ("symbol",     pa.string()),
    ("trade_id",   pa.string()),
    ("price",      pa.float64()),
    ("qty",        pa.float64()),
    ("side",       pa.string()),       # "buy" oder "sell" (Taker-Seite)
    ("ts_ms",      pa.int64()),        # Unix-Millisekunden
    ("ts_iso",     pa.string()),       # UTC ISO8601
])

def okx_to_rows(records, inst_id):
    for t in records:
        yield {
            "exchange": "okx",
            "symbol":   inst_id,
            "trade_id": t["tradeId"],
            "price":    float(t["px"]),
            "qty":      float(t["sz"]),
            "side":     t["side"].lower(),
            "ts_ms":    int(t["ts"]),
            "ts_iso":   datetime.fromtimestamp(int(t["ts"])/1000,
                                               tz=timezone.utc).isoformat(),
        }

def binance_to_rows(records, symbol):
    for t in records:
        yield {
            "exchange": "binance",
            "symbol":   symbol,
            "trade_id": str(t["id"]),
            "price":    float(t["price"]),
            "qty":      float(t["qty"]),
            # isBuyerMaker == True  ->  Käufer ist Maker -> Taker hat verkauft
            "side":     "sell" if t["isBuyerMaker"] else "buy",
            "ts_ms":    int(t["time"]),
            "ts_iso":   datetime.fromtimestamp(int(t["time"])/1000,
                                               tz=timezone.utc).isoformat(),
        }

def build_arrow_table(okx_records, okx_symbol, bin_records, bin_symbol):
    df = pd.DataFrame(list(okx_to_rows(okx_records, okx_symbol))
                    + list(binance_to_rows(bin_records, bin_symbol)))
    # Cast strikt nach UNIFIED_SCHEMA – Arrow liefert sonst Float32 statt Float64
    table = pa.Table.from_pandas(df, schema=UNIFIED_SCHEMA, preserve_index=False)
    return table

Schritt 4 – Partitioniertes Parquet schreiben

Partitionierung nach exchange und symbol aktiviert Pruning bei analytischen Queries. Snappy-Komprimierung ist Standard, weil sie lesegeschwindigkeits-optimal ist. Wir nutzen row_group_size=100_000, was für DuckDB-Pushdown ideal ist.

def write_silver_parquet(table: pa.Table, base_path="./trades_silver"):
    pq.write_to_dataset(
        table,
        root_path=base_path,
        partition_cols=["exchange", "symbol"],
        compression="snappy",
        use_dictionary=True,
        write_statistics=True,
        row_group_size=100_000,
        data_page_size=1 << 20,  # 1 MiB
    )
    print(f"Geschrieben: {base_path}/exchange=…/symbol=…")

---------- Pipeline-Lauf ----------
if __name__ == "__main__":
    okx_raw  = fetch_okx_trades("BTC-USDT",  total_rows=20_000)
    bin_raw  = fetch_binance_trades("BTCUSDT", total_rows=20_000)
    table    = build_arrow_table(okx_raw, "BTC-USDT", bin_raw, "BTCUSDT")
    write_silver_parquet(table)

Vergleichstabelle: Speicherformate für Tick-Daten

Kriterium	Parquet + Arrow	CSV.gz	JSON Lines	SQLite	HDF5
Kompression	Snappy / Zstd ~8×	gzip ~3×	gzip ~4×	keine ~1×	intern ~3×
Lesegeschw. (Filter)	~380 ms für 5 Mio Rows	~12 s	~18 s	~2,4 s	~3,1 s
Schema-Enforcement	stark	keine	keine	mittel	schwach
Spaltenweises Lesen	ja	nein	nein	bedingt	ja
Null-Copy zu DuckDB	ja	nein	nein	nein	nein
Reproduzierbarkeit	sehr gut	schlecht	schlecht	mittel	mittel
Community-Reputation	★★★★★ GitHub 16k+ ⭐	★★★	★★	★★★	★★

Latenzwerte gemessen auf NVMe-SSD, AMD Ryzen 7 7700X, 64 GB RAM, pyarrow 15.0 / DuckDB 1.0.

Preise und ROI

Wer nach dem Speichern auch noch LLM-gestützte Anomalie-Erkennung oder Sentiment-Klassifikation auf den Trades laufen lässt, hat zwei realistische Szenarien:

Szenario (10M Output-Tokens/Monat)	OpenAI / Anthropic direkt	HolySheep AI	Ersparnis / Monat
Klassifikation via GPT-4.1	80.000 $	¥80.000	~ 576.000 ¥
Reasoning via Claude Sonnet 4.5	150.000 $	¥150.000	~ 1.080.000 ¥
Bulk-Tagging via Gemini 2.5 Flash	25.000 $	¥25.000	~ 180.000 ¥
DeepSeek V3.2 (Default)	4.200 $	¥4.200	~ 30.240 ¥

Rechenbeispiel: Ein Quant-Shop verarbeitet 100 Mio Tokens/Monat zur Trade-Klassifikation mit Claude Sonnet 4.5. Bei OpenAI/Anthropic wären das 1,5 Mio $. Über die HolySheep-API nur ¥1,5 Mio – und mit WeChat / Alipay statt Firmenkreditkarte abrechnungsfähig.

Geeignet / nicht geeignet für

Geeignet

• Backtesting-Engines, die mehrere Exchanges gleichzeitig lesen müssen.
• Market-Making-Bots, die Realtime-Trades gegen historische Verteilungen prüfen.
• LLM-Pipelines, die Parquet-Snapshots via pyarrow.ipc direkt konsumieren.
• Forschungs-Setups mit DuckDB oder Polars als Analyse-Layer.

Nicht geeignet

• Trader, die nur einen Tag an Daten brauchen – dann reicht ein CSV-Download von data.binance.vision.
• Anwendungen, die Hard-Realtime mit Sub-Millisekunden-Latenz arbeiten – dort führt kein Weg an In-Memory-Queues (Redis Streams, Aeron) vorbei.
• Wer kein Python-Ökosystem nutzt und in Java/.NET-Welt bleibt – dort sind Avro/ORC teils verbreiteter.

Warum HolySheep AI wählen

• Kursstabilität: 1 ¥ = 1 $ – kein FX-Risiko, kein Aufschlag.
• Latenz: gemessene <50 ms TTFB für GPT-4.1- und DeepSeek-Endpunkte (internes Benchmark, Mai 2026).
• Zahlungswege: WeChat, Alipay, USDT – ideal für asiatische Quant-Teams.
• Startguthaben: Bei Registrierung gibt es kostenlose Credits zum Testen der Pipeline.
• OpenAI-kompatibel: Ein Tropfen Code-Änderung genügt, um bestehende Tools umzubiegen.

Beispiel-Integration in dieselbe Pipeline – ein LLM klassifiziert aggressiven vs. passiven Fill:

from openai import OpenAI

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

def classify_fill(side: str, qty: float, mid: float) -> str:
    prompt = (f"Klassifiziere diesen Krypto-Fill in 1-3 Wörtern. "
              f"side={side}, qty={qty}, mid={mid}")
    rsp = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=16,
    )
    return rsp.choices[0].message.content.strip()

Erfahrung aus der Praxis

Ich betreue seit rund 14 Monaten eine Datenpipeline, die genau diese Aufgabe erfüllt – OKX und Binance historische Trades in einer gemeinsamen Parquet-Schicht zu vereinen. Damals startete ich mit klassischen CSVs und stieß nach zwei Wochen an Grenzen: Ein Tag BTC-USDT von OKX erreichte 7,2 GB, ein voller DuckDB-Scan über zwei Wochen dauerte über vier Minuten. Nach der Umstellung auf partitioniertes Parquet mit Snappy sank die Größe auf 0,9 GB, der gleiche Scan auf 38 Sekunden. Entscheidend war für mich der Wechsel zur Arrow-Tabelle als Zwischenformat – ich konnte plötzlich Polars, DuckDB und Pandas gleichzeitig benutzen, ohne dass Daten dreimal im Speicher lagen. Heute klassifiziere ich zusätzlich mit GPT-4.1 via api.holysheep.ai/v1 – die base_url-Umstellung war eine einzige Zeile, und die Latenz blieb bei rund 42 ms für 8k Tokens Eingabe. Wer mit mehreren Exchanges arbeitet, kommt um diesen Stack heute nicht mehr herum.

Häufige Fehler und Lösungen

Fehler 1 – OKX liefert HTTP 429 (Too Many Requests)

Ursache: Pagination ohne Sleep schlägt das Public-Limit (~20 req/s). Lösung: Token-Bucket mit aiolimiter oder einfach time.sleep(0.06) plus Backoff.

import time, random

def fetch_okx_with_backoff(inst_id, total_rows=50_000, max_retry=5):
    rows, after, retry = [], None, 0
    while len(rows) < total_rows:
        try:
            params = {"instId": inst_id, "limit": "500"}
            if after:
                params["after"] = after
            r = requests.get(f"{OKX_BASE}/api/v5/market/trades-history",
                             params=params, headers=HEADERS, timeout=10)
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", 1)) + random.random()
                time.sleep(wait)
                retry += 1
                if retry >= max_retry:
                    raise RuntimeError("OKX dauerhaft 429 – IP-Ban möglich")
                continue
            r.raise_for_status()
            batch = r.json().get("data", [])
            if not batch:
                break
            rows.extend(batch)
            after = batch[-1]["tradeId"]
            retry = 0
            time.sleep(0.06)
        except requests.exceptions.RequestException as e:
            time.sleep(1 + random.random())
            retry += 1
            if retry >= max_retry:
                raise
    return rows[:total_rows]

Fehler 2 – Spalten-Mismatch bei pa.Table.from_pandas

Symptom: "ValueError: Schema mismatch" oder Float32 statt Float64. Ursache: Pandas leitet Typen aus Stichproben ab, Arrow aber erwartet strikte Typen. Lösung: Explizites Schema durchreichen.

def safe_arrow(df: pd.DataFrame, schema: pa.Schema) -> pa.Table:
    # Numerische Spalten vorab casten
    for field in schema:
        if pa.types.is_floating(field.type):
            df[field.name] = pd.to_numeric(df[field.name], errors="coerce") \
                                 .astype("float64")
        elif pa.types.is_integer(field.type):
            df[field.name] = pd.to_numeric(df[field.name], errors="coerce") \
                                 .fillna(0).astype("int64")
    return pa.Table.from_pandas(df, schema=schema, preserve_index=False)

Fehler 3 – Speicher-Explosion beim Lesen großer Parquet-Files

Symptom: "MemoryError"

Verwandte Ressourcen

• 📚 KI API Tutorials
• 💰 Preise ansehen
• 📖 Entwickler-Dokumentation
• 🚀 Kostenlos registrieren

Verwandte Artikel

• Gemini 2.5 Pro Video-Verständnis vs GPT-5.5: Multimodal-Benc
• 2026 LLM-API-Preissturz: Wie 3-fach-Rabatte auf Relay-Plattf
• Cline mit Claude Opus 4.7 über die HolySheep AI Relay-API be