Code Examples

This page provides complete code examples for common WebSocket operations using the Paradex API.

Public market data channels are delivered as SBE binary frames. The examples below use the Paradex Python SDK, which negotiates the encoding and decodes frames for you. See Binary Encoding (SBE) for the wire format, and for generating a decoder in another language.

Prerequisites

pip install "paradex-py>=0.7.0"

Establishing a Connection

import asyncio
from paradex_py.api.ws_client import ParadexWebsocketClient
from paradex_py.environment import TESTNET
async def main():
    ws_client = ParadexWebsocketClient(env=TESTNET, sbe_enabled=True)
    await ws_client.connect()
    print("Connected")
asyncio.run(main())

Passing sbe_enabled=True appends the schema parameters to the connection URL. Omitting it falls back to JSON, which is deprecated for public channels.

Subscribing to a Channel

import asyncio
from paradex_py.api.ws_client import ParadexWebsocketChannel, ParadexWebsocketClient
from paradex_py.environment import TESTNET
async def on_message(channel, message):
    print(channel, message["params"]["data"])
async def main():
    ws_client = ParadexWebsocketClient(env=TESTNET, sbe_enabled=True)
    await ws_client.connect()
    await ws_client.subscribe(
        ParadexWebsocketChannel.TRADES,
        callback=on_message,
        params={"market": "ETH-USD-PERP"},
    )
    await asyncio.sleep(30)
asyncio.run(main())

Frames are decoded before your callback runs, so the payload arrives as a plain dictionary. Subscribe requests and their acknowledgements remain JSON-RPC text frames; only the channel payload is binary.

Authenticating a Connection

Private channels require authentication. Build the client with SBE enabled, then attach the account:

import asyncio
from paradex_py import Paradex
from paradex_py.api.ws_client import ParadexWebsocketChannel, ParadexWebsocketClient
from paradex_py.environment import TESTNET
async def on_message(channel, message):
    print(channel, message["params"]["data"])
async def main():
    paradex = Paradex(
        env=TESTNET,
        l1_address="0xYOUR_L1_ADDRESS",
        l2_private_key="0xYOUR_L2_PRIVATE_KEY",
    )
    ws_client = ParadexWebsocketClient(env=TESTNET, sbe_enabled=True)
    ws_client.init_account(paradex.account)
    await ws_client.connect()
    await ws_client.subscribe(
        ParadexWebsocketChannel.FILLS,
        callback=on_message,
        params={"market": "ETH-USD-PERP"},
    )
    await asyncio.sleep(30)
asyncio.run(main())

Private channels have no SBE encoder, so their payloads are JSON either way. sbe_enabled=True matters here only for the public channels the same connection subscribes to.

Unsubscribing from a Channel

await ws_client.unsubscribe_by_name("trades.ETH-USD-PERP")

Implementation Notes

• Encoding: Public channels are SBE binary. Use sbe_enabled=True, or generate a decoder from the schema if you are not using the Python SDK.
• Error Handling: These examples include no error handling. In production, handle decode errors and connection loss explicitly.
• Authentication: Never commit private keys. Load them from the environment or a secret store.
• Reconnection: The SDK reconnects automatically. If you implement your own client, use exponential backoff and re-subscribe after reconnecting.