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)
:ok = 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 (:websocket_not_configured,:unsupported_exchange, channel shape errors, transport errors, …)
Channels that span more than one authored host are one call: every host is accepted, or hosts that already succeeded are unsubscribed and the failing host's error is returned. A retry of the same list does not stack a leftover subscription on the host that worked.
Correlated JSON-RPC replies (deribit) and asynchronous acks (alpaca, bybit,
okx, hyperliquid, derive, binance, lighter) 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. Lighter's subscribed/* reply is both the acknowledgement
and the first snapshot, so it is re-queued after subscribe/3 returns :ok.
Scope
Ten runtime venues have WS transport config. Coinbase Exchange remains the registered runtime venue without one.
Summary
Functions
Runs the venue's auth handshake on an open connection.
Closes the WebSocket connection and every routed public-host connection it owns.
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, connect_fun: (String.t(), keyword() -> {:ok, ZenWebsocket.Client.t()} | {:error, term()}), connect_opts: keyword(), connection_owner: pid() | 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 and every routed public-host connection it owns.
@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. connect_fun: replaces ZenWebsocket.Client.connect/2
for instrumented transports and is reused for authored secondary hosts.
Returns {:error, :websocket_not_configured} if a runtime-supported exchange
has no WS config, {:error, :unsupported_exchange} if the exchange itself is
unsupported, 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.
Alpaca's public market-data socket also requires its key/secret handshake;
its config declares :public in auth_sections, so the same guarantee
applies there without enabling the private trading stream.
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 in the URL.
USD-M uses the provider's listenKey and events query parameters; COIN-M
keeps its 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.
Routed hosts stay with the originating Bourse.WS until close/1.
@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. The handle carries the effective socket, which can differ
from the supplied socket when a stream uses another authored host. 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.