API 教程(五) Websocket API使用介绍和注意事项

import { Callout } from '@/components/Callout'

为什么你的交易程序总是延迟几秒才收到行情数据？REST API每秒只能请求20次，但Websocket可以实时推送上千条数据。对于量化交易和高频策略来说，Websocket API是获取实时市场数据的最佳选择。

Websocket API基础概念

什么是Websocket API

Websocket是一种全双工通信协议，建立连接后服务器可以主动向客户端推送数据。

与REST API的对比：

| 特性 | REST API | Websocket API | |------|----------|---------------| | 通信方式 | 请求-响应 | 双向推送 | | 数据实时性 | 需要轮询 | 实时推送 | | 频率限制 | 20次/秒 | 无限制 | | 连接开销 | 每次请求建立连接 | 长连接复用 | | 适用场景 | 查询历史数据 | 实时行情/订单 |

OKX Websocket端点

公共频道（无需认证）：

正式环境：wss://ws.okx.com:8443/ws/v5/public
模拟环境：wss://wspap.okx.com:8443/ws/v5/public?brokerId=9999

私有频道（需要认证）：

正式环境：wss://ws.okx.com:8443/ws/v5/private
模拟环境：wss://wspap.okx.com:8443/ws/v5/private?brokerId=9999

业务频道（交易信号）：

正式环境：wss://ws.okx.com:8443/ws/v5/business

连接建立与认证

公共频道连接示例

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f”收到数据: {data}”)

def on_open(ws):
    # 订阅BTC-USDT现货行情
    subscribe_msg = {
        “op”: “subscribe”,
        “args”: [{
            “channel”: “tickers”,
            “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_open=on_open
)

ws.run_forever()

私有频道认证

私有频道需要API Key签名认证：

import hmac
import base64
import time

def generate_signature(timestamp, method, request_path, secret_key):
    message = timestamp + method + request_path
    mac = hmac.new(
        bytes(secret_key, encoding='utf8'),
        bytes(message, encoding='utf-8'),
        digestmod='sha256'
    )
    return base64.b64encode(mac.digest()).decode()

def on_open(ws):
    timestamp = str(int(time.time()))
    sign = generate_signature(
        timestamp, 
        'GET', 
        '/users/self/verify',
        'YOUR_SECRET_KEY'
    )
    
    # 发送登录请求
    login_msg = {
        “op”: “login”,
        “args”: [{
            “apiKey”: “YOUR_API_KEY”,
            “passphrase”: “YOUR_PASSPHRASE”,
            “timestamp”: timestamp,
            “sign”: sign
        }]
    }
    ws.send(json.dumps(login_msg))

认证参数说明：

| 参数 | 说明 | 示例 | |------|------|------| | apiKey | API密钥 | 93c8f8d7-xxx | | passphrase | API密码短语 | MyPass123 | | timestamp | Unix时间戳（秒） | 1678886400 | | sign | HMAC SHA256签名 | base64编码 |

订阅频道详解

公共频道类型

1. 行情频道（tickers）

实时推送最新成交价、24小时涨跌幅等数据：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “tickers”,
    “instId”: “BTC-USDT”
  }]
}

2. K线频道（candle）

支持1m、3m、5m、15m、30m、1H、2H、4H、6H、12H、1D、1W、1M等周期：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “candle1m”,
    “instId”: “ETH-USDT”
  }]
}

3. 深度频道（books）

提供5档、50档、400档深度数据：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “books5”,
    “instId”: “BTC-USDT”
  }]
}

4. 成交频道（trades）

实时推送每笔成交记录：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “trades”,
    “instId”: “BTC-USDT”
  }]
}

私有频道类型

1. 账户频道（account）

推送账户余额变化：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “account”,
    “ccy”: “USDT”
  }]
}

2. 持仓频道（positions）

推送持仓变化：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “positions”,
    “instType”: “SWAP”,
    “instId”: “BTC-USDT-SWAP”
  }]
}

3. 订单频道（orders）

推送订单状态变化：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “orders”,
    “instType”: “SPOT”,
    “instId”: “BTC-USDT”
  }]
}

批量订阅

一次订阅多个频道可以减少消息数量：

{
  “op”: “subscribe”,
  “args”: [
    {“channel”: “tickers”, “instId”: “BTC-USDT”},
    {“channel”: “tickers”, “instId”: “ETH-USDT”},
    {“channel”: “candle1m”, “instId”: “BTC-USDT”}
  ]
}

订阅限制：

单次最多订阅100个频道
单个连接最多订阅240个频道
超过限制需要建立多个连接

连接管理最佳实践

心跳机制

服务器每30秒发送一次ping，客户端必须在5秒内回复pong，否则连接会被断开：

def on_message(ws, message):
    if message == “ping”:
        ws.send(“pong”)
        print(“已回复心跳”)
    else:
        data = json.loads(message)
        # 处理业务数据

断线重连

网络波动可能导致连接断开，需要实现自动重连：

import time

def connect_with_retry(url, max_retries=5):
    retry_count = 0
    
    while retry_count < max_retries:
        try:
            ws = websocket.WebSocketApp(
                url,
                on_message=on_message,
                on_error=on_error,
                on_close=on_close,
                on_open=on_open
            )
            ws.run_forever()
        except Exception as e:
            retry_count += 1
            wait_time = min(2 ** retry_count, 60)  # 指数退避
            print(f”连接失败，{wait_time}秒后重试...”)
            time.sleep(wait_time)
    
    print(“达到最大重试次数，停止重连”)

订阅状态管理

重连后需要重新订阅之前的频道：

class WebsocketManager:
    def __init__(self):
        self.subscriptions = []  # 保存订阅列表
        
    def subscribe(self, channel, inst_id):
        sub = {“channel”: channel, “instId”: inst_id}
        self.subscriptions.append(sub)
        
        msg = {“op”: “subscribe”, “args”: [sub]}
        self.ws.send(json.dumps(msg))
    
    def resubscribe_all(self):
        “””重连后重新订阅”””
        if self.subscriptions:
            msg = {“op”: “subscribe”, “args”: self.subscriptions}
            self.ws.send(json.dumps(msg))
            print(f”已重新订阅{len(self.subscriptions)}个频道”)

常见问题与解决方案

问题1：连接频繁断开

原因：

未正确处理心跳
网络不稳定
订阅频道过多

解决方案：

# 1. 确保心跳响应
def on_message(ws, message):
    if message == “ping”:
        ws.send(“pong”)
        return

# 2. 使用连接池分散订阅
def create_connection_pool(channels, channels_per_conn=100):
    connections = []
    for i in range(0, len(channels), channels_per_conn):
        batch = channels[i:i+channels_per_conn]
        ws = create_websocket(batch)
        connections.append(ws)
    return connections

问题2：数据延迟或丢失

原因：

客户端处理速度慢
网络带宽不足
订阅了过多高频数据

解决方案：

import queue
import threading

# 使用队列异步处理
message_queue = queue.Queue()

def on_message(ws, message):
    message_queue.put(message)  # 快速入队

def process_messages():
    while True:
        message = message_queue.get()
        # 耗时的数据处理
        process_data(message)

# 启动处理线程
threading.Thread(target=process_messages, daemon=True).start()

问题3：认证失败

常见错误码：

| 错误码 | 说明 | 解决方法 | |--------|------|----------| | 60009 | 时间戳过期 | 检查系统时间是否准确 | | 50113 | 签名错误 | 检查签名算法和密钥 | | 50111 | API Key无效 | 确认API Key是否正确 |

调试技巧：

# 打印签名信息用于调试
print(f”Timestamp: {timestamp}”)
print(f”Message: {timestamp}GET/users/self/verify”)
print(f”Signature: {sign}”)

问题4：订阅无响应

检查订阅确认消息：

def on_message(ws, message):
    data = json.loads(message)
    
    if data.get(“event”) == “subscribe”:
        if data.get(“code”) == “0”:
            print(f”订阅成功: {data['arg']}”)
        else:
            print(f”订阅失败: {data['msg']}”)

实战案例：实时监控价格突破

构建一个监控BTC价格突破的程序：

class PriceMonitor:
    def __init__(self, threshold_price):
        self.threshold = threshold_price
        self.triggered = False
        
    def on_message(self, ws, message):
        if message == “ping”:
            ws.send(“pong”)
            return
            
        data = json.loads(message)
        
        if data.get(“arg”, {}).get(“channel”) == “tickers”:
            ticker = data[“data”][0]
            current_price = float(ticker[“last”])
            
            if current_price > self.threshold and not self.triggered:
                self.triggered = True
                print(f”⚠️ 价格突破警报！当前价格: ${current_price}”)
                # 发送通知或执行交易
                self.send_alert(current_price)
    
    def send_alert(self, price):
        # 实现通知逻辑（邮件、短信、Telegram等）
        pass

# 使用示例
monitor = PriceMonitor(threshold_price=50000)
ws = websocket.WebSocketApp(
    “wss://ws.okx.com:8443/ws/v5/public”,
    on_message=monitor.on_message,
    on_open=lambda ws: ws.send(json.dumps({
        “op”: “subscribe”,
        “args”: [{“channel”: “tickers”, “instId”: “BTC-USDT”}]
    }))
)
ws.run_forever()

性能优化建议

1. 选择合适的深度档位

# 高频策略：使用5档深度（数据量小）
{“channel”: “books5”, “instId”: “BTC-USDT”}

# 做市策略：使用50档深度（平衡性能）
{“channel”: “books50”, “instId”: “BTC-USDT”}

# 深度分析：使用400档深度（完整数据）
{“channel”: “books”, “instId”: “BTC-USDT”}

2. 使用增量推送

深度数据支持增量更新，减少带宽消耗：

{
  “op”: “subscribe”,
  “args”: [{
    “channel”: “books”,
    “instId”: “BTC-USDT”,
    “snapshot”: false
  }]
}

3. 合理设置K线周期

# 避免订阅过多小周期K线
# ❌ 不推荐：同时订阅1m、3m、5m
# ✅ 推荐：只订阅1m，本地计算其他周期

总结

Websocket API是构建实时交易系统的核心技术。掌握连接管理、订阅机制和异常处理后，你可以构建稳定高效的量化交易程序。

关键要点：

公共频道无需认证，私有频道需要签名
必须正确处理心跳和断线重连
单连接最多订阅240个频道
使用异步队列处理高频数据
重连后需要重新订阅频道

下一步行动：

在模拟环境测试连接和订阅
实现完整的异常处理机制
监控连接状态和数据延迟
逐步迁移到生产环境

想深入了解API交易？查看API教程(一) V5 API总览和API教程(四) 获取个人数据。

免责声明

本文章可能包含不适用于您所在地区的产品相关内容。本文仅致力于提供一般性信息，不对其中的任何事实错误或遗漏负责任。本文仅代表作者个人观点，不代表欧易的观点。本文无意提供以下任何建议，包括但不限于：(i) 投资建议或投资推荐；(ii) 购买、出售或持有数字资产的要约或招揽；或 (iii) 财务、会计、法律或税务建议。持有的数字资产 (包括稳定币) 涉及高风险，可能会大幅波动，甚至变得毫无价值。您应根据自己的财务状况仔细考虑交易或持有数字资产是否适合您。有关您具体情况的问题，请咨询您的法律/税务/投资专业人士。本文中出现的信息 (包括市场数据和统计信息，如果有) 仅供一般参考之用。尽管我们在准备这些数据和图表时已采取了所有合理的谨慎措施，但对于此处表达的任何事实错误或遗漏，我们不承担任何责任。 © 2025 OKX。本文可以全文复制或分发，也可以使用本文 100 字或更少的摘录，前提是此类使用是非商业性的。整篇文章的任何复制或分发亦必须突出说明：”本文版权所有 © 2025 OKX，经许可使用。”允许的摘录必须引用文章名称并包含出处，例如”文章名称，[作者姓名 (如适用)]，© 2025 OKX”。部分内容可能由人工智能（AI）工具生成或辅助生成。不允许对本文进行衍生作品或其他用途。

展开