Python WebSocket Guide for Backpack Exchange API

This guide demonstrates how to use the Backpack Exchange WebSocket API with Python. WebSockets provide real-time data streams for market data and account updates.

Prerequisites

Get your API keys if you are going to use private streams:

Install the required Python libraries:

websockets - for WebSocket connections
cryptography - for X-Signature (private streams only)

pip install websockets cryptography

Install dotenv-python to securely manage your keys using environment variables if you are going to use private streams

pip install python-dotenv

Create a .env file and store your keys like this:

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

Create a .gitignore file and add .env to exclude it from version control.

.env

Import the necessary libraries:

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 Basics

The Backpack Exchange WebSocket API is available at wss://ws.backpack.exchange.

WebSocket streams are named using the format: <type>.<symbol>

For example:

depth.SOL_USDC - Order book for SOL/USDC
trade.SOL_USDC - Trades for SOL/USDC

Why Use Async with WebSockets

WebSockets are designed for long-lived connections that receive data in real-time. Using asynchronous programming with WebSockets offers several advantages:

Non-blocking I/O: Async allows your application to handle multiple connections without blocking the main thread.
Resource Efficiency: Async uses fewer resources than creating multiple threads for concurrent connections.
Better Performance: Async can handle many connections with less overhead than synchronous approaches.
Real-time Processing: Async is ideal for real-time data streams where you need to continuously receive and process data.

Synchronous approaches have several drawbacks:

Complex Threading: Requires manual thread management
Resource Intensive: Each connection needs its own thread
Difficult Error Handling: Error propagation across threads is complex
Scaling Issues: Does not scale well with many connections

The async approach (as shown in our examples) is much cleaner, more efficient, and easier to maintain.

Public Streams

Public streams don't require authentication. You can subscribe to them directly.

Example: Subscribing to a Public Stream

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())

Example: Subscribing to Multiple Public Streams

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())

Private Streams

Private streams require authentication with your API keys. These streams are prefixed with account. and provide updates about your account.

Authentication for Private Streams

To authenticate for private streams, you need to:

Create a signature string of the form: instruction=subscribe&timestamp=1614550000000&window=5000
Sign it with your private key
Include the signature data in your subscription message as an array: "signature": ["<verifying key>", "<signature>", "<timestamp>", "<window>"]

Private streams are prefixed with account. and require signature data to be submitted in the subscribe parameters. The verifying key and signature should be base64 encoded.

# 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)
)

Example: Subscribing to a Private Stream

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&timestamp={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 connections require a ping-pong mechanism to keep the connection alive. The good news is that the Python websockets library handles this automatically

Built-in Ping/Pong Support

The websockets library provides native support for WebSocket protocol-level ping/pong frames:

Automatic Pong Responses: The library automatically responds to ping frames with pong frames.
Automatic Ping Sending: The library can be configured to send ping frames at regular intervals.
Connection Health Monitoring: The library can automatically close connections if pong responses aren't received.

How to Configure Ping/Pong

When creating a connection, you can configure the heartbeat mechanism:

async with websockets.connect(
    uri,
    ping_interval=20,  # Send a ping every 20 seconds
    ping_timeout=10    # Wait 10 seconds for a pong response
) as websocket:
    # Your code here

This is much more efficient than implementing application-level ping-pong messages and is handled transparently by the library.

Sources

PreviousBackpack Exchange Python API guide NextWallet

Last updated 1 day ago

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())

# 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&timestamp={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())