Bourse.WS (bourse v0.9.0)

Copy Markdown View Source

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"])
# Market data arrives as {:websocket_message, decoded_map}. Heartbeat
# replies and other transport control arrive as {:websocket_unmatched_response, _}.
{:ok, observations} = Bourse.WS.health(ws)
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 with ack_timeout_ms: 0
  • {:error, {:subscription_rejected, frame}} — the venue rejected it; frame is 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, coinbaseexchange) 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

All eleven runtime venues have WS transport config. Coinbase Exchange is public-only (matches and heartbeat on the Exchange feed, no credentials).

Summary

Types

What the venue disclosed about the accepted handshake.

One socket's liveness observation from health/1.

t()

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.

Connection-scoped liveness observations for the primary socket and every routed host this connection still owns.

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

auth_info()

@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 opened with authenticate: false. 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, the derive subaccount list after public/login. Bourse.WS.Adapter reads it to schedule renewal without re-running the handshake to find out.

section()

@type section() :: :public | :private

socket_health()

@type socket_health() :: %{
  url: String.t(),
  role: :primary | :routed,
  connection_state: :connecting | :connected | :disconnected,
  heartbeat: map() | nil
}

One socket's liveness observation from health/1.

heartbeat is the locked dependency's get_heartbeat_health/1 map, or nil when the client process cannot be queried.

t()

@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

authenticate(ws, opts \\ [])

@spec authenticate(t(), keyword()) :: {:ok, map()} | {:error, term()}

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_key is not in that set: connect/3 resolves 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).

close(ws)

@spec close(t()) :: :ok

Closes the WebSocket connection and every routed public-host connection it owns.

connect(exchange, section, opts \\ [])

@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. Only :deribit and :ping_pong are accepted; unsupported types return {:error, {:unsupported_heartbeat, type}}. 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.

A :private section whose authored spec declares no auth_pattern is an error (:no_auth_pattern): the socket is closed rather than returned. The one runtime venue with a nil pattern and a working private stream is hyperliquid, and it has no private URL — its private subscriptions are scoped by address on the public socket, which is a different condition.

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.

get_state(ws)

@spec get_state(t()) :: :connecting | :connected | :disconnected

Returns the current connection state (:connecting, :connected, or :disconnected).

get_url(ws)

@spec get_url(t()) :: String.t()

Returns the resolved WS URL this connection is using.

health(ws)

@spec health(t()) ::
  {:ok, [socket_health()]}
  | {:error, :connection_closed | {:connection_owner_down, term()}}

Connection-scoped liveness observations for the primary socket and every routed host this connection still owns.

Heartbeat fields are the locked dependency's get_heartbeat_health/1 map (active_heartbeats, last_heartbeat_at, failure_count, config, timer_active). Market-data silence is not treated as a miss. A missing or closed owner is an error rather than an empty healthy list.

send_message(ws, payload)

@spec send_message(t(), String.t() | map()) :: :ok | {:ok, map()} | {:error, term()}

Sends a raw (already-encoded or map) payload. Delegates to zen_websocket.

subscribe(ws, channels, opts \\ [])

@spec subscribe(t(), [String.t() | map()], keyword() | map()) ::
  :ok | {:error, term()}

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.

unsubscribe(handle)

@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.

watch_order_book(ws, symbol, opts \\ [])

@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}.

watch_orders(ws, opts \\ [])

@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 the venue's template interpolates a symbol. Bybit's order stream is the account-wide order topic and does not need one.

Bybit private order pushes carry an "id" field, so zen_websocket's correlator delivers them as {:websocket_unmatched_response, frame} rather than {:websocket_message, frame} — no in-flight request matches that id. Callers of watch_orders/2 must handle both tags; the bybit trader journey pins this.

watch_ticker(ws, symbol, opts \\ [])

@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.

watch_trades(ws, symbol, opts \\ [])

@spec watch_trades(t(), String.t(), keyword()) ::
  {:ok, Bourse.WS.Handle.t()} | {:error, term()}

Subscribes to public trade updates for symbol.