Sie möchten Kryptowährungs-Daten automatisiert abrufen und damit Ihre eigene Trading-Strategie testen? Dann sind Sie hier genau richtig. In diesem umfassenden Tutorial zeige ich Ihnen Schritt für Schritt, wie Sie die OKX Exchange API nutzen, um historische Marktdaten zu erhalten und damit Backtests durchzuführen.

📌 Hinweis für Einsteiger: Dieser Leitfaden setzt keinerlei Vorwissen voraus. Ich erkläre jeden Begriff und jede Codezeile verständlich. Los geht's!

Was ist eine API und warum ist sie wichtig?

Bevor wir mit dem Programmieren beginnen, klären wir eine grundlegende Frage: Was ist überhaupt eine API?

Stellen Sie sich eine API wie einen Kellner in einem Restaurant vor. Sie (der Nutzer) sitzen am Tisch und möchten etwas bestellen. Der Kellner nimmt Ihre Bestellung entgegen und bringt Ihnen das Essen. Ohne den Kellner müssten Sie direkt in die Küche gehen und dort mit dem Koch kommunizieren – was kompliziert und chaotisch wäre.

Eine API (Application Programming Interface) funktioniert genauso: Sie ist der Vermittler zwischen Ihnen und einem Datenanbieter (hier: OKX-Börse). Anstatt komplizierte Anfragen direkt an die Server zu senden, nutzen Sie die API als standardisierte Schnittstelle.

Warum OKX für historische Krypto-Daten?

Voraussetzungen für dieses Tutorial

Bevor wir starten, benötigen Sie folgende Dinge:

Schritt 1: Python-Umgebung einrichten

Zunächst installieren wir die notwendigen Python-Bibliotheken. Öffnen Sie Ihr Terminal (CMD unter Windows, Terminal unter macOS/Linux) und geben Sie folgenden Befehl ein:

pip install requests pandas numpy matplotlib python-dotenv

Diese Bibliotheken haben folgende Funktionen:

Schritt 2: Die OKX Public API verstehen

OKX bietet verschiedene API-Endpunkte. Für historische Kursdaten nutzen wir den Endpunkt für K-Linien-Daten (Candlestick/Kerzenchart-Daten). Die Basis-URL für die öffentliche API lautet:

https://www.okx.com

Der spezifische Endpunkt für K-Linien-Daten ist:

/api/v5/market/history-candles

Schritt 3: Erste API-Abfrage durchführen

Erstellen Sie eine neue Python-Datei namens okx_data.py und fügen Sie folgenden Code ein:

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

============================================

KONFIGURATION - HIER ANPASSEN

============================================

BASE_URL = "https://www.okx.com" INSTRUMENT_ID = "BTC-USDT" # Trading-Paar (Bitcoin zu USDT) TIMEFRAME = "1H" # 1H = 1 Stunde, auch möglich: 1m, 5m, 15m, 1H, 4H, 1D LIMIT = 100 # Anzahl der Datenpunkte (max. 100 pro Anfrage) def get_ohlcv_data(instrument_id, timeframe, limit=100): """ Ruft OHLCV-Daten (Open, High, Low, Close, Volume) von OKX ab. Args: instrument_id: Das Trading-Paar, z.B. "BTC-USDT" timeframe: Zeitrahmen, z.B. "1H" für stündliche Daten limit: Anzahl der abzurufenden Kerzen (max. 100) Returns: DataFrame mit den Kursdaten oder None bei Fehler """ endpoint = f"{BASE_URL}/api/v5/market/history-candles" params = { "instId": instrument_id, "bar": timeframe, "limit": limit } try: response = requests.get(endpoint, params=params) response.raise_for_status() # Wirft einen Fehler bei HTTP-Fehlern data = response.json() # Prüfe, ob die API-Anfrage erfolgreich war if data.get("code") == "0": candles = data["data"] # Daten in ein DataFrame umwandeln df = pd.DataFrame(candles, columns=[ "timestamp", "open", "high", "low", "close", "volume", "quote_volume" ]) # Datentypen konvertieren for col in ["open", "high", "low", "close", "volume", "quote_volume"]: df[col] = pd.to_numeric(df[col], errors='coerce') # Timestamp in lesbare Datumszeit umwandeln df["datetime"] = pd.to_datetime( df["timestamp"].astype(float), unit="ms" ) # Spalten neu anordnen df = df[["datetime", "open", "high", "low", "close", "volume", "quote_volume"]] return df else: print(f"API-Fehler: {data.get('msg')}") return None except requests.exceptions.RequestException as e: print(f"Netzwerkfehler: {e}") return None

============================================

DATEN ABRUFEN

============================================

if __name__ == "__main__": print("Rufe Kryptowährungsdaten von OKX ab...") df = get_ohlcv_data(INSTRUMENT_ID, TIMEFRAME, LIMIT) if df is not None: print(f"\n✅ Erfolgreich {len(df)} Datenpunkte abgerufen!") print(f"Zeitraum: {df['datetime'].min()} bis {df['datetime'].max()}") print("\n--- Vorschau der ersten 5 Datenpunkte ---") print(df.head()) print("\n--- Statistik ---") print(df[["open", "high", "low", "close", "volume"]].describe()) # Daten als CSV speichern df.to_csv("btc_usdt_data.csv", index=False) print("\n💾 Daten wurden als 'btc_usdt_data.csv' gespeichert!") else: print("❌ Fehler beim Abrufen der Daten.")

Führen Sie das Skript aus mit:

python okx_data.py

Sie sollten eine Ausgabe ähnlich dieser sehen:

Rufe Kryptowährungsdaten von OKX ab...
✅ Erfolgreich 100 Datenpunkte abgerufen!
Zeitraum: 2026-01-15 14:00:00 bis 2026-01-19 17:00:00

--- Vorschau der ersten 5 Datenpunkte ---
             datetime        open        high         low       close      volume  quote_volume
0 2026-01-15 14:00:00  96500.50   96800.30   96200.10   96650.25   1250.50   120854000.00
1 2026-01-15 15:00:00  96650.25   97100.00   96500.00   97000.50   1340.20   129850000.00
2 2026-01-15 16:00:00  97000.50   97450.75   96800.00   97200.30   1180.90   114560000.00

--- Statistik ---
                  open        high         low       close       volume
count   100.000000  100.000000  100.000000  100.000000  100.000000
mean    97250.35    97680.45    96850.20    97350.50    1285.60
std     1520.30     1580.45     1450.20     1550.60     320.45
min     94200.10    94500.30    93800.00    94100.50    850.20
max    100500.80   101200.50   99900.20    100850.30   1980.40

💾 Daten wurden als 'btc_usdt_data.csv' gespeichert!

Schritt 4: Mehr historische Daten abrufen (Backtesting-Vorbereitung)

Für ein aussagekräftiges Backtesting benötigen Sie in der Regel mehr als 100 Datenpunkte. Da die OKX-API auf 100 Datenpunkte pro Anfrage begrenzt ist, müssen wir die Anfragen in Schleifen ausführen und die Daten zusammenfügen.

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

============================================

ERWEITERTE KONFIGURATION

============================================

BASE_URL = "https://www.okx.com" INSTRUMENT_ID = "BTC-USDT" TIMEFRAME = "1H" TOTAL_DAYS = 90 # Anzahl der Tage für das Backtesting def get_historical_data(instrument_id, timeframe, days=90): """ Ruft historische Daten für einen längeren Zeitraum ab. Da die API max. 100 Punkte pro Anfrage liefert, werden mehrere Anfragen mit zeitlicher Filterung durchgeführt. Args: instrument_id: Trading-Paar timeframe: Zeitrahmen days: Anzahl der vergangenen Tage Returns: DataFrame mit allen historischen Daten """ all_data = [] end_time = int(datetime.now().timestamp() * 1000) # Berechne Startzeit basierend auf der gewünschten Tageanzahl # Multipliziere mit 2, da wir nicht alle Daten auf einmal bekommen start_time = int((datetime.now() - timedelta(days=days * 2)).timestamp() * 1000) print(f"Rufe Daten ab für: {days} Tage historische Daten...") while True: # API-Anfrage mit Zeitfilter params = { "instId": instrument_id, "bar": timeframe, "limit": 100, "after": str(end_time), # Ende des Zeitraums (abwärts) "before": str(start_time) # Anfang des Zeitraums (aufwärts) } try: response = requests.get( f"{BASE_URL}/api/v5/market/history-candles", params=params, timeout=30 ) response.raise_for_status() data = response.json() if data.get("code") != "0": print(f"API-Fehler: {data.get('msg')}") break candles = data["data"] if not candles: print("Keine weiteren Daten verfügbar.") break all_data.extend(candles) print(f" ➕ {len(candles)} Datenpunkte abgerufen (Gesamt: {len(all_data)})") # Der letzte Timestamp ist der älteste Punkt # Verwende ihn als neues "after" für die nächste Iteration end_time = int(candles[-1][0]) # Rate Limiting: Kurze Pause zwischen Anfragen time.sleep(0.2) # Sicherheitslimit if len(all_data) >= days * 24 * 30: # Max. ca. 30 Tage pro Stunde print("⚠️ Sicherheitslimit erreicht.") break except requests.exceptions.RequestException as e: print(f"Netzwerkfehler: {e}") time.sleep(5) # Warte bei Fehlern länger continue if not all_data: return None # DataFrame erstellen df = pd.DataFrame(all_data, columns=[ "timestamp", "open", "high", "low", "close", "volume", "quote_volume" ]) # Duplikate entfernen und sortieren df = df.drop_duplicates(subset=["timestamp"]) df = df.sort_values("timestamp") # Datentypen konvertieren for col in ["open", "high", "low", "close", "volume", "quote_volume"]: df[col] = pd.to_numeric(df[col], errors='coerce') df["datetime"] = pd.to_datetime( df["timestamp"].astype(float), unit="ms" ) df = df[["datetime", "open", "high", "low", "close", "volume", "quote_volume"]] # Auf den gewünschten Zeitraum filtern cutoff_date = datetime.now() - timedelta(days=days) df = df[df["datetime"] >= cutoff_date] return df.reset_index(drop=True)

============================================

HISTORISCHE DATEN ABRUFEN

============================================

if __name__ == "__main__": print("=" * 50) print("OKX HISTORISCHE DATEN SAMMLER") print("=" * 50) df = get_historical_data(INSTRUMENT_ID, TIMEFRAME, TOTAL_DAYS) if df is not None: print(f"\n✅ Gesamt: {len(df)} Datenpunkte abgerufen!") print(f"Zeitraum: {df['datetime'].min()} bis {df['datetime'].max()}") # Speichern für spätere Analysen filename = f"{INSTRUMENT_ID.lower().replace('-', '_')}_{TOTAL_DAYS}d.csv" df.to_csv(filename, index=False) print(f"\n💾 Daten gespeichert als: {filename}") else: print("❌ Keine Daten erhalten.")

Schritt 5: Backtesting implementieren

Nun kommen wir zum spannenden Teil: dem Backtesting. Backtesting bedeutet, dass wir eine Trading-Strategie mit historischen Daten testen, um herauszufinden, ob sie profitabel gewesen wäre.

Beispiel-Strategie: Einfacher gleitender Durchschnitt (SMA Crossover)

Eine beliebte Einsteiger-Strategie ist der Crossover der gleitenden Durchschnitte:

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime

============================================

KONFIGURATION

============================================

INITIAL_CAPITAL = 10000 # Starting capital in USDT TRADING_FEE = 0.001 # 0.1% Trading fee SMA_SHORT = 10 # Short-term SMA (periods) SMA_LONG = 30 # Long-term SMA (periods) FILENAME = "btc_usdt_90d.csv" # Our previously saved data

============================================

STRATEGIE DEFINITION

============================================

def calculate_sma(data, period): """ Berechnet den Simple Moving Average (SMA). Args: data: Datenreihe (z.B. Schlusskurse) period: Anzahl der Perioden für den Durchschnitt Returns: SMA-Werte """ return data.rolling(window=period).mean() def generate_signals(df, sma_short, sma_long): """ Generiert Trading-Signale basierend auf SMA-Crossover. Args: df: DataFrame mit OHLCV-Daten sma_short: Periode für kurzen SMA sma_long: Periode für langen SMA Returns: DataFrame mit zusätzlichen Signalkolonnen """ df = df.copy() # SMA berechnen df[f"SMA_{sma_short}"] = calculate_sma(df["close"], sma_short) df[f"SMA_{sma_long}"] = calculate_sma(df["close"], sma_long) # Signale generieren # 1 = Long-Position, 0 = Keine Position, -1 = Short (hier nicht verwendet) df["signal"] = 0 df.loc[df[f"SMA_{sma_short}"] > df[f"SMA_{sma_long}"], "signal"] = 1 # Signal-Wechsel erkennen df["position_change"] = df["signal"].diff() # Kaufsignal (signal wechselt von 0 auf 1) df["buy_signal"] = df["position_change"] == 1 # Verkaufssignal (signal wechselt von 1 auf 0) df["sell_signal"] = df["position_change"] == -1 return df def backtest_strategy(df, initial_capital, trading_fee): """ Führt das Backtesting durch. Args: df: DataFrame mit Signalen initial_capital: Starting capital trading_fee: Handelsgebühr (in Prozent) Returns: Dictionary mit Backtesting-Ergebnissen """ df = df.copy() # Starting values capital = initial_capital position = 0 # Anzahl der Assets entry_price = 0 # Track trades trades = [] # Portfolio value over time portfolio_values = [] for i, row in df.iterrows(): current_price = row["close"] # Kaufsignal if row["buy_signal"] and position == 0: # Alles investieren (abzüglich Gebühren) position = (capital * (1 - trading_fee)) / current_price entry_price = current_price capital = 0 trades.append({ "datetime": row["datetime"], "type": "BUY", "price": current_price, "quantity": position, "capital_before": capital + (position * current_price) }) # Verkaufssignal elif row["sell_signal"] and position > 0: capital = position * current_price * (1 - trading_fee) profit = (capital - initial_capital) / initial_capital * 100 trades.append({ "datetime": row["datetime"], "type": "SELL", "price": current_price, "quantity": position, "capital_after": capital, "profit_pct": profit }) position = 0 # Portfolio-Wert berechnen portfolio_value = capital + (position * current_price) portfolio_values.append({ "datetime": row["datetime"], "value": portfolio_value }) # Finale Berechnungen final_value = capital + (position * df.iloc[-1]["close"]) total_return = (final_value - initial_capital) / initial_capital * 100 # Buy & Hold Vergleich buy_hold_return = (df.iloc[-1]["close"] - df.iloc[0]["open"]) / df.iloc[0]["open"] * 100 # Anzahl der Trades num_trades = len([t for t in trades if t["type"] == "BUY"]) # Gewinnrate sell_trades = [t for t in trades if t["type"] == "SELL"] profitable_trades = len([t for t in sell_trades if t["profit_pct"] > 0]) win_rate = (profitable_trades / len(sell_trades) * 100) if sell_trades else 0 return { "initial_capital": initial_capital, "final_value": final_value, "total_return": total_return, "buy_hold_return": buy_hold_return, "num_trades": num_trades, "win_rate": win_rate, "trades": trades, "portfolio_values": portfolio_values, "df": df } def plot_results(results): """ Erstellt Visualisierungen der Backtesting-Ergebnisse. """ df = results["df"] portfolio_df = pd.DataFrame(results["portfolio_values"]) fig, axes = plt.subplots(3, 1, figsize=(14, 12)) fig.suptitle("Backtesting Results: SMA Crossover Strategie", fontsize=16, fontweight='bold') # Plot 1: Preis und SMAs ax1 = axes[0] ax1.plot(df["datetime"], df["close"], label="Schlusskurs", alpha=0.7, linewidth=1) ax1.plot(df["datetime"], df[f"SMA_{SMA_SHORT}"], label=f"SMA {SMA_SHORT}", linewidth=1.5, linestyle="--") ax1.plot(df["datetime"], df[f"SMA_{SMA_LONG}"], label=f"SMA {SMA_LONG}", linewidth=1.5, linestyle="--") # Kaufsignale markieren buy_signals = df[df["buy_signal"]] ax1.scatter(buy_signals["datetime"], buy_signals["close"], marker="^", color="green", s=100, label="Kauf", zorder=5) # Verkaufssignale markieren sell_signals = df[df["sell_signal"]] ax1.scatter(sell_signals["datetime"], sell_signals["close"], marker="v", color="red", s=100, label="Verkauf", zorder=5) ax1.set_ylabel("Preis (USDT)") ax1.set_title("Kursverlauf mit SMA-Crossover Signalen") ax1.legend(loc="upper left") ax1.grid(True, alpha=0.3) ax1.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d")) # Plot 2: Portfolio-Wert ax2 = axes[1] ax2.plot(portfolio_df["datetime"], portfolio_df["value"], label="Portfolio-Wert", color="blue", linewidth=2) ax2.axhline(y=INITIAL_CAPITAL, color="gray", linestyle="--", label=f"Startkapital: ${INITIAL_CAPITAL:,.0f}") ax2.fill_between(portfolio_df["datetime"], INITIAL_CAPITAL, portfolio_df["value"], where=portfolio_df["value"] > INITIAL_CAPITAL, alpha=0.3, color="green", label="Gewinn") ax2.fill_between(portfolio_df["datetime"], INITIAL_CAPITAL, portfolio_df["value"], where=portfolio_df["value"] < INITIAL_CAPITAL, alpha=0.3, color="red", label="Verlust") ax2.set_ylabel("Wert (USDT)") ax2.set_title("Portfolio-Entwicklung über Zeit") ax2.legend(loc="upper left") ax2.grid(True, alpha=0.3) ax2.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d")) # Plot 3: Return-Diagramm ax3 = axes[2] portfolio_df["returns"] = portfolio_df["value"].pct_change() * 100 ax3.bar(portfolio_df["datetime"], portfolio_df["returns"], color=["green" if r > 0 else "red" for r in portfolio_df["returns"].fillna(0)], alpha=0.7, width=0.8) ax3.axhline(y=0, color="black", linewidth=0.5) ax3.set_ylabel("Tägliche Returns (%)") ax3.set_xlabel("Datum") ax3.set_title("Tägliche Portfolio-Returns") ax3.grid(True, alpha=0.3, axis="y") ax3.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d")) plt.tight_layout() plt.savefig("backtest_results.png", dpi=150, bbox_inches="tight") plt.show() print("📊 Chart gespeichert als: backtest_results.png")

============================================

BACKTESTING AUSFÜHREN

============================================

if __name__ == "__main__": print("=" * 60) print("SMA CROSSOVER BACKTESTING") print("=" * 60) # Daten laden print(f"\n📂 Lade Daten aus: {FILENAME}") df = pd.read_csv(FILENAME, parse_dates=["datetime"]) print(f" Geladen: {len(df)} Datenpunkte") print(f" Zeitraum: {df['datetime'].min()} bis {df['datetime'].max()}") # Signale generieren print(f"\n🔧 Generiere Signale (SMA {SMA_SHORT} / SMA {SMA_LONG})...") df = generate_signals(df, SMA_SHORT, SMA_LONG) # Backtesting print("\n🚀 Starte Backtesting...") results = backtest_strategy(df, INITIAL_CAPITAL, TRADING_FEE) # Ergebnisse ausgeben print("\n" + "=" * 60) print("📊 ERGEBNISSE") print("=" * 60) print(f"Startkapital: ${results['initial_capital']:,.2f}") print(f"Endwert: ${results['final_value']:,.2f}") print(f"Rendite Strategie: {results['total_return']:+.2f}%") print(f"Rendite Buy & Hold: {results['buy_hold_return']:+.2f}%") print(f"Anzahl Trades: {results['num_trades']}") print(f"Win-Rate: {results['win_rate']:.1f}%") print(f"Outperformance: {results['total_return'] - results['buy_hold_return']:+.2f}%") # Trades anzeigen print("\n" + "-" * 60) print("📋 TRADE-HISTORIE") print("-" * 60) for i, trade in enumerate(results["trades"], 1): if trade["type"] == "BUY": print(f"{i}. {trade['datetime'].strftime('%Y-%m-%d %H:%M')} | " f"KAUF @ ${trade['price']:,.2f} | " f"Menge: {trade['quantity']:.6f} BTC") else: print(f"{i}. {trade['datetime'].strftime('%Y-%m-%d %H:%M')} | " f"VERKAUF @ ${trade['price']:,.2f} | " f"Profit: {trade['profit_pct']:+.2f}%") # Visualisierung print("\n📈 Erstelle Charts...") plot_results(results) # Daten speichern mit Signalen df.to_csv("backtest_data_with_signals.csv", index=False) print("💾 Daten mit Signalen gespeichert als: backtest_data_with_signals.csv")

Schritt 6: Ergebnisse interpretieren

Nach Ausführung des Backtesting-Skripts erhalten Sie eine umfassende Analyse. Hier ein Beispiel für mögliche Ergebnisse:

============================================================
📊 ERGEBNISSE
============================================================
Startkapital:        $10,000.00
Endwert:             $12,450.50
Rendite Strategie:   +24.51%
Rendite Buy & Hold:  +18.30%
Anzahl Trades:       6
Win-Rate:            66.7%
Outperformance:      +6.21%

📋 TRADE-HISTORIE
------------------------------------------------------------
1. 2025-10-15 09:00 | KAUF @ $62,500.00 | Menge: 0.1592 BTC
2. 2025-11-20 14:00 | VERKAUF @ $68,200.00 | Profit: +9.12%
3. 2025-12-05 11:00 | KAUF @ $67,800.00 | Menge: 0.1476 BTC
4. 2026-01-02 16:00 | VERKAUF @ $72,500.00 | Profit: +6.93%
5. 2026-01-10 10:00 | KAUF @ $71,200.00 | Menge: 0.1392 BTC
6. 2026-01-15 13:00 | VERKAUF @ $74,800.00 | Profit: +5.06%

HolySheep AI Integration: KI-gestützte Strategieoptimierung

Nachdem Sie nun Ihre erste Backtesting-Strategie erstellt haben, fragen Sie sich vielleicht: Wie kann ich diese Strategie weiter verbessern? Hier kommt HolySheep AI ins Spiel.

Mit HolySheep können Sie:

import requests
import json

============================================

HOLYSHEEP AI API INTEGRATION

============================================

BASE_URL = "https://api.holysheep.ai/v1" API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key def analyze_strategy_with_ai(strategy_description, backtest_results): """ Nutzt HolySheep AI, um eine Trading-Strategie zu analysieren und Optimierungsvorschläge zu erhalten. Args: strategy_description: Textuelle Beschreibung der Strategie backtest_results: Dictionary mit Backtesting-Ergebnissen Returns: KI-generierte Analyse und Empfehlungen """ headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # Erstelle einen strukturierten Prompt prompt = f""" Als erfahrener Krypto-Trading-Analyst, analysiere bitte folgende Backtesting-Ergebnisse und gib konkrete Verbesserungsvorschläge. STRATEGIE: {strategy_description} BACKTESTING-ERGEBNISSE: - Startkapital: ${backtest_results['initial_capital']:,.2f} - Endwert: ${backtest_results['final_value']:,.2f} - Gesamtrendite: {backtest_results['total_return']:+.2f}% - Buy & Hold Rendite: {backtest_results['buy_hold_return']:+.2f}% - Anzahl Trades: {backtest_results['num_trades']} - Win-Rate: {backtest_results['win_rate']:.1f}% Bitte gib mir: 1. Eine Bewertung der Strategie-Performance 2. Identifizierte Stärken und Schwächen 3. Konkrete Optimierungsvorschläge 4. Risikoeinschätzung """ payload = { "model": "gpt-4.1", "messages": [ {"role": "system", "content": "Du bist ein erfahrener Krypto-Trading-Analyst mit Fokus auf Backtesting-Optimierung."}, {"role": "user", "content": prompt} ], "temperature": 0.7, "max_tokens": 2000 } try: response