WebSocket entry point. Thin wrapper around ZenWebsocket.Client that binds
a %Bourse.Exchange{} to a connection so subscribe/3 can pick the correct
exchange-native frame builder.
Structure
Pure URL resolution (Bourse.WS.URLRouting) + connection lifecycle (this
module). connect/3 authenticates a :private section before returning it,
so the socket a caller holds is one the venue accepted; authenticate/2
exposes the same handshake for callers driving it themselves. Reconnection,
backoff, heartbeat, and subscription restoration come from zen_websocket.
Bourse.WS.Adapter adds what a long-lived process needs on top: auth state,
re-auth before session expiry, and routing frames onto Bourse.WS.Broadcast.
Usage
{:ok, ws} = Bourse.WS.connect(exchange, :public)
{:ok, sub} = Bourse.WS.watch_ticker(ws, "BTC/USDT")
:ok = Bourse.WS.unsubscribe(sub)
Bourse.WS.close(ws)Lower-level subscribe with pre-formatted channels still works:
:ok = Bourse.WS.subscribe(ws, ["tickers.BTCUSDT"])
# Data messages arrive at the calling process as {:websocket_message, decoded_map}
Bourse.WS.close(ws)Subscribe return shape (unified)
subscribe/3 always returns :ok | {:error, term()} across venues:
:ok— the venue accepted the subscription, or acknowledgement waiting was explicitly disabled withack_timeout_ms: 0{:error, {:subscription_rejected, frame}}— the venue rejected it;frameis the raw exchange envelope{:error, :subscription_ack_timeout}— no accept/reject outcome arrived within the acknowledgement window{:error, reason}— build/send failures (:unsupported_exchange, channel shape errors, transport errors, …)
Correlated JSON-RPC replies (deribit) and asynchronous acks (bybit, okx,
hyperliquid, derive, binance) are classified by Bourse.WS.SubscribeAck.
Rejection frames that arrive asynchronously are still consumed and returned
as errors; non-ack data frames that arrive during the wait are re-queued to
the caller mailbox.
Scope
Seven runtime venues have WS config: binance, binanceusdm, bybit,
deribit, derive, hyperliquid, okx.
Summary
Functions
Runs the venue's auth handshake on an open connection.
Closes the WebSocket connection.
Connects to the exchange's WebSocket endpoint for the given section
(:public or :private).
Returns the current connection state (:connecting, :connected, or :disconnected).
Returns the resolved WS URL this connection is using.
Sends a raw (already-encoded or map) payload. Delegates to zen_websocket.
Sends an exchange-native subscribe frame for the given channels and waits for the venue's accept/reject outcome.
Unsubscribes using a handle from watch_*/3.
Subscribes to order book updates for symbol.
Subscribes to private order updates.
Subscribes to ticker updates for symbol.
Subscribes to public trade updates for symbol.
Types
@type auth_info() :: %{pattern: Bourse.WS.Auth.pattern(), meta: map()}
What the venue disclosed about the accepted handshake.
nil on a public connection, and on a private one that connected without a
handshake. Present, it names the pattern that succeeded and carries the
pattern's own metadata — a ttl_ms where the venue discloses one, the listen
key session where the credential lives in the URL. Bourse.WS.Adapter reads
it to schedule renewal without re-running the handshake to find out.
@type section() :: :public | :private
@type t() :: %Bourse.WS{ auth: auth_info() | nil, exchange: Bourse.Exchange.t(), section: section(), url: String.t(), zen_client: ZenWebsocket.Client.t() }
Functions
Runs the venue's auth handshake on an open connection.
Called for you by connect/3 on a :private section; call it directly only
after connect(exchange, :private, authenticate: false), or to re-authenticate
a connection whose credentials have expired.
Returns {:ok, meta} where meta carries whatever the venue disclosed about
the session — %{ttl_ms: milliseconds} on deribit, %{} where the venue says
nothing. A caller that wants to re-authenticate before expiry reads ttl_ms;
Bourse.WS.Adapter does exactly that.
Errors:
{:error, :no_auth_pattern}— the authored spec declares no handshake{:error, :no_credentials}— the exchange carries none{:error, {:pre_auth_required, data}}— the pattern needs a REST round-trip first (:rest_token), which this function does not perform.:listen_keyis not in that set:connect/3resolves it, and calling this on such a connection returns the session it already holds.{:error, {:auth_failed, reason}}— the venue rejected the credentials{:error, :auth_ack_timeout}— no verdict arrived within the window
Pass auth_timeout_ms: to change the wait (default 10_000).
@spec close(t()) :: :ok
Closes the WebSocket connection.
@spec connect(Bourse.Exchange.t(), section(), keyword()) :: {:ok, t()} | {:error, term()}
Connects to the exchange's WebSocket endpoint for the given section
(:public or :private).
Extra opts are forwarded to ZenWebsocket.Client.connect/2. The connection's
heartbeat config is resolved from Bourse.WS.Config unless the caller overrides
heartbeat_config in opts.
Returns {:error, :unsupported_exchange} if the exchange has no WS config,
or {:error, :no_url_configured} if the requested section is absent.
Private connections authenticate
A :private connection runs the venue's auth handshake before it is handed
back, so a socket a caller holds is one the venue has accepted. A handshake
that fails closes the socket and surfaces the venue's reason — an open but
unauthenticated private connection is never returned, because the failure it
produces later is a silently empty stream rather than an error.
Venues whose authored spec carries no auth_pattern connect without a
handshake: there is no frame to send. Hyperliquid is the real case — its
private subscriptions are scoped by address rather than by a login.
Pass authenticate: false to skip the handshake and drive it yourself with
authenticate/2; the connection is then unauthenticated until you do.
Credentials that live in the URL
The :listen_key venues (binance USD-M and COIN-M) authenticate before the
socket exists: the venue issues a key over REST and it travels as a path
segment. connect/3 performs that round-trip and connects to the resulting
URL, so there is nothing left to authenticate afterwards and
authenticate: false is refused with {:error, {:auth_not_optional, :listen_key}} rather than silently returning a stream that delivers nothing.
@spec get_state(t()) :: :connecting | :connected | :disconnected
Returns the current connection state (:connecting, :connected, or :disconnected).
Returns the resolved WS URL this connection is using.
Sends a raw (already-encoded or map) payload. Delegates to zen_websocket.
Sends an exchange-native subscribe frame for the given channels and waits for the venue's accept/reject outcome.
The frame is built by the exchange's registered subscription_pattern module
(via Bourse.WS.Subscription.build_subscribe/3), encoded as JSON, and sent via
ZenWebsocket.Client.send_message/2.
Pattern modules return either a single map (most exchanges) or a list of maps
(:sub_subscribe and :custom with array_format — HTX/Upbit emit one
frame per channel). List returns are sent sequentially.
Return shape
Always :ok | {:error, term()} — never {:ok, envelope}. Venue rejections
(correlated or async) surface as {:error, {:subscription_rejected, frame}}.
Pass ack_timeout_ms: (keyword or map) to override the async-ack wait
(default 3000); 0 explicitly disables acknowledgement waiting. Other opts
merge into the exchange's subscription_config from Bourse.WS.Config —
used for runtime overrides like a fresh JSON-RPC id. Keys must be atoms to
override the atom-keyed base config; string-keyed maps coexist rather than
override.
Per-frame auth injection (Bourse.WS.Auth.build_subscribe_auth/5, used by the
:rest_token and :inline_subscribe patterns) is not called here. No runtime
venue uses either pattern — both belong to exchanges outside the supported eleven
— so this is a gap only for a venue promoted with one of them.
@spec unsubscribe(Bourse.WS.Handle.t()) :: :ok | {:ok, map()} | {:error, term()}
Unsubscribes using a handle from watch_*/3.
Sends the exchange-native unsubscribe frame built from the stored channels.
@spec watch_order_book(t(), String.t(), keyword()) :: {:ok, Bourse.WS.Handle.t()} | {:error, term()}
Subscribes to order book updates for symbol.
Pass limit: in opts when the exchange template includes {limit}.
@spec watch_orders( t(), keyword() ) :: {:ok, Bourse.WS.Handle.t()} | {:error, term()}
Subscribes to private order updates.
Requires a :private connection (Bourse.WS.connect(exchange, :private)).
Optional symbol: in opts scopes the stream when templates require it.
@spec watch_ticker(t(), String.t(), keyword()) :: {:ok, Bourse.WS.Handle.t()} | {:error, term()}
Subscribes to ticker updates for symbol.
Builds the channel from websocket.subscribe.channels and returns a handle
for unsubscribe/1. Pass channel: to supply a pre-formatted channel when
templates are missing or unresolved.
@spec watch_trades(t(), String.t(), keyword()) :: {:ok, Bourse.WS.Handle.t()} | {:error, term()}
Subscribes to public trade updates for symbol.