client.realtime is the AsyncRealtimeClient that every service subscribe* method uses.
Order updates, market data, balances, and triggers multiplex over the binary Centrifugo Protobuf
WebSocket protocol (not ConnectRPC streaming, not SBE). Prefer the typed service helpers; reach for client.realtime.subscribe_proto(...) only when you need a raw channel.
The transport negotiates centrifuge-protobuf and uses length-delimited Protobuf control frames
and publication payloads. JSON wire mode applies only to ConnectRPC debugging, never to realtime.
Inbound WebSocket messages are capped at 4 MiB and oversized messages fail the subscription
closed.
Subscription contract
Service subscribe methods return an AsyncSubscription[T]:
| API | Behavior |
|---|---|
async for item in sub | Yield decoded events until close/error |
async with sub | Ensures aclose() on exit |
await sub.aclose() | Stop the subscription |
sub.error | Terminal error if the stream failed |
sub.set_on_error(...) | Observe background transport/feed errors |
sub = await client.orders.subscribe(account_id=account_id)
sub.set_on_error(lambda error: print(f"realtime interruption: {error}"))
async with sub:
async for order in sub:
print(order.status, order.order_id)Consume events with the async iterator. set_on_error is for feed-health notification; it does
not replace event consumption. Callback exceptions are isolated from the subscription worker.
Handshake before return
subscribe_proto (and every service helper that uses it) awaits the Centrifugo
connect/subscribe handshake, including private token fetch, before returning the subscription.
- Initial auth / handshake failures raise instead of reconnecting forever in the background.
- Do not treat “no event for a few seconds” as proof that auth succeeded; success means the subscribe call returned without raising.
After a successful handshake, transient transport failures may reconnect (when auto_reconnect=True, the default for most channels). Reconnect uses capped exponential backoff
with per-subscription jitter and resets after a successful resubscription.
Private channels need Account ID
Private channels require API-key credentials and an Account ID. Pass account_id= to the
subscribe helper, or set default_account_id on the client / POLYESTER_ACCOUNT_ID in the
environment.
Without credentials, private subscribe raises synchronously. Hyphenated channel segments such as api-keys are valid for API-key auth; failures here are usually signing or handshake errors.
Fail-closed overflow
PolyesterRealtimeOverflowError and faults the subscription. It does not silently drop updates. Resubscribe after you catch up.Lower-level API
from polyester.codecs.realtime_decode import decode_order_bytes
from polyester.gen.chain.zipper.v1.zipper_pb2 import ZippedAssetSupplyBatch
def decode_supply(payload: bytes) -> ZippedAssetSupplyBatch:
message = ZippedAssetSupplyBatch()
message.ParseFromString(payload)
return message
# Public protobuf example
public_sub = await client.realtime.subscribe_proto(
"public:chain:zipped-asset:supply:proto",
decode=decode_supply,
)
# Private protobuf example, credentials + account scoping required by channel name
private_sub = await client.realtime.subscribe_proto(
f"private:spot:orders:{account_id}:proto",
decode=decode_order_bytes,
)Managed snapshot-then-stream helpers (order book, market overview) typically disable blind
auto-reconnect so they can rebuild REST state between attempts. Their create methods await both
the WebSocket handshake and initial snapshot, and their on_error= callback is isolated from the
worker. See the Streaming guide.