# Руководство по Python WebSocket для API Backpack Exchange
Это руководство демонстрирует, как использовать Backpack Exchange WebSocket API с Python. WebSockets обеспечивают потоки данных в реальном времени для рыночных данных и обновлений аккаунта.
### Предварительные требования
Получите ваши API ключи, если вы собираетесь использовать приватные потоки:
Установите необходимые библиотеки 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
### Почему использовать Async с WebSockets
WebSockets предназначены для долгоживущих соединений, которые получают данные в реальном времени. Использование асинхронного программирования с WebSockets предлагает несколько преимуществ:
1. **Неблокирующий I/O**: Async позволяет вашему приложению обрабатывать множество соединений без блокировки основного потока.
2. **Эффективность ресурсов**: Async использует меньше ресурсов, чем создание множества потоков для параллельных соединений.
3. **Лучшая производительность**: Async может обрабатывать много соединений с меньшими накладными расходами, чем синхронные подходы.
4. **Обработка в реальном времени**: Async идеален для потоков данных в реальном времени, где вам нужно непрерывно получать и обрабатывать данные.
Синхронные подходы имеют несколько недостатков:
1. **Сложное управление потоками**: Требует ручного управления потоками
2. **Ресурсоемкость**: Каждое соединение нуждается в своем собственном потоке
3. **Сложная обработка ошибок**: Распространение ошибок через потоки является сложным
4. **Проблемы масштабирования**: Плохо масштабируется при многих соединениях
Асинхронный подход (как показано в наших примерах) намного чище, эффективнее и проще в обслуживании.
### Публичные потоки
Публичные потоки не требуют аутентификации. Вы можете подписаться на них напрямую.
#### Пример: Подписка на публичный поток
```
async def subscribe_to_public_stream():
uri = "wss://ws.backpack.exchange"
async with websockets.connect(uri) as websocket:
# Subscribe to the depth stream for SOL/USDC
subscribe_message = {
"method": "SUBSCRIBE",
"params": ["depth.SOL_USDC"]
}
await websocket.send(json.dumps(subscribe_message))
print(f"Subscribed to depth.SOL_USDC stream")
# Process incoming messages
while True:
response = await websocket.recv()
data = json.loads(response)
print(f"Received: {data}")
# You can process the data here based on your needs
# For example, update a local order book
# To run the async function in a Jupyter notebook, use:
# await subscribe_to_public_stream()
#
# TO run in your code
# 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 to multiple streams
subscribe_message = {
"method": "SUBSCRIBE",
"params": ["depth.SOL_USDC", "trade.SOL_USDC"]
}
await websocket.send(json.dumps(subscribe_message))
print(f"Subscribed to multiple streams")
# Process incoming messages
while True:
response = await websocket.recv()
data = json.loads(response)
print(f"Received: {data}")
# Process different stream data based on the stream name
if "stream" in data and "data" in data:
stream_name = data["stream"]
stream_data = data["data"]
if stream_name.startswith("depth."):
# Process order book data
print(f"Order book update: {stream_data}")
elif stream_name.startswith("trade."):
# Process trade data
print(f"Trade update: {stream_data}")
# Run the async function in a Jupyter notebook:
# await subscribe_to_multiple_streams()
#
# To run in your code
# asyncio.run(subscribe_to_multiple_streams())
```
### Приватные потоки
Приватные потоки требуют аутентификации с вашими API ключами. Эти потоки имеют префикс `account.` и предоставляют обновления о вашем аккаунте.
#### Аутентификация для приватных потоков
Для аутентификации приватных потоков вам нужно:
1. Создать строку подписи в форме: `instruction=subscribe×tamp=1614550000000&window=5000`
2. Подписать её вашим приватным ключом
3. Включить данные подписи в ваше сообщение подписки как массив: `"signature": ["", "", "", ""]`
Приватные потоки имеют префикс `account.` и требуют отправки данных подписи в параметрах подписки. Ключ проверки и подпись должны быть закодированы в base64.
```
# Load API keys from .env file
# load_dotenv(find_dotenv())
# public_key = os.getenv("PUBLIC_KEY")
# secret_key = os.getenv("SECRET_KEY")
# For demonstration purposes only - don't hardcode keys in production
public_key = "5+yQgwU0ZdJ/9s+GXfuPFfo7yQQpl9CgvQedJXne30o="
secret_key = "TDSkv44jf/iD/QCKkyCdixO+p1sfLXxk+PZH7mW/ams="
# Create private key from secret key
private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
base64.b64decode(secret_key)
)
```
#### Пример: Подписка на приватный поток
```
async def subscribe_to_private_stream():
uri = "wss://ws.backpack.exchange"
# Generate authentication parameters
timestamp = int(time() * 1e3) # Unix time in milliseconds
window = "5000" # Time window in milliseconds
# Create signature string
sign_str = f"instruction=subscribe×tamp={timestamp}&window={window}"
# Sign the string
signature_bytes = private_key.sign(sign_str.encode())
encoded_signature = base64.b64encode(signature_bytes).decode()
async with websockets.connect(uri) as websocket:
# Subscribe to the order stream with authentication
subscribe_message = {
"method": "SUBSCRIBE",
"params": ["account.orderUpdate"],
"signature": [public_key, encoded_signature, str(timestamp), window]
}
await websocket.send(json.dumps(subscribe_message))
print(f"Subscribed to account.order stream")
# Process incoming messages
while True:
response = await websocket.recv()
data = json.loads(response)
print(f"Received: {data}")
# Process order updates
if "stream" in data and data["stream"] == "account.order" and "data" in data:
order_data = data["data"]
print(f"Order update: {order_data}")
# Run the async function in a Jupyter notebook:
await subscribe_to_private_stream()
#
# If using nest_asyncio (recommended):
# asyncio.run(subscribe_to_private_stream())
```
### WebSocket Ping/Pong
Соединения WebSocket требуют механизма ping-pong для поддержания соединения в активном состоянии. Хорошая новость в том, что библиотека Python `websockets` обрабатывает это автоматически
### Источники
Для получения дополнительной информации посетите официальную документацию:
---
# 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/ru/birzha/api-i-dokumentaciya-dlya-razrabotchikov/rukovodstvo-po-python-websocket-dlya-api-backpack-exchange.md?ask=
```
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.