跳到主要内容

合约 WebSocket 指南

BigONE 合约 WebSocket API 与现货 API 显著不同。它使用 基于 URL 的订阅 模型用于公共数据，以及 基于请求头的认证 模型用于私有数据流。

准备工作​

  • 公共 URL 基础: wss://api.big.one/ws/contract/v2
  • 私有 URL: wss://api.big.one/ws/contract/v2/stream
  • 认证: 在握手期间通过 HTTP Authorization 头处理。

第一课：公共市场数据 (基于 URL)​

与现货 API（您发送 JSON 消息进行订阅）不同，合约 API 根据您连接的 URL 路径 自动订阅。

逻辑​

要观看特定频道，您需要打开一个连接到该特定 URL 的新连接。如果您想观看多个市场，则需要打开多个连接。

可用频道:

  • 订单簿: .../depth@{symbol}
  • 成交: .../trades@{symbol}
  • K 线: .../candlesticks/{period}@{symbol}

代码示例：观看 BTCUSD 订单簿​

import asyncio
import websockets
import json

# 注意: symbol 是 URL 的一部分
URL = "wss://api.big.one/ws/contract/v2/depth@BTCUSD"

async def watch_depth():
    async with websockets.connect(URL) as ws:
        print("Connected to BTCUSD Order Book")
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            
            # 第一条消息是全量快照
            # 后续消息是增量更新
            print(f"Update received. Best Bid: {data['bestPrices']['bid']}")

if __name__ == "__main__":
    asyncio.run(watch_depth())

第二课：私有数据流 (Header 认证)​

要接收关于您的 仓位、订单 和 余额 的更新，您需要连接到一个统一的 "Stream" 频道。

逻辑​

  1. 生成标准的 JWT。
  2. 在 WebSocket 握手期间将 JWT 传递给 Authorization 头。
  3. 服务器验证 Header。如果无效，连接将以 401 错误关闭。

私有数据流 URL: wss://api.big.one/ws/contract/v2/stream

代码示例：私有更新​

import asyncio
import websockets
import json
import time
import jwt

API_KEY = "YOUR_KEY"
SECRET_KEY = "YOUR_SECRET"

def generate_token():
    payload = {
        "sub": API_KEY,
        "nonce": int(time.time() * 1000000),
        "iat": int(time.time()),
        "exp": int(time.time()) + 60
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

async def watch_private():
    token = generate_token()
    headers = {"Authorization": f"Bearer {token}"}
    
    url = "wss://api.big.one/ws/contract/v2/stream"
    
    async with websockets.connect(url, extra_headers=headers) as ws:
        print("Connected to Private Stream")
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            
            # 响应是一个包含对象
            if "positions" in data:
                for pos in data['positions']:
                    print(f"Position Update: {pos['symbol']} size: {pos['size']}")
            
            if "orders" in data:
                for order in data['orders']:
                    print(f"Order {order['id']} is {order['status']}")

if __name__ == "__main__":
    asyncio.run(watch_private())

完整示例：多流机器人​

对于真正的交易机器人，通常需要结合两者。

import asyncio
import websockets
import json

async def main():
    # 并发运行公共和私有监听器
    await asyncio.gather(
        watch_depth(),   # 来自第一课
        watch_private()  # 来自第二课
    )

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

结论​

合约 WebSocket API 专为高性能而设计，采用了基于 URL 的路由。

要点:

  • URL 订阅: 每个公共市场/频道都需要自己的连接。
  • Header 认证: 私有流在握手期间认证，而不是通过消息。
  • 快照: 期望每个频道的第一条消息都是全量快照。

下一步: