radion-sdk

Official, async-first, fully-typed SDK for the Radion platform.

One client, one API key, every Radion product surface. Install radion-sdk; import it as radion.

radion = Radion(api_key=os.getenv("RADION_API_KEY"))
await radion.realtime.connect()
await radion.realtime.subscribe(Subscription(id="trading", channel="trading"))

@radion.realtime.on_channel("trading")
async def handle_trade(event):
    print(event.id, event.data)

Features

• Unified client — Radion(api_key=...) is the single entry point for every product surface
• Auto-reconnect — exponential backoff with jitter; stops on graceful shutdown
• Subscription restore — active channels are re-subscribed after every reconnect
• Heartbeats — ping/pong keep-alive that detects stale connections and reconnects
• Typed end-to-end — channel names, frame models, and errors
• Async-first — built on asyncio; handlers may be sync or async
• Webhook helpers — verify delivery signatures and parse bodies into the same typed events, stdlib only

Requirements

• Python >= 3.10

Install

uv add radion-sdk
# or: pip install radion-sdk

Quick start

import asyncio
import os

from radion import Radion, Subscription


async def main() -> None:
    radion = Radion(api_key=os.getenv("RADION_API_KEY"))

    await radion.realtime.connect()
    await radion.realtime.subscribe(Subscription(id="trading", channel="trading"))

    @radion.realtime.on_channel("trading")
    async def handle_trade(event):
        print(event.id, event.channel, event.data)

    await asyncio.sleep(60)
    await radion.realtime.close()


asyncio.run(main())

Usage

Configuration

Radion(api_key, *, base_url=..., ws_url=..., token=None, auth_in_query=False,
       reconnect=True, heartbeat=True, heartbeat_interval=15.0,
       heartbeat_timeout=10.0)

Argument	Type	Default	Description
api_key	str	—	Required. Sent as the X-API-Key header.
base_url	str	https://api.radion.app	Base URL for the Radion API.
ws_url	str	wss://api.radion.app/ws	Override the realtime endpoint.
token	str | callable	None	User JWT for the pk_jwt_ flow (see below).
auth_in_query	bool	False	Send credentials in the WS URL query string.
reconnect	bool	True	Auto-reconnect on unexpected disconnect.
heartbeat	bool	True	Enable heartbeats / stale detection.
heartbeat_interval	float	15.0	Seconds between pings.
heartbeat_timeout	float	10.0	Extra grace before a quiet connection is stale.

Extra keyword arguments are forwarded to the realtime client.

Authentication

Two credential schemes, both keyed on api_key (sent as X-API-Key):

Secret key (sk_ / rk_):

radion = Radion(api_key=os.getenv("RADION_API_KEY"))

Public JWT (pk_jwt_) — pair the public key with a per-user JWT minted by your own auth provider. Pass token as a callable so a fresh JWT is fetched on every (re)connect:

radion = Radion(api_key="pk_jwt_...", token=fetch_user_jwt)  # str, or sync/async callable

Set auth_in_query=True to move credentials into the WS URL query string — for a proxy or gateway that strips the X-API-Key / Authorization headers.

Realtime client

radion.realtime is a RealtimeClient. It can also be imported and constructed standalone:

from radion import RealtimeClient

client = RealtimeClient(api_key=os.getenv("RADION_API_KEY"))

Method	Description
await connect()	Open the connection. Resolves once established.
await subscribe(subscription)	Subscribe with Subscription(id, channel, filters). Replayed on reconnect.
await unsubscribe(id)	Unsubscribe by subscription id.
on_channel(channel)	Decorator for a specific channel handler (fires for both the confirmed and pending feed).
off_channel(channel, handler=None)	Remove a channel handler (or all for that channel).
on_any_channel()	Decorator for a wildcard handler — every channel event.
off_any_channel(handler=None)	Remove a wildcard handler (or all of them).
on_lifecycle(event)	Decorator for a lifecycle handler: open, close, reconnect, error, warning.
off_lifecycle(event, handler=None)	Remove a lifecycle handler (or all for that event).
await close(code=1000, ...)	Graceful shutdown. Stops reconnect attempts.
connected	Property — whether the socket is currently open.

Subscriptions & filters

A subscription is Subscription(id, channel, filters=None, confirmed=True). The id is your own string, echoed back on every event so you can tell subscriptions apart. Some channels require a filter:

from radion import ChannelFilters, Subscription

await radion.realtime.subscribe(
    Subscription(id="whales", channel="trading", filters=ChannelFilters(min_usd=10_000))
)

# on_any_channel fires for every channel; the event carries id + channel + data.
@radion.realtime.on_any_channel()
async def on_any(event):
    print(event.id, event.channel, event.data)

ChannelFilters: wallets, market_ids, token_ids, min_usd.

Every event also carries seq — a per-connection counter that increments by one per event frame, so a jump means dropped frames — and sent_at_ms, the Unix ms the server sent the frame. Server→client latency is your receive time minus event.sent_at_ms. Pending (mempool) events additionally keep data["seen_at_ms"], the block-detection time, for block→client latency.

Channels

trading · fees · oracle · resolution · lifecycle · positions
combos · transfers · accounts · wallets · markets

Available as the CHANNELS tuple and the Channel type.

from radion import CHANNELS, Subscription

for channel in CHANNELS:
    await radion.realtime.subscribe(Subscription(id=channel, channel=channel))

Confirmed vs pending feed

Every topic and filter channel serves two feeds. Subscribe with the default confirmed=True for confirmed on-chain events, or confirmed=False for the pending feed — same bare channel name, no mempool. prefix:

await radion.realtime.subscribe(
    Subscription(id="pending-trades", channel="trading", confirmed=False)
)

@radion.realtime.on_channel("trading")
async def on_trade(event):
    if event.confirmed:
        print("confirmed", event.data)
    else:
        # pending event: event.data is a MempoolPayload
        print("pending", event.data["call"]["notional_usd"])

A single on_channel handler fires for both feeds; tell them apart with event.confirmed (or route by event.id). For min_usd, the confirmed feed matches the actual filled USD, the pending feed the intended fill notional (call.notional_usd). If the node has no pending stream, the server sends a non-fatal mempool_unavailable warning on the warning lifecycle event:

@radion.realtime.on_lifecycle("warning")
async def on_warning(info):
    print(info["code"], info["message"])

CLOB channels

The CLOB (central-limit order book) family is a separate, first-class set of subscribable channels:

clob.book · clob.prices · clob.last_trade
clob.midpoint · clob.tick_size · clob.best_bid_ask

Available as the CLOB_CHANNELS tuple and the ClobChannel type. Each CLOB channel requires a token_ids filter and has no mempool. companion. Their payloads carry no type discriminator.

from radion import ChannelFilters, Subscription

await radion.realtime.subscribe(
    Subscription(
        id="book",
        channel="clob.book",
        filters=ChannelFilters(token_ids=["123..."]),
    )
)

@radion.realtime.on_channel("clob.book")
async def on_book(event):
    print(event.data["bids"], event.data["asks"])

Lifecycle events

@radion.realtime.on_lifecycle("open")
async def opened(_):
    print("connected")

@radion.realtime.on_lifecycle("close")
async def closed(info):
    print(info["code"], info["reason"])

@radion.realtime.on_lifecycle("reconnect")
async def reconnecting(info):
    print(info["attempt"], info["delay"])

@radion.realtime.on_lifecycle("error")
async def errored(err):
    print(err)

Reconnect & subscription restore

On an unexpected disconnect the client reconnects with exponential backoff and re-sends every active subscription once the socket reopens. After close() no further attempts run.

Heartbeats

A ping is sent every heartbeat_interval seconds. Any inbound frame counts as liveness; if the connection goes quiet past the timeout window it is terminated and reconnected.

Webhooks

Radion webhooks POST the same event frames as the WebSocket, signed with your endpoint's secret. The SDK ships standalone helpers to consume them — no client instance needed: verify_webhook_signature checks the X-Radion-Signature HMAC (constant-time, with replay protection), and parse_webhook_event validates the body into the same typed EventFrame the realtime client emits. Standard library only — works in any web framework.

import os

from radion import parse_webhook_event, verify_webhook_signature

async def handler(request):  # any framework: raw body + headers
    raw_body = await request.body()

    authentic = verify_webhook_signature(
        payload=raw_body,  # raw bytes, exactly as received
        secret=os.environ["RADION_WEBHOOK_SECRET"],  # whsec_…
        signature=request.headers.get("x-radion-signature", ""),
        timestamp=request.headers.get("x-radion-timestamp", ""),
    )
    if not authentic:
        return Response(status_code=401)

    event = parse_webhook_event(raw_body)
    if event is not None and event.channel == "trading":
        print(event.seq, event.data)
    return Response(status_code=200)

Argument	Type	Default	Description
payload	str | bytes	—	Required. Raw request body, exactly as received.
signature	str	—	Required. The X-Radion-Signature header.
timestamp	int | str	—	Required. The X-Radion-Timestamp header (Unix ms).
secret	str | Sequence[str]	—	Required. Signing secret(s); pass both during a rotation window.
tolerance_ms	int	5 minutes	Max delivery age before it is rejected as a replay.

Verify over the raw body exactly as received — parse JSON only after the signature checks out. Deliveries are retried and unordered; deduplicate on (id, seq) or on fields of data.

Error handling

from radion import RadionConnectionError, RadionServerError

@radion.realtime.on_lifecycle("error")
async def errored(err):
    if isinstance(err, RadionServerError):
        print("server error", err.code, err.channel)
    elif isinstance(err, RadionConnectionError):
        print("connection error", err)

A throwing consumer handler is reported via the error event and never retried.

License

MIT