OKX 交易所 WebSocket 实时行情 API 接入：从零开始手把手教学

作为一名在量化交易领域摸爬滚打 5 年的工程师，我第一次接触 WebSocket 时也是一头雾水——服务器推送是什么？双向通信怎么实现？为什么我的行情数据总是慢半拍？这些问题困扰了我很久。今天我要用最通俗的语言，带你从零掌握 OKX WebSocket 实时行情的接入方法，保证看完就能跑通代码。

一、WebSocket 是什么？为什么获取行情必须用它？

想象你给餐厅打电话订餐：传统 HTTP 就像你每隔 5 秒打一次电话问「菜做好了吗」，而 WebSocket 就像服务员做好菜后主动给你送过来。WebSocket 建立连接后，服务器可以随时主动把数据「推」给你，不需要你一直去问。

在加密货币交易场景中，价格每秒可能变动几十次，用传统方式询问会：

延迟高：每次询问都有网络往返时间
浪费资源：大量重复的请求占用带宽
容易超时：频繁请求可能被服务器拒绝

所以，获取实时行情 WebSocket 是必选项。

二、准备工作：注册 OKX 账号并创建 API Key

第一步：注册 OKX 账号

访问 OKX 官网，使用手机号或邮箱完成注册。国内用户建议用手机号，验证更便捷。

（文字提示：截图位置 — 注册页面，填写手机号和验证码的区域）

第二步：创建 API Key

登录后按以下路径操作：我的面板 → API → 创建 API Key

（文字提示：截图位置 — 顶部导航栏「我的面板」下拉菜单）

创建时注意勾选以下权限：

读取（Read）：获取行情数据必选
现货交易（Spot Trade）：如需现货交易
合约交易（Derivatives Trade）：如需合约交易

（文字提示：截图位置 — API 权限设置页面，3 个可勾选复选框）

第三步：保存关键信息

创建成功后页面会显示 API Key 和 Secret Key，页面关闭后无法再查看 Secret Key，请立即复制保存到本地 txt 文件。建议格式如下：

API Key: a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6
Secret Key: A1B2C3D4E5F6G7H8I9J0K1L2M3N4O5P6Q7R8S9T0
Passphrase: your_passphrase_here

三、Python 环境准备

安装 Python（如果还没安装）

前往 Python 官网下载最新版本。安装时务必勾选「Add Python to PATH」。

（文字提示：截图位置 — Python 安装界面，底部「Add Python to PATH」复选框）

安装行情库

打开命令行（Windows 按 Win+R，输入 cmd），依次执行：

pip install websocket-client requests

如果遇到权限报错，改用：

pip install websocket-client requests --user

四、连接 OKX WebSocket 获取实时行情

1. 获取实时成交数据

import websocket
import json
import pandas as pd
import threading

全局变量存储实时数据
latest_trades = []

def on_message(ws, message):
    """收到消息时的回调函数"""
    data = json.loads(message)
    # 成交数据的 event 为 trade
    if data.get('data'):
        for trade in data['data']:
            latest_trades.append({
                'inst_id': trade['instId'],
                'price': float(trade['px']),
                'size': float(trade['sz']),
                'side': trade['side'],  # buy 或 sell
                'timestamp': trade['ts']
            })
            print(f"【成交】{trade['instId']} | 价格: {trade['px']} | 数量: {trade['sz']} | 方向: {trade['side']}")

def on_error(ws, error):
    """错误处理"""
    print(f"WebSocket 错误: {error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭时的回调"""
    print(f"连接已关闭，状态码: {close_status_code}")

def on_open(ws):
    """连接建立时的回调 - 订阅频道"""
    # 订阅 BTC-USDT 的成交数据
    subscribe_msg = {
        "op": "subscribe",
        "args": [
            {
                "channel": "trades",
                "instId": "BTC-USDT"
            }
        ]
    }
    ws.send(json.dumps(subscribe_msg))
    print("已订阅 BTC-USDT 成交数据")

建立连接
ws = websocket.WebSocketApp(
    "wss://ws.okx.com:8443/ws/v5/public",
    on_message=on_message,
    on_error=on_error,
    on_close=on_close,
    on_open=on_open
)

启动连接（非阻塞方式）
ws_thread = threading.Thread(target=ws.run_forever)
ws_thread.start()

运行 30 秒后关闭
import time
time.sleep(30)
ws.close()
print(f"采集完成，共获取 {len(latest_trades)} 条成交记录")

2. 获取实时 K 线数据（1分钟）

import websocket
import json
import time

def on_message(ws, message):
    """处理接收到的 K 线数据"""
    data = json.loads(message)
    if data.get('data'):
        for candle in data['data']:
            print(f"""
【K线更新】
品种: {candle['instId']}
时间: {candle['ts']}
开盘: {candle['open']}
最高: {candle['high']}
最低: {candle['low']}
收盘: {candle['close']}
成交量: {candle['vol']}
""")

def on_open(ws):
    """订阅 1 分钟 K 线"""
    subscribe_msg = {
        "op": "subscribe",
        "args": [
            {
                "channel": "candle1m",
                "instId": "BTC-USDT"
            }
        ]
    }
    ws.send(json.dumps(subscribe_msg))
    print("已订阅 BTC-USDT 1分钟K线")

简化的连接方式
ws = websocket.WebSocketApp(
    "wss://ws.okx.com:8443/ws/v5/public",
    on_message=on_message,
    on_open=on_open
)

print("开始接收 K 线数据（5秒后自动关闭）...")
ws.run_forever(ping_timeout=20)
time.sleep(5)
ws.close()

3. 获取实时深度数据（订单簿）

import websocket
import json

def on_message(ws, message):
    """处理深度数据"""
    data = json.loads(message)
    if data.get('data'):
        depth = data['data'][0]
        asks = depth.get('asks', [])[:5]  # 卖盘前5档
        bids = depth.get('bids', [])[:5]  # 买盘前5档
        
        print("【订单簿深度】")
        print("-" * 40)
        print("卖盘（Asks）:")
        for price, size in asks:
            print(f"  价格: {price:>12} | 数量: {size}")
        print("-" * 40)
        print("买盘（Bids）:")
        for price, size in bids:
            print(f"  价格: {price:>12} | 数量: {size}")
        print("-" * 40)

def on_open(ws):
    """订阅深度数据（5档）"""
    subscribe_msg = {
        "op": "subscribe",
        "args": [
            {
                "channel": "books5",  # books5 = 5档深度
                "instId": "BTC-USDT"
            }
        ]
    }
    ws.send(json.dumps(subscribe_msg))

ws = websocket.WebSocketApp(
    "wss://ws.okx.com:8443/ws/v5/public",
    on_message=on_message,
    on_open=on_open
)

ws.run_forever(ping_timeout=20)

运行结果预览

执行上面任意一个脚本，你应该能看到类似这样的输出：

已订阅 BTC-USDT 成交数据
【成交】BTC-USDT | 价格: 67842.50 | 数量: 0.0152 | 方向: buy
【成交】BTC-USDT | 价格: 67842.30 | 数量: 0.0200 | 方向: sell
【成交】BTC-USDT | 价格: 67843.00 | 数量: 0.0100 | 方向: buy
...

（文字提示：截图位置 — 终端运行窗口，显示实时成交数据滚动）

五、常见报错排查

错误一：ConnectionRefusedError 连接被拒绝

错误信息：

ConnectionRefusedError: [WinError 10061] 由于目标计算机积极拒绝，无法连接。

原因分析：

OKX 服务器地址或端口写错
国内网络无法直接访问 OKX 节点
防火墙或代理软件拦截了请求

解决方案：

# 方法1：检查地址是否正确
正确地址（现货/合约通用）
ws_url = "wss://ws.okx.com:8443/ws/v5/public"

方法2：测试网络连通性
在命令行执行：
ping ws.okx.com

方法3：配置代理（如果需要）
import os
os.environ["HTTP_PROXY"] = "http://127.0.0.1:7890"
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:7890"

方法4：使用国内可访问的节点（如果 OKX 提供）
某些行情服务商会提供镜像节点
ws_url = "wss://wspush.okxcn.com:8443/ws/v5/public"

错误二：IllegalArgument 订阅格式错误

错误信息：

{'event': 'error', 'msg': 'Illegal argument: unknown channel type', 'code': '60021'}

原因分析：

channel 名称拼写错误
instId 格式不对
交易对标识错误（现货 vs 合约）

解决方案：

# 常见 channel 类型对照表
channels = {
    "trades": "成交数据",
    "candle1m": "1分钟K线",
    "candle5m": "5分钟K线",
    "candle1H": "1小时K线",
    "books5": "5档深度",
    "books50": "50档深度",
    "tickers": "行情快照"
}

交易对格式对照表
inst_id_examples = {
    "BTC-USDT": "BTC/USDT 现货",
    "BTC-USDT-SWAP": "BTC/USDT 永续合约",
    "ETH-USDT-SWAP": "ETH/USDT 永续合约",
    "BTC-USD-230930": "BTC/USD 交割合约（到期日230930）"
}

正确订阅示例
subscribe_msg = {
    "op": "subscribe",
    "args": [
        {
            "channel": "trades",  # 不要写成 "trade" 或 "TRADES"
            "instId": "BTC-USDT"  # 不要写成 "BTC/USDT" 或 "BTC_USDT"
        }
    ]
}

错误三：心跳超时导致连接断开

错误信息：

ConnectionClosedOK: connection is closed
或长时间无数据输出

原因分析：

OKX WebSocket 超过 30 秒无响应会断开连接
网络不稳定导致心跳丢失
服务器端维护或限流

解决方案：

import websocket
import threading
import time

class OkxWebSocketClient:
    def __init__(self):
        self.ws = None
        self.running = False
        
    def connect(self):
        """建立连接并自动重连"""
        while self.running:
            try:
                self.ws = websocket.WebSocketApp(
                    "wss://ws.okx.com:8443/ws/v5/public",
                    on_message=self.on_message,
                    on_open=self.on_open,
                    on_close=self.on_close,
                    on_error=self.on_error
                )
                # ping_timeout=25 非常重要！确保自动处理心跳
                self.ws.run_forever(ping_timeout=25, ping_interval=20)
            except Exception as e:
                print(f"连接异常: {e}")
            
            # 断连后等待 5 秒重试
            if self.running:
                print("5 秒后重新连接...")
                time.sleep(5)
    
    def on_open(self, ws):
        """订阅操作"""
        subscribe_msg = {
            "op": "subscribe", 
            "args": [{"channel": "trades", "instId": "BTC-USDT"}]
        }
        ws.send(json.dumps(subscribe_msg))
        print("连接成功，已订阅")
    
    def on_message(self, ws, message):
        print(f"收到数据: {message[:100]}...")
    
    def start(self):
        """启动客户端"""
        self.running = True
        self.thread = threading.Thread(target=self.connect)
        self.thread.start()
    
    def stop(self):
        """停止客户端"""
        self.running = False
        if self.ws:
            self.ws.close()

使用示例
client = OkxWebSocketClient()
client.start()

运行 1 小时
time.sleep(3600)
client.stop()

错误四：数据解析异常 JSONDecodeError

错误信息：

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

原因分析：

收到了空消息或非 JSON 格式的响应
消息编码问题（某些节点返回 GBK 编码）
心跳响应不是 JSON 格式

解决方案：

def on_message(ws, message):
    """带容错的消息处理"""
    if not message:  # 忽略空消息
        return
    
    try:
        data = json.loads(message)
        # 处理订阅成功响应
        if data.get('event') == 'subscribe':
            print(f"订阅成功: {data.get('arg')}")
            return
        
        # 处理实际数据
        if data.get('data'):
            # 你的数据处理逻辑
            pass
            
    except json.JSONDecodeError:
        # 如果不是 JSON，可能是其他类型消息
        print(f"收到非JSON消息: {message[:50]}")
    except Exception as e:
        print(f"处理消息时出错: {e}")

六、数据处理实战：将行情存入 CSV

实时数据获取后，通常需要持久化存储用于后续分析。以下是完整的数据采集和存储方案：

import websocket
import json
import csv
from datetime import datetime
import threading
import time

class MarketDataCollector:
    def __init__(self, symbol="BTC-USDT", filename="market_data.csv"):
        self.symbol = symbol
        self.filename = filename
        self.data_buffer = []
        self.ws = None
        self.running = False
        self.setup_csv()
        
    def setup_csv(self):
        """初始化 CSV 文件"""
        with open(self.filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['timestamp', 'datetime', 'symbol', 'price', 'size', 'side'])
    
    def save_to_csv(self):
        """定期将缓冲区数据写入 CSV"""
        while self.running:
            time.sleep(10)  # 每 10 秒保存一次
            if self.data_buffer:
                with open(self.filename, 'a', newline='', encoding='utf-8') as f:
                    writer = csv.writer(f)
                    writer.writerows(self.data_buffer)
                print(f"已保存 {len(self.data_buffer)} 条数据到 {self.filename}")
                self.data_buffer.clear()
    
    def on_message(self, ws, message):
        try:
            data = json.loads(message)
            if data.get('data'):
                for trade in data['data']:
                    ts = int(trade['ts'])
                    dt = datetime.fromtimestamp(ts/1000).strftime('%Y-%m-%d %H:%M:%S')
                    self.data_buffer.append([
                        trade['ts'],
                        dt,
                        trade['instId'],
                        trade['px'],
                        trade['sz'],
                        trade['side']
                    ])
        except Exception as e:
            print(f"数据处理错误: {e}")
    
    def on_open(self, ws):
        msg = {"op": "subscribe", "args": [{"channel": "trades", "instId": self.symbol}]}
        ws.send(json.dumps(msg))
        print(f"已订阅 {self.symbol} 成交数据")
    
    def start(self):
        """启动采集"""
        self.running = True
        self.ws = websocket.WebSocketApp(
            "wss://ws.okx.com:8443/ws/v5/public",
            on_message=self.on_message,
            on_open=self.on_open
        )
        self.ws_thread = threading.Thread(target=self.ws.run_forever, kwargs={'ping_timeout': 25})
        self.save_thread = threading.Thread(target=self.save_to_csv)
        
        self.ws_thread.start()
        self.save_thread.start()
    
    def stop(self):
        """停止采集"""
        self.running = False
        self.ws.close()
        print("采集已停止")

使用示例：采集 30 分钟 BTC 成交数据
collector = MarketDataCollector("BTC-USDT", "btc_trades.csv")
collector.start()
print("开始采集数据（将运行 30 分钟）...")
time.sleep(1800)
collector.stop()

七、高阶用法：订阅多币种多周期

一个连接可以同时订阅多个频道，大幅提升效率：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if data.get('data'):
        candle = data['data'][0]
        print(f"[{candle['instId']}] {candle['ts']} | O:{candle['open']} H:{candle['high']} L:{candle['low']} C:{candle['close']}")

def on_open(ws):
    # 同时订阅多个币种的多个周期
    subscribe_msg = {
        "op": "subscribe",
        "args": [
            {"channel": "candle1m", "instId": "BTC-USDT"},
            {"channel": "candle1m", "instId": "ETH-USDT"},
            {"channel": "candle5m", "instId": "BTC-USDT"},
            {"channel": "candle1H", "instId": "ETH-USDT"},
            {"channel": "candle1H", "instId": "SOL-USDT"},
        ]
    }
    ws.send(json.dumps(subscribe_msg))
    print("已订阅 5 个频道")

ws = websocket.WebSocketApp(
    "wss://ws.okx.com:8443/ws/v5/public",
    on_message=on_message,
    on_open=on_open
)

ws.run_forever(ping_timeout=25)

八、性能优化建议

当数据量增大时（每秒数百条消息），需要注意以下优化：

1. 使用线程安全队列

from queue import Queue
from threading import Thread

data_queue = Queue(maxsize=10000)

def data_processor():
    """独立线程处理数据，避免阻塞接收"""
    while True:
        data = data_queue.get()
        # 写入数据库、计算指标等耗时操作
        process_data(data)

def on_message(ws, message):
    # 只做最小化处理，快速入队
    data_queue.put(message)

2. 批量写入数据库

# 使用 SQLAlchemy 批量插入
from sqlalchemy import create_engine, Table, Column, Float, String, Integer
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///trades.db', pool_size=10)
Session = sessionmaker(bind=engine)

批量写入（每 100 条或每 5 秒写入一次）
batch = []
while True:
    if len(batch) >= 100:
        session.bulk_insert_mappings(Trade, batch)
        session.commit()
        batch.clear()

九、下一步：用 AI 分析实时行情

获取到实时行情后，你可以结合大模型 API 进行智能分析，比如识别价格形态、生成交易信号、自动生成每日行情报告。

import requests

def analyze_market_with_ai(current_price, price_change_24h, volume_24h):
    """
    使用 HolySheep AI 分析当前市场状态
    HolySheep 汇率优势：¥1=$1无损，注册送免费额度
    官方地址：https://www.holysheep.ai
    """
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4.1",
            "messages": [
                {
                    "role": "system",
                    "content": "你是一个专业的加密货币分析师，用简洁的语言给出市场分析。"
                },
                {
                    "role": "user", 
                    "content": f"""分析以下 BTC 市场数据：
                    当前价格：${current_price}
                    24小时涨跌：{price_change_24h}%
                    24小时成交量：{volume_24h} BTC
                    请给出简短的走势判断和操作建议。"""
                }
            ],
            "temperature": 0.7,
            "max_tokens": 500
        }
    )
    
    if response.status_code == 200:
        result = response.json()
        return result['choices'][0]['message']['content']
    else:
        return f"API 请求失败: {response.status_code}"

使用示例
analysis = analyze_market_with_ai(
    current_price=67842.50,
    price_change_24h=2.35,
    volume_24h=15234.56
)
print("AI 分析结果：")
print(analysis)

推荐使用立即注册 HolySheep AI，原因如下：

汇率优势：¥1=$1 无损结算，对比官方 ¥7.3=$1，节省超过 85%
国内直连：延迟低于 50ms，无需科学上网
价格实惠：GPT-4.1 $8/MTok · Claude Sonnet 4.5 $15/MTok · Gemini 2.5 Flash $2.50/MTok · DeepSeek V3 $0.42/MTok
充值便捷：支持微信、支付宝直接充值
注册福利：新用户赠送免费调用额度

十、常见问题 FAQ

Q1：WebSocket 连接数有上限吗？

OKX 对每个账号限制最多 25 个 WebSocket 连接，每个连接最多订阅 100 个频道。建议使用单个连接订阅所有需要的频道。

Q2：数据是否有延迟？

OKX 官方标注延迟在 100ms 以内，实测国内节点通常在 20-50ms。如果延迟过高，检查网络或考虑使用代理。

Q3：能否获取历史 K 线数据？

WebSocket 只推送实时数据，历史数据需要调用 REST API 的 /api/v5/market/history-candles 接口。

Q4：连接断开后会自动重连吗？

默认不会自动重连。建议在代码中实现自动重连机制（参考本文第五节的代码示例）。

Q5：支持哪些交易产品？

WebSocket 支持现货、币币杠杆、永续合约、交割合约、期权等全品类行情。

总结

本文从零讲解了 OKX WebSocket 实时行情 API 的接入方法，涵盖：

WebSocket 基础概念和优势
OKX API Key 创建流程
Python 连接代码（成交、K线、深度三种数据类型）
4 种常见错误的排查和解决方案
数据持久化存储方案
多频道订阅和高阶用法
AI 行情分析的集成思路

完整代码可以直接复制运行，修改 instId 即可切换不同的交易对。入门后可以进一步研究 OKX REST API 实现下单交易、账户管理等功能，构建完整的量化交易系统。

如果你的量化策略需要 AI 辅助分析或需要更稳定的行情服务，推荐使用免费注册 HolySheep AI，享受 ¥1=$1 无损汇率和国内直连 50ms 以内的极速体验。

👉 免费注册 HolySheep AI，获取首月赠额度

一、WebSocket 是什么？为什么获取行情必须用它？

二、准备工作：注册 OKX 账号并创建 API Key

第一步：注册 OKX 账号

第二步：创建 API Key

第三步：保存关键信息

三、Python 环境准备

安装 Python（如果还没安装）

安装行情库

四、连接 OKX WebSocket 获取实时行情

1. 获取实时成交数据

全局变量存储实时数据

建立连接

启动连接（非阻塞方式）

运行 30 秒后关闭

2. 获取实时 K 线数据（1分钟）

简化的连接方式

3. 获取实时深度数据（订单簿）

运行结果预览

五、常见报错排查

错误一：ConnectionRefusedError 连接被拒绝

正确地址（现货/合约通用）

方法2：测试网络连通性

在命令行执行：

ping ws.okx.com

方法3：配置代理（如果需要）

方法4：使用国内可访问的节点（如果 OKX 提供）

某些行情服务商会提供镜像节点

错误二：IllegalArgument 订阅格式错误

交易对格式对照表

正确订阅示例

错误三：心跳超时导致连接断开

或长时间无数据输出

使用示例

运行 1 小时

错误四：数据解析异常 JSONDecodeError

六、数据处理实战：将行情存入 CSV

使用示例：采集 30 分钟 BTC 成交数据

七、高阶用法：订阅多币种多周期

八、性能优化建议

1. 使用线程安全队列

2. 批量写入数据库

批量写入（每 100 条或每 5 秒写入一次）

九、下一步：用 AI 分析实时行情

使用示例

十、常见问题 FAQ

Q1：WebSocket 连接数有上限吗？

Q2：数据是否有延迟？

Q3：能否获取历史 K 线数据？

Q4：连接断开后会自动重连吗？

Q5：支持哪些交易产品？

总结

相关资源

相关文章

🔥 推荐使用 HolySheep AI

`或长时间无数据输出`