# Python WebSocket 指南

适用于 Backpack Exchange API

本指南演示了如何使用 Python 连接 Backpack Exchange 的 WebSocket API。\
WebSocket 提供用于市场数据和账户更新的实时数据流。

### 前提条件

若要使用私有数据流，请先获取您的 API 密钥：\
<https://backpack.exchange/portfolio/settings/api-keys>

安装所需的 Python 库：

• websockets - 用于建立 WebSocket 连接\
• cryptography - 用于生成 X-Signature（仅用于私有数据流）

```
pip install websockets cryptography
```

如果使用私有数据流，建议安装 dotenv-python，通过环境变量安全管理密钥

```
pip install python-dotenv
```

创建 `.env` 文件并如下存储您的密钥：

```
PUBLIC_KEY=zDIJj9qneWIY0IYZ5aXoHcNMCm+XDhVcTssiT0HyY0A=
SECRET_KEY=4odxgSUxFrC/zsKWZF4OQwYAgnNu9hnWH3NxWfLAPz4=
```

创建 `.gitignore` 文件，并添加 `.env`，以避免其被纳入版本控制。

```
.env
```

导入所需库：

```
import json
import asyncio
import websockets
import base64
from time import time
import os
from cryptography.hazmat.primitives.asymmetric import ed25519
from dotenv import load_dotenv, find_dotenv
```

### WebSocket API 基础

Backpack Exchange WebSocket API 地址为：`wss://ws.backpack.exchange`

WebSocket 流的命名格式为：`<类型>.<交易对>`

例如：

• `depth.SOL_USDC` - SOL/USDC 的订单簿\
• `trade.SOL_USDC` - SOL/USDC 的成交信息

### 为什么使用异步方式处理 WebSocket？

WebSocket 适用于持续连接并实时接收数据的场景。使用异步（async）方式编程具有以下优势：

1. **非阻塞 I/O**：异步可处理多个连接而不阻塞主线程。
2. **资源高效**：相比多线程，异步使用更少的资源。
3. **性能更佳**：处理大量连接时，异步开销更小。
4. **实时处理**：适用于需持续接收和处理数据的场景。

传统同步方式存在诸多缺点：

1. **线程管理复杂**：需要手动管理多个线程
2. **资源占用高**：每个连接都需独立线程
3. **错误处理困难**：线程间错误传播复杂
4. **扩展性差**：不易扩展以支持更多连接

而异步方式（本指南中的示例）更简洁、高效且更易维护。

### 公共数据流

无需身份验证即可订阅公共数据流，直接连接即可。

#### 示例：订阅公共数据流

```
async def subscribe_to_public_stream():
    uri = "wss://ws.backpack.exchange"

    async with websockets.connect(uri) as websocket:
        # 订阅 SOL/USDC 的订单簿流
        subscribe_message = {
            "method": "SUBSCRIBE",
            "params": ["depth.SOL_USDC"]
        }

        await websocket.send(json.dumps(subscribe_message))
        print(f"已订阅 depth.SOL_USDC 数据流")

        # 处理收到的消息
        while True:
            response = await websocket.recv()
            data = json.loads(response)
            print(f"收到消息：{data}")

            # 可在此根据业务需求处理数据
            # 例如更新本地订单簿
#要在Jupyter Notebook中运行异步函数，请使用：
#await subscribe_to_public_stream()
#
#在您的代码中运行：
# asyncio.run(subscribe_to_public_stream())  
```

#### 示例：订阅多个公共数据流

```
async def subscribe_to_multiple_streams():
    uri = "wss://ws.backpack.exchange"

    async with websockets.connect(uri) as websocket:
        # 订阅多个数据流
        subscribe_message = {
            "method": "SUBSCRIBE",
            "params": ["depth.SOL_USDC", "trade.SOL_USDC"]
        }

        await websocket.send(json.dumps(subscribe_message))
        print(f"已订阅多个数据流")

        while True:
            response = await websocket.recv()
            data = json.loads(response)
            print(f"收到消息：{data}")

            # 根据流名处理不同类型数据
            if "stream" in data and "data" in data:
                stream_name = data["stream"]
                stream_data = data["data"]

                if stream_name.startswith("depth."):
                    print(f"订单簿更新：{stream_data}")
                elif stream_name.startswith("trade."):
                    print(f"成交信息更新：{stream_data}")
#在 Jupyter notebook 中运行：
# await subscribe_to_multiple_streams()
#
#在代码中运行：
#asyncio.run(subscribe_to_multiple_streams())
```

### 私有数据流

私有数据流以 `account.` 开头，需要 API 密钥认证，可获取账户相关更新信息。

#### 私有流的身份认证步骤：

1. 构造签名字符串，格式如下：\
   `instruction=subscribe&timestamp=1614550000000&window=5000`
2. 使用私钥进行签名
3. 在订阅消息中添加如下签名字段（数组形式）：\
   `"signature": ["<验证公钥>", "<签名>", "<时间戳>", "<时间窗>"]`

验证公钥和签名需使用 base64 编码。

```
# 从 .env 文件加载 API 密钥
# load_dotenv(find_dotenv())
# public_key = os.getenv("PUBLIC_KEY")
# secret_key = os.getenv("SECRET_KEY")

# 仅用于演示，请勿在生产环境中硬编码密钥
public_key = "5+yQgwU0ZdJ/9s+GXfuPFfo7yQQpl9CgvQedJXne30o="
secret_key = "TDSkv44jf/iD/QCKkyCdixO+p1sfLXxk+PZH7mW/ams="

# 使用密钥创建私钥对象
private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
    base64.b64decode(secret_key)
)
```

***

#### 示例：订阅私有数据流

```
async def subscribe_to_private_stream():
    uri = "wss://ws.backpack.exchange"

    # 生成认证参数
    timestamp = int(time() * 1e3)  # 毫秒时间戳
    window = "5000"  # 有效时间窗（毫秒）

    # 创建签名字符串
    sign_str = f"instruction=subscribe&timestamp={timestamp}&window={window}"

    # 生成签名
    signature_bytes = private_key.sign(sign_str.encode())
    encoded_signature = base64.b64encode(signature_bytes).decode()

    async with websockets.connect(uri) as websocket:
        # 带身份验证订阅订单流
        subscribe_message = {
            "method": "SUBSCRIBE",
            "params": ["account.orderUpdate"],
            "signature": [public_key, encoded_signature, str(timestamp), window]
        }

        await websocket.send(json.dumps(subscribe_message))
        print(f"已订阅 account.order 数据流")

        while True:
            response = await websocket.recv()
            data = json.loads(response)
            print(f"收到消息：{data}")

            # 处理订单更新
            if "stream" in data and data["stream"] == "account.order" and "data" in data:
                order_data = data["data"]
                print(f"订单更新：{order_data}")
#在 Jupyter notebook 中运行该函数：
#await subscribe_to_private_stream()
#若使用 nest_asyncio（推荐）：
#asyncio.run(subscribe_to_private_stream())
```

### WebSocket 保活机制：Ping/Pong

WebSocket 连接需通过 Ping-Pong 保持连接活跃。好消息是：Python 的 `websockets` 库已自动处理该机制。

### 参考资料

有关更多信息，请访问官方文档：<https://docs.backpack.exchange/#tag/Streams>


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://support.backpack.exchange/support-docs/cn/jiao-yi-suo/api-yu-kai-fa-zhe-wen-dang/python-websocket-zhi-nan.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.