OKX WebSocket私有频道：python-okx库订单更新推送

江焘钦

1253人浏览 · 2025-10-06 04:36:54

江焘钦 · 2025-10-06 04:36:54 发布

OKX WebSocket私有频道：python-okx库订单更新推送

【免费下载链接】python-okx 项目地址: https://gitcode.com/GitHub_Trending/py/python-okx

你是否还在为无法实时获取订单状态而烦恼？是否希望第一时间掌握交易动态？本文将带你快速上手python-okx库的WebSocket私有频道功能，轻松实现订单更新的实时推送，让你的交易更加高效、精准。读完本文，你将能够：理解OKX WebSocket私有频道的基本概念、掌握python-okx库中相关类的使用方法、实现订单更新推送的订阅与处理、了解常见问题及解决方法。

1. OKX WebSocket私有频道简介

OKX WebSocket（套接字）服务提供了私有频道和公共频道两种类型，私有频道主要用于推送用户的账户相关数据、订单状态、资金变动等信息。订单更新推送是私有频道中的重要功能，它能在订单状态发生变化时（如新建、成交、取消等）实时将信息推送给用户，帮助用户及时掌握交易动态，做出快速反应。

在python-okx库中，负责处理WebSocket私有频道相关操作的是okx/websocket/WsPrivateAsync.py文件中的WsPrivateAsync类。该类提供了连接、订阅、取消订阅、接收消息等一系列方法，方便开发者快速集成WebSocket私有频道功能。

2. WsPrivateAsync类核心功能解析

2.1 类初始化

WsPrivateAsync类的初始化需要传入apiKey、passphrase、secretKey、url和useServerTime等参数。其中，apiKey、passphrase和secretKey是用户在OKX平台申请的API密钥信息，用于身份验证；url是WebSocket私有频道的连接地址；useServerTime用于指定是否使用服务器时间进行签名计算。

ws = WsPrivateAsync(
    apiKey="your apiKey",
    passphrase="your passphrase",
    secretKey="your secretKey",
    url=url,
    useServerTime=False
)

2.2 连接WebSocket

通过调用start()方法可以建立与OKX WebSocket私有频道的连接。该方法内部会先调用connect()方法创建WebSocket连接，然后创建一个任务来消费接收到的消息。

await ws.start()

2.3 订阅订单更新频道

要接收订单更新推送，需要订阅相应的频道。使用subscribe()方法，传入订阅参数和消息回调函数。订阅参数中需要指定channel为"orders"，并可通过instType指定交易品种类型（如"ANY"表示所有品种）。

args = [{"channel": "orders", "instType": "ANY"}]
await ws.subscribe(args, callback=privateCallback)

2.4 消息处理

当有订单更新消息推送时，会触发设置的回调函数。在回调函数中，可以对接收到的消息进行解析和处理，如提取订单ID、订单状态、成交数量等关键信息。

def privateCallback(message):
    print("privateCallback", message)
    # 解析消息内容，处理订单更新信息
    data = json.loads(message)
    if data.get("event") == "update" and data.get("arg", {}).get("channel") == "orders":
        order_info = data.get("data", [{}])[0]
        order_id = order_info.get("ordId")
        order_status = order_info.get("state")
        print(f"订单ID: {order_id}, 订单状态: {order_status}")

2.5 取消订阅

如果不再需要接收某个频道的推送，可以使用unsubscribe()方法取消订阅，传入相应的取消订阅参数和回调函数。

args2 = [{"channel": "orders", "instType": "ANY"}]
await ws.unsubscribe(args2, callback=privateCallback)

3. 订单更新推送实现步骤

3.1 准备工作

首先，确保你已经安装了python-okx库。如果尚未安装，可以通过以下命令进行安装：

pip install python-okx

同时，你需要在OKX平台申请API密钥，获取apiKey、passphrase和secretKey。

3.2 编写代码

下面是一个完整的订单更新推送实现示例，基于test/WsPrivateAsyncTest.py文件修改而来：

import asyncio
import json
from okx.websocket.WsPrivateAsync import WsPrivateAsync

def privateCallback(message):
    print("接收到订单更新消息:", message)
    try:
        data = json.loads(message)
        if data.get("event") == "update" and data.get("arg", {}).get("channel") == "orders":
            order_list = data.get("data", [])
            for order_info in order_list:
                order_id = order_info.get("ordId")
                inst_id = order_info.get("instId")
                order_type = order_info.get("ordType")
                order_status = order_info.get("state")
                fill_vol = order_info.get("fillVol")
                print(f"订单ID: {order_id}, 交易对: {inst_id}, 订单类型: {order_type}, 订单状态: {order_status}, 成交数量: {fill_vol}")
    except json.JSONDecodeError as e:
        print("消息解析错误:", e)

async def main():
    url = "wss://wspap.okx.com:8443/ws/v5/private?brokerId=9999"
    # 替换为你的API密钥信息
    api_key = "your_api_key"
    passphrase = "your_passphrase"
    secret_key = "your_secret_key"
    
    ws = WsPrivateAsync(
        apiKey=api_key,
        passphrase=passphrase,
        secretKey=secret_key,
        url=url,
        useServerTime=False
    )
    
    await ws.start()
    # 订阅订单更新频道
    subscribe_args = [{"channel": "orders", "instType": "ANY"}]
    await ws.subscribe(subscribe_args, callback=privateCallback)
    
    # 保持连接30秒，接收消息
    await asyncio.sleep(30)
    
    # 取消订阅订单更新频道
    unsubscribe_args = [{"channel": "orders", "instType": "ANY"}]
    await ws.unsubscribe(unsubscribe_args, callback=privateCallback)
    print("已取消订单更新频道订阅")
    
    # 等待一段时间后关闭连接
    await asyncio.sleep(5)

if __name__ == '__main__':
    asyncio.run(main())