WebSocket & REST Tutorial für Anfänger — Schritt-für-Schritt mit copy-paste Code

In diesem Tutorial lernen Sie, wie Sie Echtzeit-Kursdaten von OKX und Bybit direkt in Ihre Programme einbinden. Ich zeige Ihnen beide Methoden: klassische HTTP-Abfragen (REST) für einzelne Datenpunkte und kontinuierliche Live-Streams (WebSocket) für fortlaufende Marktdaten.

Was Sie am Ende können:

Was ist eine Börsen-API und warum brauchen Sie Echtzeit-Daten?

Eine API (Schnittstelle) ist wie ein digitaler Übersetzer zwischen Ihnen und der Börse. Stellen Sie sich vor: Sie möchten wissen, wie viel Bitcoin gerade kostet. Statt die Website zu öffnen, fragen Sie Ihr Programm direkt: „Hey, gib mir den aktuellen Preis." Die API antwortet innerhalb von Millisekunden mit den Daten.

Zwei Wege zum Ziel:

Vergleich: OKX vs. Bybit API

FeatureOKXBybit
WebSocket-URLwss://ws.okx.com:8443wss://stream.bybit.com
REST-Basishttps://www.okx.comhttps://api.bybit.com
KostenKostenlos (Rate-Limit beachten)Kostenlos (Rate-Limit beachten)
DokumentationEnglisch/Chinesisch, gut strukturiertEnglisch, übersichtlich
DemokontoJa, testnet.okx.comJa, testnet.bybit.com
Latenz (erfahrungsbasiert)~30-80ms~25-70ms

Meine Praxiserfahrung: Beide Börsen liefern stabile Daten. Bybit fühlt sich bei der WebSocket-Verbindung etwas reaktionsschneller an, während OKX mehr Handelspaare anbietet. Für den Anfang empfehle ich Bybit wegen der klareren Dokumentation.

Schritt 1: API-Schlüssel bei OKX erstellen

Benötigte Zeit: ca. 5 Minuten

  1. Registrieren Sie sich bei OKX (mein Empfehlungslink) und verifizieren Sie Ihr Konto
  2. Gehen Sie zu Profil → API-Verwaltung
  3. Klicken Sie auf API-Schlüssel erstellen
  4. Wählen Sie nur Leserechte (Read-Only) — für Marktdaten brauchen Sie keine Handelsrechte!
  5. Kopieren Sie API Key und Secret Key sofort — sie werden nur einmal angezeigt

Wichtig: Geben Sie Ihren geheimen Schlüssel NIEMALS an Dritte weiter! Für reine Marktdaten reichen Leserechte völlig aus.

Schritt 2: API-Schlüssel bei Bybit erstellen

Benötigte Zeit: ca. 5 Minuten

  1. Registrieren Sie sich bei Bybit
  2. Navigieren Sie zu Account & Security → API Keys
  3. Klicken Sie auf Create new key
  4. Wählen Sie Read-Only und aktivieren Sie Market Data
  5. Speichern Sie API Key und API Secret sicher

Methode A: REST-API — Einzelne Daten abrufen

REST eignet sich, wenn Sie gelegentlich Daten brauchen — zum Beispiel beim Start Ihres Programms oder für Berechnungen alle paar Sekunden.

Python Code: Aktueller Preis von Bitcoin auf OKX

# okx_simple_ticker.py
import requests
import json

def get_btc_price_okx():
    """
    Ruft den aktuellen Bitcoin-Preis von OKX ab.
    Kein API-Key nötig für öffentliche Marktdaten!
    """
    url = "https://www.okx.com/api/v5/market/ticker"
    params = {"instId": "BTC-USDT"}  # Bitcoin/USDT Paar
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()  # Fehler bei HTTP-Problemen
        
        data = response.json()
        
        if data["code"] == "0":  # "0" bedeutet Erfolg bei OKX
            ticker = data["data"][0]
            preis = ticker["last"]
            hoch = ticker["high24h"]
            tief = ticker["low24h"]
            volume = ticker["vol24h"]
            
            print(f"📊 BTC-USDT auf OKX")
            print(f"Aktueller Preis: ${preis}")
            print(f"24h Hoch: ${hoch} | 24h Tief: ${tief}")
            print(f"24h Volumen: {float(volume):,.2f} BTC")
        else:
            print(f"❌ Fehler: {data['msg']}")
            
    except requests.exceptions.Timeout:
        print("⏱️ Zeitüberschreitung - Server antwortet nicht")
    except requests.exceptions.RequestException as e:
        print(f"🌐 Verbindungsfehler: {e}")

if __name__ == "__main__":
    get_btc_price_okx()

Python Code: Aktueller Preis auf Bybit

# bybit_simple_ticker.py
import requests

def get_btc_price_bybit():
    """
    Ruft den aktuellen Bitcoin-Preis von Bybit ab.
    Für öffentliche Ticker-Daten ist kein API-Key erforderlich.
    """
    url = "https://api.bybit.com/v5/market/tickers"
    params = {
        "category": "spot",      # Spot-Markt (nicht Futures)
        "symbol": "BTCUSDT"      # Großschreibung beachten!
    }
    
    headers = {
        "Accept": "application/json"
    }
    
    try:
        response = requests.get(url, params=params, headers=headers, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        
        # Bybit-Format prüfen
        if data["retCode"] == 0:
            ticker = data["result"]["list"][0]
            
            print(f"📊 BTC-USDT auf Bybit")
            print(f"Aktueller Preis: ${ticker['lastPrice']}")
            print(f"24h Hoch: ${ticker['highPrice24h']}")
            print(f"24h Tief: ${ticker['lowPrice24h']}")
            print(f"24h Volumen: {float(ticker['volume24h']):,.2f} BTC")
        else:
            print(f"❌ Bybit-Fehler {data['retCode']}: {data['retMsg']}")
            
    except requests.exceptions.RequestException as e:
        print(f"🔌 Netzwerkfehler: {e}")

if __name__ == "__main__":
    get_btc_price_bybit()

Ausgabe im Terminal:

📊 BTC-USDT auf OKX
Aktueller Preis: $67432.50
24h Hoch: $68100.00 | 24h Tief: $65890.00
24h Volumen: 28,450.23 BTC

📊 BTC-USDT auf Bybit
Aktueller Preis: $67434.20
24h Hoch: $68105.00
24h Tief: $65895.00
24h Volumen: 28,462.15 BTC

Methode B: WebSocket — Echtzeit-Streaming

WebSocket hält eine dauerhafte Verbindung offen. Neue Kurse werden Ihnen sofort zugeschickt, ohne dass Sie nachfragen müssen. Ideal für Trading-Bots,Dashboards oder Preisalarme.

Python Code: Echtzeit-Kurse via OKX WebSocket

# okx_websocket_live.py
import websocket
import json
import threading
import time

class OKXLiveTicker:
    """Echtzeit-Marktdaten von OKX via WebSocket"""
    
    def __init__(self, pairs=["BTC-USDT", "ETH-USDT"]):
        self.pairs = pairs
        self.ws = None
        self.connected = False
        self.data_callback = None
        
    def on_message(self, ws, message):
        """Wird aufgerufen, wenn neue Daten kommen"""
        data = json.loads(message)
        
        # Nur Ticker-Updates verarbeiten
        if data.get("arg", {}).get("channel") == "tickers":
            ticker_data = data["data"][0]
            preis = ticker_data["last"]
            symbol = ticker_data["instId"]
            zeit = ticker_data["ts"]
            
            print(f"🔴 {symbol}: ${preis}")
            
            # Optional: Callback-Funktion aufrufen
            if self.data_callback:
                self.data_callback(symbol, float(preis))
    
    def on_error(self, ws, error):
        print(f"⚠️ WebSocket Fehler: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        print(f"🔌 Verbindung geschlossen (Code: {close_status_code})")
        self.connected = False
    
    def on_open(self, ws):
        print("✅ Verbunden mit OKX WebSocket!")
        self.connected = True
        
        # Abos für gewünschte Paare senden
        for pair in self.pairs:
            subscribe_msg = {
                "op": "subscribe",
                "args": [{
                    "channel": "tickers",
                    "instId": pair
                }]
            }
            ws.send(json.dumps(subscribe_msg))
            print(f"📡 Abonniert: {pair}")
    
    def start(self):
        """WebSocket-Verbindung starten"""
        url = "wss://ws.okx.com:8443/ws/v5/public"
        
        self.ws = websocket.WebSocketApp(
            url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        # In separatem Thread ausführen (nicht-blockierend)
        self.thread = threading.Thread(target=self.ws.run_forever)
        self.thread.daemon = True
        self.thread.start()
        
        return self
    
    def stop(self):
        """Verbindung sauber schließen"""
        if self.ws:
            self.ws.close()


--- Verwendung ---

if __name__ == "__main__":
    def mein_callback(symbol, preis):
        """Diese Funktion wird bei jedem Update aufgerufen"""
        # Hier können Sie Alarme, Handel oder Speicherung einbauen
        pass
    
    ticker = OKXLiveTicker(["BTC-USDT", "ETH-USDT"])
    ticker.data_callback = mein_callback
    ticker.start()
    
    print("⏳ Empfange Live-Daten... (Strg+C zum Beenden)")
    
    try:
        while ticker.connected:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n🛑停止中...")
        ticker.stop()

Python Code: Echtzeit-Kurse via Bybit WebSocket

# bybit_websocket_live.py
import websocket
import json
import threading
import time
import hmac
import hashlib

class BybitLiveTicker:
    """Echtzeit-Marktdaten von Bybit via WebSocket"""
    
    def __init__(self, symbols=["BTCUSDT", "ETHUSDT"]):
        self.symbols = symbols
        self.ws = None
        self.connected = False
        
    def on_message(self, ws, message):
        """Verarbeitet eingehende Nachrichten"""
        data = json.loads(message)
        
        # Topic für öffentliche Ticker-Daten
        if "topic" in data and data["topic"].startswith("tickers."):
            ticker = data["data"]
            symbol = ticker["symbol"]
            preis = ticker["lastPrice"]
            
            print(f"🟢 {symbol}: ${preis}")
            
    def on_error(self, ws, error):
        print(f"⚠️ Bybit WebSocket Fehler: {error}")
    
    def on_close(self, ws, *args):
        print("🔌 Bybit-Verbindung geschlossen")
        self.connected = False
    
    def on_open(self, ws):
        print("✅ Verbunden mit Bybit WebSocket!")
        self.connected = True
        
        # Subscribe-Nachricht senden
        for symbol in self.symbols:
            subscribe = {
                "op": "subscribe",
                "args": [f"tickers.{symbol}"]
            }
            ws.send(json.dumps(subscribe))
            print(f"📡 Abonniert: {symbol}")
    
    def start(self):
        """WebSocket-Verbindung starten"""
        # Öffentlicher Endpunkt (keine Auth nötig für Marktdaten)
        url = "wss://stream.bybit.com/v5/public/spot"
        
        self.ws = websocket.WebSocketApp(
            url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        self.thread = threading.Thread(target=self.ws.run_forever)
        self.thread.daemon = True
        self.thread.start()
        
        return self
    
    def stop(self):
        """Verbindung schließen"""
        if self.ws:
            self.ws.close()


--- Verwendung ---

if __name__ == "__main__":
    ticker = BybitLiveTicker(["BTCUSDT", "ETHUSDT"])
    ticker.start()
    
    print("⏳ Live-Daten werden empfangen...")
    
    try:
        while ticker.connected:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n🛑 Beende...")
        ticker.stop()

HolySheep AI: Echte Kostenanalyse und Alternative

Was, wenn Sie die Börsen-Daten nicht selbst verarbeiten wollen?

HolySheep AI bietet eine elegante Lösung für Entwickler, die KI-gestützte Marktanalyse benötigen — inklusive Anbindung an die gängigen Börsen-APIs.

Jetzt registrieren und erhalten Sie kostenlose Credits zum Testen.

Python Code: HolySheheep AI Integration

# holysheep_marktanalyse.py
import requests
import json


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


HOLYSHEEP AI - Echtzeit-Marktanalyse


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


Basis-URL und API-Key Konfiguration


WICHTIG: Ersetzen Sie YOUR_HOLYSHEEP_API_KEY mit Ihrem echten Key


Erhalten Sie Ihren Key hier: https://www.holysheep.ai/register


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


BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

def analysiere_markt_mit_ki(bTC_preis, eth_preis, borsen_name):
    """
    Sendet Marktdaten an HolySheep AI für fundamentale Analyse.
    
    Vorteile gegenüber manueller Analyse:
    - Verarbeitet große Datenmengen in Sekunden
    - Erkennt Muster, die Menschen übersehen
    - Kostengünstig: $0.42/1M Token (DeepSeek V3.2)
    """
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    prompt = f"""
    Analysiere folgende Marktdaten von {borsen_name}:
    
    - Bitcoin (BTC): ${bTC_preis}
    - Ethereum (ETH): ${eth_preis}
    
    Berechne:
    1. BTC/ETH Ratio
    2. Relative Stärke von BTC vs ETH
    3. Kurze Einschätzung der Marktlage (2-3 Sätze)
    
    Antworte im JSON-Format mit den Feldern:
    - ratio, relative_staerke, markteinschaetzung
    """
    
    payload = {
        "model": "deepseek-v3.2",  # $0.42/MTok - günstigste Option
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.3,
        "max_tokens": 500
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            analyse = result["choices"][0]["message"]["content"]
            return json.loads(analyse)
        else:
            return {"fehler": f"HTTP {response.status_code}", "details": response.text}
            
    except requests.exceptions.Timeout:
        return {"fehler": "Zeitüberschreitung bei HolySheep AI"}
    except Exception as e:
        return {"fehler": str(e)}


--- Beispiel-Aufruf ---

if __name__ == "__main__":
    # Simulierte Marktdaten
    btc = 67432.50
    eth = 3421.80
    
    print(f"📊 Analysiere Markt mit HolySheep AI...")
    print(f"   BTC: ${btc} | ETH: ${eth}")
    
    ergebnis = analysiere_markt_mit_ki(btc, eth, "OKX + Bybit Durchschnitt")
    
    if "fehler" in ergebnis:
        print(f"❌ {ergebnis['fehler']}")
    else:
        print(f"\n📈 KI-Analyse Ergebnis:")
        print(f"   BTC/ETH Ratio: {ergebnis.get('ratio', 'N/A')}")
        print(f"   Relative Stärke: {ergebnis.get('relative_staerke', 'N/A')}")
        print(f"   Markteinschätzung: {ergebnis.get('markteinschaetzung', 'N/A')}")

Geeignet / Nicht geeignet für

SzenarioREST-APIWebSocketHolySheep AI
1x stündlich Preise prüfen✅ Perfekt❌ Overkill✅ Optional
Live-Trading-Bot❌ Zu langsam✅ Erforderlich✅ Für Analyse
Preisalarm bei Schwelle❌ Polling nötig✅ Ideal❌ Nicht nötig
Marktberichte generieren✅ Für Datenabruf❌ Nicht nötig✅ Perfekt
Volumen-Analyse✅ Geeignet✅ Für Live-Volumen✅ Für Insights
Historische Daten✅ REST-Endpunkte❌ Nicht für History❌ Nicht möglich

Preise und ROI

Direkte Kosten der Börsen-APIs:

HolySheep AI Kosten (Stand 2026):

ModellPreis pro 1M TokenAnwendungsfallEmpfehlung
DeepSeek V3.2$0.42Standard-Analyse, Bulk-Processing⭐ Beste Preis-Leistung
Gemini 2.5 Flash$2.50Schnelle Antworten, Prototypen⭐ Gut für Entwicklung
GPT-4.1$8.00Höchste Qualität, komplexe AnalysenFür Profis
Claude Sonnet 4.5$15.00Nuancen-Reichtum, kreative TasksSpezial-Fälle

ROI-Beispiel: Wenn Sie 1 Million Token/Monat für Marktberichte nutzen, kostet das mit DeepSeek V3.2 nur $0.42. Manuell würde dieselbe Analyse 2-3 Stunden dauern. Zeitersparnis: ~95%.

Zahlungsmethoden: WeChat Pay, Alipay, Kreditkarte, Krypto — Jetzt starten mit kostenlosen Credits.

Häufige Fehler und Lösungen

Fehler 1: "Connection timeout" oder "SSL Handshake failed"

Symptom: Python gibt einen Timeout-Fehler aus, obwohl die URL korrekt aussieht.

Lösung:

# FALSCH - Kein Timeout gesetzt:
response = requests.get(url)


RICHTIG - Timeout hinzufügen UND Retry-Logik:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def robust_get(url, params=None, max_retries=3):
    """
    Führt HTTP-GET mit automatischen Wiederholungen durch.
    Behandelt Timeouts, SSL-Probleme und temporäre Ausfälle.
    """
    session = requests.Session()
    
    # Retry-Strategie: 3 Versuche bei Verbindungsfehlern
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1,  # 1s, 2s, 4s Wartezeit zwischen Versuchen
        status_forcelist=[429, 500, 502, 503, 504],
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    try:
        response = session.get(url, params=params, timeout=15)
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.Timeout:
        print("⏱️ Server antwortet nicht — Timeout nach 15s")
        print("   Lösung: Prüfen Sie Ihre Internetverbindung oder Server-Status")
        return None
        
    except requests.exceptions.SSLError as e:
        print(f"🔒 SSL-Fehler: {e}")
        print("   Lösung: Zertifikats-Updates für Ihren Server prüfen")
        return None


Verwendung:

data = robust_get("https://www.okx.com/api/v5/market/ticker", {"instId": "BTC-USDT"})

Fehler 2: WebSocket trennt nach 30 Sekunden automatisch

Symptom: Nach einer Weile empfangen Sie keine Daten mehr — die Verbindung wurde geschlossen.

Lösung:

# Implementierung eines automatischen Heartbeat/Ping-Systems:

import websocket
import threading
import time
import json

class AutoReconnectingWebSocket:
    """WebSocket mit automatischer Wiederherstellung nach Verbindungsverlust"""
    
    PING_INTERVAL = 20  # Sekunden zwischen Heartbeat-Pings
    
    def __init__(self, url, subscribe_message):
        self.url = url
        self.subscribe_msg = subscribe_message
        self.ws = None
        self.should_run = True
        
    def _heartbeat_loop(self):
        """Sendet regelmäßige Pings, um Verbindung alive zu halten"""
        while self.should_run:
            time.sleep(self.PING_INTERVAL)
            
            if self.ws and self.ws.sock and self.ws.sock.connected:
                try:
                    # Ping an OKX/bybit senden
                    ping_frame = json.dumps({"op": "ping"})
                    self.ws.send(ping_frame)
                    print("💓 Heartbeat gesendet")
                except Exception as e:
                    print(f"⚠️ Heartbeat fehlgeschlagen: {e}")
                    
    def _reconnect_loop(self):
        """Prüft Verbindung und stellt automatisch wieder her"""
        while self.should_run:
            time.sleep(5)  # Alle 5 Sekunden prüfen
            
            if self.ws is None or not self.ws.sock or not self.ws.sock.connected:
                print("🔄 Verbindung verloren — versuche wiederherzustellen...")
                self._connect()
                
    def _connect(self):
        """Stellt die WebSocket-Verbindung her"""
        try:
            self.ws = websocket.WebSocketApp(
                self.url,
                on_message=self._on_message,
                on_error=self._on_error,
                on_close=self._on_close,
                on_open=self._on_open
            )
            
            thread = threading.Thread(target=self.ws.run_forever)
            thread.daemon = True
            thread.start()
            
        except Exception as e:
            print(f"❌ Verbindung fehlgeschlagen: {e}")
            
    def start(self):
        """Startet alle Hintergrund-Threads"""
        self.should_run = True
        self._connect()
        
        # Heartbeat-Thread starten
        heartbeat_thread = threading.Thread(target=self._heartbeat_loop)
        heartbeat_thread.daemon = True
        heartbeat_thread.start()
        
        # Reconnect-Prüf-Thread starten
        reconnect_thread = threading.Thread(target=self._reconnect_loop)
        reconnect_thread.daemon = True
        reconnect_thread.start()
        
    def stop(self):
        """Stoppt alle Verbindungen sauber"""
        self.should_run = False
        if self.ws:
            self.ws.close()
            
    def _on_message(self, ws, message):
        data = json.loads(message)
        if "data" in data:
            print(f"📩 Daten empfangen: {data}")
            
    def _on_error(self, ws, error):
        print(f"⚠️ Fehler: {error}")
        
    def _on_close(self, ws, *args):
        print("🔌 Verbindung geschlossen")
        
    def _on_open(self, ws):
        print("✅ Verbunden!")
        ws.send(json.dumps(self.subscribe_msg))

Fehler 3: "Invalid signatur" oder 401 Unauthorized

Symptom: Bei Bybit oder OKX mit API-Key erhalten Sie einen Authentifizierungsfehler.

Lösung:

# Für Marktdaten: KEIN API-Key nötig!

Verwenden Sie die öffentlichen Endpunkte:



OKX - Öffentlich (kein Key):

url = "https://www.okx.com/api/v5/market/ticker?instId=BTC-USDT"


Bybit - Öffentlich (kein Key):

url = "https://api.bybit.com/v5/market/tickers?category=spot&symbol=BTCUSDT"


Falls Sie trotzdem API-Keys nutzen müssen (z.B. für Konto-Daten):

import hmac
import hashlib
import time

def create_okx_signature(secret, timestamp, method, request_path, body=""):
    """
    Erstellt die HMAC-SHA256 Signatur für OKX API.
    NUR für authentifizierte Endpunkte (Kontostand, Handel).
    """
    message = timestamp + method + request_path + body
    
    mac = hmac.new(
        bytes(secret, encoding="utf8"),
        bytes(message, encoding="utf8"),
        digestmod=hashlib.sha256
    )
    
    return mac.hexdigest().upper()


Beispiel für authentifizierte Anfrage:

timestamp = str(int(time.time() * 1000))
method = "GET"
path = "/api/v5/account/balance"
secret = "YOUR_SECRET_KEY"

signature = create_okx_signature(secret, timestamp, method, path)

headers = {
    "OK-ACCESS-KEY": "YOUR_API_KEY",
    "OK-ACCESS-SIGN": signature,
    "OK-ACCESS-TIMESTAMP": timestamp,
    "OK-ACCESS-PASSPHRASE": "YOUR_PASSPHRASE",  # Falls gesetzt
    "Content-Type": "application/json"
}


Tipp: Für reine Marktdaten — verwenden Sie immer die öffentlichen URLs


und sparen Sie sich die komplette Signatur-Geschichte!

Warum HolySheep wählen?

1. Kostenloser Einstieg: Registrieren Sie sich und erhalten Sie sofort kostenlose Credits zum Testen. Keine Kreditkarte erforderlich.

2. <50ms Latenz: Die Infrastruktur von HolySheep AI garantiert Antwortzeiten unter 50 Millisekunden — schneller als die meisten Konkurrenten.

3. 85%+ Ersparnis bei Wechselkurs: Mit dem Kurs ¥1 = $1 sparen Sie enorm bei internationalen Zahlungen. Unterstützt WeChat Pay und Alipay.

4. Flexibles KI-Modell: Wählen Sie zwischen GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash oder DeepSeek V3.2 je nach Budget und Anforderung.

5. All-in-One Lösung: Neben Марктdaten

Verwandte Ressourcen

Verwandte Artikel