Realtime¶

The WebSocket (JSON-RPC 2.0) clients for bitFlyer's streaming API, the message they yield, and the helpers that build channel names.

Clients¶

RealtimeClient ¶

RealtimeClient(
    api_key: str = "",
    api_secret: str = "",
    *,
    url: str = REALTIME_URL,
    reconnect: bool = True,
    max_reconnect_attempts: int | None = None,
    reconnect_backoff: float = 1.0,
    max_reconnect_backoff: float = 60.0,
    open_timeout: float | None = 10.0,
    ping_interval: float | None = 20.0,
)

Bases: _RealtimeBase

Synchronous Realtime API client.

Iterate listen() to consume messages. Subscribing connects on demand, so the shortest useful program is::

with RealtimeClient() as rt:
    rt.subscribe(channels.ticker(ProductCode.FX_BTC_JPY))
    for message in rt.listen():
        print(message.data.ltp)

Private channels need credentials, and the client waits for the auth response before it will subscribe to them::

with RealtimeClient(api_key, api_secret) as rt:
    rt.subscribe(channels.CHILD_ORDER_EVENTS)
    for message in rt.listen():
        for event in message.data:
            print(event.event_type, event.child_order_acceptance_id)

Parameters:

Name	Type	Description	Default
`api_key`	`str`	API key. Needed only for the private channels, and the key must carry the "receive order events" permission.	`''`
`api_secret`	`str`	API secret.	`''`
`url`	`str`	WebSocket endpoint. Override to point at a fake server.	`REALTIME_URL`
`reconnect`	`bool`	Reconnect and replay `subscriptions` when the connection drops mid-`listen()`. With this off, a drop raises `websockets.exceptions.ConnectionClosed`.	`True`
`max_reconnect_attempts`	`int \| None`	Give up after this many consecutive failed reconnects, raising `RealtimeError`. `None` retries forever, which is usually what a long-running bot wants.	`None`
`reconnect_backoff`	`float`	Base reconnect delay in seconds; doubles each attempt.	`1.0`
`max_reconnect_backoff`	`float`	Ceiling for a single reconnect delay.	`60.0`
`open_timeout`	`float \| None`	Seconds to wait for the handshake.	`10.0`
`ping_interval`	`float \| None`	Keepalive ping interval, passed to `websockets`. `None` disables keepalive.	`20.0`

A dropped connection loses order events that fired while it was down; the stream has no replay. After a reconnect, reconcile with get_child_orders rather than assuming continuity.

connect ¶

connect() -> None

Open the connection, authenticating when credentials were supplied.

Called for you by subscribe() and listen().

close ¶

close() -> None

Close the connection if one is open. Safe to call repeatedly.

subscribe ¶

subscribe(*channels: str) -> None

Subscribe to one or more channels, connecting first if needed.

Raises:

Type	Description
`RealtimeError`	If the server rejects a channel, or a private channel is requested without authentication.

unsubscribe ¶

unsubscribe(*channels: str) -> None

Stop receiving messages from one or more channels.

listen ¶

listen() -> Iterator[RealtimeMessage]

Yield messages as they arrive, forever.

Frames that are not channel messages, such as late responses to a subscribe, are skipped. When reconnect is on, a dropped connection is re-established and every channel in subscriptions resubscribed before iteration continues.

AsyncRealtimeClient ¶

AsyncRealtimeClient(
    api_key: str = "",
    api_secret: str = "",
    *,
    url: str = REALTIME_URL,
    reconnect: bool = True,
    max_reconnect_attempts: int | None = None,
    reconnect_backoff: float = 1.0,
    max_reconnect_backoff: float = 60.0,
    open_timeout: float | None = 10.0,
    ping_interval: float | None = 20.0,
)

Bases: _RealtimeBase

Asyncio Realtime API client.

The same surface as RealtimeClient with coroutines and an async iterator::

async with AsyncRealtimeClient() as rt:
    await rt.subscribe(channels.executions(ProductCode.BTC_JPY))
    async for message in rt.listen():
        for trade in message.data:
            print(trade.price, trade.size)

Takes the same arguments as RealtimeClient. One instance drives one connection, so drive it from a single task; run several instances if you want concurrent streams.

connect `async` ¶

connect() -> None

Open the connection, authenticating when credentials were supplied.

aclose `async` ¶

aclose() -> None

Close the connection if one is open. Safe to call repeatedly.

subscribe `async` ¶

subscribe(*channels: str) -> None

Subscribe to one or more channels, connecting first if needed.

unsubscribe `async` ¶

unsubscribe(*channels: str) -> None

Stop receiving messages from one or more channels.

listen `async` ¶

listen() -> AsyncIterator[RealtimeMessage]

Yield messages as they arrive, forever.

Messages¶

RealtimeMessage `dataclass` ¶

RealtimeMessage(channel: str, data: Any, raw: Any)

One channelMessage pushed by the server.

Attributes:

Name	Type	Description
`channel`	`str`	The channel it arrived on.
`data`	`Any`	The payload, parsed into the model for that channel — a `Board` for `lightning_board`, a `Ticker` for `lightning_ticker_`, a list of `Execution` for `lightning_executions_*`, and a list of `ChildOrderEvent` or `ParentOrderEvent` for the private channels. A channel this release does not know passes through as decoded JSON.
`raw`	`Any`	The decoded but unvalidated payload, for anything the models drop.

Channels¶

channels ¶

Channel names for the Realtime API.

Build names with these helpers rather than by hand. subscribe validates only the channel prefix, so a typo in the product code — lightning_ticker_BTCJPY — is answered with result: true and then silently delivers nothing, forever.

CHILD_ORDER_EVENTS `module-attribute` ¶

CHILD_ORDER_EVENTS = 'child_order_events'

Your own order lifecycle events. Requires authentication.

PARENT_ORDER_EVENTS `module-attribute` ¶

PARENT_ORDER_EVENTS = 'parent_order_events'

Your own parent order lifecycle events. Requires authentication.

PRIVATE_CHANNELS `module-attribute` ¶

PRIVATE_CHANNELS = frozenset(
    {CHILD_ORDER_EVENTS, PARENT_ORDER_EVENTS}
)

Channels that can only be subscribed after a successful auth.

board_snapshot ¶

board_snapshot(product_code: str) -> str

Full order book snapshots, sent periodically.

Delivery is throttled, and the bids/asks order is not guaranteed.

board ¶

board(product_code: str) -> str

Incremental order book updates.

Each entry carries the new total size at that price; size: 0 means the level is gone. Apply these on top of a board_snapshot to maintain a local book.

ticker ¶

ticker(product_code: str) -> str

Ticker updates.

Throttled, so the ltp here can lag. Use executions if you need every trade.

executions ¶

executions(product_code: str) -> str

Public trades, delivered in batches.