Veröffentlicht: 15. Januar 2026 | Lesezeit: 12 Minuten | Schwierigkeit: Einsteiger

Einleitung: Was ist ein WebSocket und warum brauchen Sie ihn?

Bevor wir in die technischen Details eintauchen, klären wir eine wichtige Frage: Was bedeutet "WebSocket" eigentlich?

Stellen Sie sich vor, Sie möchten die aktuellen Bitcoin-Kurse in Echtzeit auf Ihrer Website anzeigen. Bei einer herkömmlichen Methode (REST-API) müssten Sie alle paar Sekunden beim Server nachfragen: "Gibt es neue Daten?" – das ist wie ständiges Telefonieren. Ein WebSocket funktioniert hingegen wie eine permanente Telefonleitung: Der Server ruft Sie automatisch an, sobald sich etwas ändert.

In diesem Tutorial lernen Sie, wie Sie die OKX-Börsen-WebSocket-Verbindung einrichten, um Krypto-Kursdaten in Echtzeit zu empfangen – ganz ohne Vorwissen über Programmierung!

💡 Pro-Tipp: Für KI-gestützte Analyse dieser Echtzeit-Daten empfehle ich die HolySheep AI Plattform, die eine latency von unter 50ms bietet und über 85% günstiger ist als herkömmliche Lösungen.

Voraussetzungen: Was Sie benötigen

Schritt 1: Die richtige Umgebung vorbereiten

Öffnen Sie Ihr Terminal (bei Windows: "Eingabeaufforderung" oder "PowerShell", bei Mac/Linux: "Terminal") und geben Sie folgenden Befehl ein:

# Erstellen Sie einen neuen Ordner für Ihr Projekt
mkdir okx-websocket-tutorial
cd okx-websocket-tutorial


Erstellen Sie eine virtuelle Umgebung (isoliert Ihr Projekt)

python -m venv mein-env


Aktivieren Sie die Umgebung


Bei Windows:

mein-env\Scripts\activate

Bei Mac/Linux:

source mein-env/bin/activate


Installieren Sie die benötigte Bibliothek

pip install websocket-client

📸 Screenshot-Hinweis: Ihr Terminal sollte nach erfolgreicher Installation etwa so aussehen: grüner Text "Successfully installed websocket-client-X.X.X"

Schritt 2: Grundlegendes WebSocket-Skript erstellen

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

#!/usr/bin/env python3
"""
OKX WebSocket Grundlagen - Echtzeit-Kursdaten empfangen
Dieses Skript verbindet sich mit OKX und empfängt BTC/USDT-Kurse
"""

import json
import time
from websocket import create_connection, WebSocketException


OKX WebSocket-Endpunkt (öffentliche Daten, kein API-Key nötig!)

OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"

def verbinde_mit_okx():
    """Stellt eine Verbindung zum OKX WebSocket her"""
    try:
        print("🔌 Verbinde mit OKX WebSocket...")
        ws = create_connection(OKX_WS_URL, timeout=10)
        print("✅ Verbindung erfolgreich hergestellt!")
        return ws
    except WebSocketException as e:
        print(f"❌ Verbindungsfehler: {e}")
        return None

def abonniere_kursdaten(ws):
    """Abonniert BTC/USDT-Kursdaten"""
    # Formatierte Subscribe-Nachricht für OKX
    nachricht = {
        "op": "subscribe",
        "args": [{
            "channel": "tickers",      # Kursdaten-Kanal
            "instId": "BTC-USDT"       # Handelspaar
        }]
    }
    
    # Nachricht senden
    ws.send(json.dumps(nachricht))
    print("📡 Abonnement für BTC/USDT gesendet...")

def empfange_daten(ws, dauer=30):
    """Empfängt Daten für eine bestimmte Zeit (in Sekunden)"""
    print(f"⏳ Empfange Daten für {dauer} Sekunden...\n")
    
    startzeit = time.time()
    while time.time() - startzeit < dauer:
        try:
            # Daten empfangen (mit kurzem Timeout)
            ws.settimeout(1)
            nachricht = ws.recv()
            
            if nachricht:
                daten = json.loads(nachricht)
                verarbeite_daten(daten)
                
        except Exception as e:
            continue
    
    print("\n⏹️ Datenempfang beendet.")

def verarbeite_daten(daten):
    """Verarbeitet und zeigt die empfangenen Daten an"""
    # Prüfe ob es Kursdaten sind
    if "data" in daten:
        for eintrag in daten["data"]:
            symbol = eintrag.get("instId", "N/A")
            preis = eintrag.get("last", "N/A")
            high_24h = eintrag.get("high24h", "N/A")
            low_24h = eintrag.get("low24h", "N/A")
            volumen = eintrag.get("vol24h", "N/A")
            
            print(f"📊 {symbol}")
            print(f"   Preis: ${preis}")
            print(f"   24h Hoch: ${high_24h} | 24h Tief: ${low_24h}")
            print(f"   Volumen: {volumen} BTC")
            print("-" * 50)

def main():
    """Hauptfunktion"""
    print("=" * 60)
    print("🚀 OKX WebSocket Tutorial - Grundlagen")
    print("=" * 60)
    
    # Verbindung herstellen
    ws = verbinde_mit_okx()
    
    if ws:
        try:
            # Kursdaten abonnieren
            abonniere_kursdaten(ws)
            
            # Daten empfangen (30 Sekunden)
            empfange_daten(ws, dauer=30)
            
        finally:
            ws.close()
            print("🔒 Verbindung geschlossen.")

if __name__ == "__main__":
    main()

📸 Screenshot-Hinweis: Erstellen Sie die Datei in VS Code: Rechtsklick im Explorer → "Neue Datei" → Namen eingeben → Code einfügen

Schritt 3: Das Skript ausführen

Führen Sie das Skript aus, indem Sie im Terminal eingeben:

python grundlagen.py

Erwartete Ausgabe:

============================================================
🚀 OKX WebSocket Tutorial - Grundlagen
============================================================
🔌 Verbinde mit OKX WebSocket...
✅ Verbindung erfolgreich hergestellt!
📡 Abonnement für BTC/USDT gesendet...
⏳ Empfange Daten für 30 Sekunden...

📊 BTC-USDT
   Preis: 43251.50
   24h Hoch: 44120.00 | 24h Tief: 42850.00
   Volumen: 23456.78 BTC
--------------------------------------------------
📊 BTC-USDT
   Preis: 43255.80
   24h Hoch: 44120.00 | 24h Tief: 42850.00
   Volumen: 23478.12 BTC
--------------------------------------------------
⏹️ Datenempfang beendet.
🔒 Verbindung geschlossen.

Schritt 4: Mehrere Handelspaare abonnieren

Möchten Sie mehrere Kryptowährungen gleichzeitig verfolgen? Kein Problem! Erweitern Sie Ihr Skript:

#!/usr/bin/env python3
"""
OKX WebSocket - Mehrere Handelspaare gleichzeitig
"""

import json
import time
from websocket import create_connection

OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"


Liste der Handelspaare, die Sie abonnieren möchten

HANDELSPAARE = [
    "BTC-USDT",
    "ETH-USDT",
    "SOL-USDT",
    "DOGE-USDT"
]

def abonniere_alle_paare(ws, paare):
    """Abonniert mehrere Handelspaare gleichzeitig"""
    nachricht = {
        "op": "subscribe",
        "args": [
            {"channel": "tickers", "instId": paar} 
            for paar in paare
        ]
    }
    
    ws.send(json.dumps(nachricht))
    print(f"📡 {len(paare)} Handelspaare abonniert: {', '.join(paare)}")

def main():
    print("=" * 60)
    print("🚀 OKX WebSocket - Multi-Paar Monitoring")
    print("=" * 60)
    
    ws = create_connection(OKX_WS_URL, timeout=10)
    print("✅ Verbindung hergestellt!\n")
    
    try:
        abonniere_alle_paare(ws, HANDELSPAARE)
        
        # 60 Sekunden lang Daten empfangen
        empfangszeit = 60
        print(f"\n⏳ Empfange Daten für {empfangszeit} Sekunden...\n")
        
        startzeit = time.time()
        while time.time() - startzeit < empfangszeit:
            try:
                ws.settimeout(1)
                nachricht = ws.recv()
                
                if nachricht:
                    daten = json.loads(nachricht)
                    
                    # Daten formatieren und anzeigen
                    if "data" in daten:
                        for eintrag in daten["data"]:
                            preis = float(eintrag.get("last", 0))
                            symbol = eintrag.get("instId", "N/A")
                            
                            # Emoji basierend auf Symbol
                            emojis = {
                                "BTC-USDT": "₿",
                                "ETH-USDT": "Ξ",
                                "SOL-USDT": "◎",
                                "DOGE-USDT": "Ð"
                            }
                            emoji = emojis.get(symbol, "💰")
                            
                            print(f"{emoji} {symbol}: ${preis:,.2f}")
                            
            except:
                continue
        
        print("\n⏹️ Beendet.")
        
    finally:
        ws.close()

if __name__ == "__main__":
    main()

Schritt 5: Daten in eine Datei speichern

Für spätere Analysen ist es sinnvoll, die Daten zu speichern:

#!/usr/bin/env python3
"""
OKX WebSocket - Daten in CSV-Datei speichern
"""

import json
import csv
import time
from datetime import datetime
from websocket import create_connection

OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"


CSV-Datei vorbereiten

CSV_DATEI = "kursdaten.csv"

def initialisiere_csv():
    """Erstellt die CSV-Datei mit Kopfzeile"""
    with open(CSV_DATEI, mode='w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            "Zeitstempel",
            "Symbol",
            "Preis",
            "24h-Hoch",
            "24h-Tief",
            "Volumen"
        ])
    print(f"📁 CSV-Datei erstellt: {CSV_DATEI}")

def speichere_in_csv(daten):
    """Speichert einen Datensatz in die CSV-Datei"""
    if "data" in daten:
        with open(CSV_DATEI, mode='a', newline='') as f:
            writer = csv.writer(f)
            
            for eintrag in daten["data"]:
                zeile = [
                    datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    eintrag.get("instId", ""),
                    eintrag.get("last", ""),
                    eintrag.get("high24h", ""),
                    eintrag.get("low24h", ""),
                    eintrag.get("vol24h", "")
                ]
                writer.writerow(zeile)

def main():
    print("=" * 60)
    print("🚀 OKX WebSocket - Datenlogger")
    print("=" * 60)
    
    initialisiere_csv()
    
    ws = create_connection(OKX_WS_URL, timeout=10)
    print("✅ Verbindung hergestellt!\n")
    
    # BTC und ETH abonnieren
    nachricht = {
        "op": "subscribe",
        "args": [
            {"channel": "tickers", "instId": "BTC-USDT"},
            {"channel": "tickers", "instId": "ETH-USDT"}
        ]
    }
    ws.send(json.dumps(nachricht))
    print("📡 BTC und ETH abonniert...")
    
    daten_count = 0
    startzeit = time.time()
    
    try:
        while time.time() - startzeit < 120:  # 2 Minuten
            try:
                ws.settimeout(1)
                nachricht = ws.recv()
                
                if nachricht:
                    daten = json.loads(nachricht)
                    speichere_in_csv(daten)
                    daten_count += 1
                    
                    if daten_count % 10 == 0:
                        print(f"📝 {daten_count} Datensätze gespeichert...")
                        
            except:
                continue
                
    finally:
        ws.close()
        print(f"\n✅ Fertig! {daten_count} Datensätze in {CSV_DATEI} gespeichert.")

if __name__ == "__main__":
    main()

Praxisbeispiel: Automatischer Trading-Alert

Ein praktisches Beispiel: Erstellen Sie einen Alarm, der Sie benachrichtigt, wenn der Bitcoin-Preis einen bestimmten Wert erreicht:

#!/usr/bin/env python3
"""
OKX WebSocket - Trading-Alarm System
"""

import json
import time
from websocket import create_connection

OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"


Ihre Alarm-Schwellenwerte

ALARM_THRESHOLD = {
    "BTC-USDT": {"max_preis": 50000, "min_preis": 40000},
    "ETH-USDT": {"max_preis": 3000, "min_preis": 2500}
}

def checke_alarm(symbol, preis):
    """Prüft ob ein Alarm ausgelöst werden soll"""
    if symbol not in ALARM_THRESHOLD:
        return None
    
    threshold = ALARM_THRESHOLD[symbol]
    
    if preis >= threshold["max_preis"]:
        return f"🔔 ALARM: {symbol} hat ${preis} erreicht (über ${threshold['max_preis']}!)"
    
    if preis <= threshold["min_preis"]:
        return f"🔔 ALARM: {symbol} ist auf ${preis} gefallen (unter ${threshold['min_preis']}!)"
    
    return None

def main():
    print("=" * 60)
    print("🚀 OKX WebSocket - Trading Alarm System")
    print("=" * 60)
    
    print("\n📊 Überwachte Paare:")
    for symbol, data in ALARM_THRESHOLD.items():
        print(f"   {symbol}: ${data['min_preis']} - ${data['max_preis']}")
    print()
    
    ws = create_connection(OKX_WS_URL, timeout=10)
    print("✅ Verbindung hergestellt!")
    
    nachricht = {
        "op": "subscribe",
        "args": [
            {"channel": "tickers", "instId": symbol}
            for symbol in ALARM_THRESHOLD.keys()
        ]
    }
    ws.send(json.dumps(nachricht))
    print("📡 Überwachung aktiv...\n")
    
    try:
        while True:
            try:
                ws.settimeout(1)
                daten = json.loads(ws.recv())
                
                if "data" in daten:
                    for eintrag in daten["data"]:
                        symbol = eintrag.get("instId", "")
                        preis = float(eintrag.get("last", 0))
                        
                        alarm = checke_alarm(symbol, preis)
                        if alarm:
                            print(f"\n{'='*50}")
                            print(f"⚠️  {alarm}")
                            print(f"   Zeit: {time.strftime('%H:%M:%S')}")
                            print(f"{'='*50}\n")
                            
            except:
                continue
                
    except KeyboardInterrupt:
        print("\n⏹️ Überwachung beendet.")
    finally:
        ws.close()

if __name__ == "__main__":
    main()

Geeignet / Nicht geeignet für

✅ Geeignet für ❌ Nicht geeignet für
  • Hobby-Trader und Einsteiger
  • Portfolio-Tracking-Apps
  • Trading-Bots mit geringem Volumen
  • Education und Lernzwecke
  • Preisüberwachung und Alerts
  • Hochfrequenz-Trading (HFT)
  • Professionelle Trading-Systeme
  • Millisekunden-genaue Orderausführung
  • Compliance-pflichtige Anwendungen
  • Trading mit >100 Positionen gleichzeitig

Preise und ROI

Anbieter Monatliche Kosten Latenz Kosten pro 1M Token Ideal für
OKX WebSocket (öffentlich) Kostenlos ~100-200ms N/A Grundlegende Daten
OKX WebSocket (API-Key) Ab $0 ~80-150ms N/A Private Daten
HolySheep AI Ab $0 (kostenlose Credits) <50ms GPT-4.1: $8
Claude Sonnet 4.5: $15
DeepSeek V3.2: $0.42		 KI-Analyse + Daten
OpenAI API Ab $0 + Nutzung ~200-500ms $2-15 Allgemeine KI

💰 ROI-Analyse: Mit HolySheep AI sparen Sie gegenüber OpenAI 85-94% bei vergleichbarer Qualität. DeepSeek V3.2 kostet nur $0.42 pro Million Token – ideal für die Analyse Ihrer OKX-Marktdaten.

Warum HolySheep wählen

Die OKX WebSocket-Verbindung liefert Ihnen Echtzeit-Kurse, aber wie analysieren Sie diese Daten intelligent? Hier kommt HolySheep AI ins Spiel:

#!/usr/bin/env python3
"""
Beispiel: OKX-Daten mit HolySheep AI analysieren
"""

import requests
import json


HolySheep AI Konfiguration

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Ersetzen Sie mit Ihrem Key

def analysiere_markttrend(marktdaten_text):
    """
    Analysiert Marktdaten mit KI
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-v3.2",  # Günstigste Option: $0.42/MTok
        "messages": [
            {
                "role": "system",
                "content": "Du bist ein erfahrener Krypto-Analyst. Analysiere Marktdaten und gebe Handlungsempfehlungen."
            },
            {
                "role": "user", 
                "content": f"Analysiere folgende Marktdaten:\n{marktdaten_text}"
            }
        ],
        "temperature": 0.7,
        "max_tokens": 500
    }
    
    antwort = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload
    )
    
    if antwort.status_code == 200:
        ergebnis = antwort.json()
        return ergebnis["choices"][0]["message"]["content"]
    else:
        return f"Fehler: {antwort.status_code}"


Beispiel-Nutzung

if __name__ == "__main__":
    beispiel_daten = """
    BTC-USDT: $43.251,50 (24h: +2.3%)
    ETH-USDT: $2.651,80 (24h: +1.8%)
    SOL-USDT: $98,45 (24h: +5.2%)
    DOGE-USDT: $0,0823 (24h: -0.5%)
    """
    
    analyse = analysiere_markttrend(beispiel_daten)
    print("📊 KI-Analyse:")
    print(analyse)

Häufige Fehler und Lösungen

1. Verbindungsfehler: "Connection refused"

Ursache: Falsche WebSocket-URL oder Firewall blockiert die Verbindung.

# ❌ FALSCH - Häufige Fehler
OKX_WS_URL = "wss://ws.okx.com/ws/v5/public"  # Fehlender Port!
OKX_WS_URL = "https://ws.okx.com:8443/ws/v5/public"  # https statt wss!


✅ RICHTIG

OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"  # Öffentliche Daten
OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/private"  # Private Daten (API-Key nötig)

2. Subscribe-Fehler: "Illegal argument"

Ursache: Falsches Format der Subscribe-Nachricht.

# ❌ FALSCH - Häufige Fehler
nachricht = {"op": "subscribe", "channel": "tickers", "instId": "BTC-USDT"}


✅ RICHTIG - OKX erwartet "args" als Array

nachricht = {
    "op": "subscribe",
    "args": [
        {"channel": "tickers", "instId": "BTC-USDT"}
    ]
}


Bei mehreren Paaren:

nachricht = {
    "op": "subscribe",
    "args": [
        {"channel": "tickers", "instId": "BTC-USDT"},
        {"channel": "tickers", "instId": "ETH-USDT"}
    ]
}

3. Timeout beim Empfang von Daten

Ursache: Blockierender Empfangsaufruf ohne Timeout-Einstellung.

# ❌ FALSCH - Hängt unbegrenzt
while True:
    nachricht = ws.recv()  # Kann ewig warten!


✅ RICHTIG - Mit Timeout

import socket

ws.settimeout(5)  # 5 Sekunden Timeout
while True:
    try:
        nachricht = ws.recv()
        daten = json.loads(nachricht)
        verarbeite_daten(daten)
    except socket.timeout:
        print("⏰ Timeout - prüfe Verbindung...")
        # Hier Reconnection-Logik einfügen
        break
    except Exception as e:
        print(f"Fehler: {e}")

4. Doppelte oder fehlende Nachrichten

Ursache: Erneutes Senden des Subscribe-Befehls oder Verbindungsprobleme.

# ✅ KOMPLETTE Fehlerbehandlung mit Reconnection
import json
import time
from websocket import create_connection, WebSocketException

class OKXWebSocketClient:
    def __init__(self):
        self.ws = None
        self.max_retries = 5
        self.retry_delay = 3
        
    def verbinde(self):
        """Verbindet sich mit automatischer Wiederholung"""
        for versuch in range(self.max_retries):
            try:
                self.ws = create_connection(
                    "wss://ws.okx.com:8443/ws/v5/public",
                    timeout=10
                )
                print("✅ Verbindung hergestellt")
                return True
            except WebSocketException as e:
                print(f"❌ Versuch {versuch+1}/{self.max_retries} fehlgeschlagen")
                time.sleep(self.retry_delay)
        return False
    
    def subscribe(self, channel, instId):
        """Sendet Subscribe-Nachricht sicher"""
        nachricht = {
            "op": "subscribe",
            "args": [{"channel": channel, "instId": instId}]
        }
        self.ws.send(json.dumps(nachricht))
        print(f"📡 Abonniert: {channel} - {instId}")
    
    def empfange_mit_handling(self, dauer=60):
        """Empfängt Daten mit vollständiger Fehlerbehandlung"""
        self.ws.settimeout(1)
        start = time.time()
        
        while time.time() - start < dauer:
            try:
                nachricht = self.ws.recv()
                daten = json.loads(nachricht)
                
                # Heartbeat-Nachrichten ignorieren
                if "event" in daten:
                    continue
                    
                return daten
                
            except Exception as e:
                continue
        
        return None


Nutzung:

if __name__ == "__main__":
    client = OKXWebSocketClient()
    if client.verbinde():
        client.subscribe("tickers", "BTC-USDT")
        daten = client.empfange_mit_handling()

Zusammenfassung und nächste Schritte

In diesem Tutorial haben Sie gelernt:

Nächste Schritte zum Vertiefen:

  1. Private Daten: Erstellen Sie einen OKX API-Key für den Zugriff auf Ihre Kontodaten
  2. Order-Platzierung: Nutzen Sie den trade-Kanal für automatisierte Orders
  3. KI-Integration: Verbinden Sie OKX-Daten mit HolySheep AI für intelligente Analyse
  4. Multi-Exchange: Erweitern Sie Ihr System um Binance, Coinbase und andere Börsen

Kaufempfehlung

Für Anfänger und Hobby-Trader ist die OKX WebSocket-Verbindung völlig kostenlos und ideal zum Lernen. Wenn Sie jedoch KI-gestützte Marktanalyse benötigen, ist HolySheep AI die beste Wahl:

🎯 Meine Empfehlung
Nur Marktdaten: OKX WebSocket (kostenlos, wie in diesem Tutorial)
Marktdaten + KI-Analyse: HolySheep AI mit DeepSeek V3.2 ($0.42/MTok)
Premium KI: HolySheep mit GPT-4.1 oder Claude Sonnet 4.5

Die Kombination aus OKX WebSocket für Echtzeit-Daten und HolySheep AI für die Analyse bietet das beste Preis-Leistungs-Verhältnis am Markt.

👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive

Viel Erfolg beim Programmieren! 🚀

Verwandte Ressourcen

Verwandte Artikel