Bourse.WS (bourse v0.7.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"])
# 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 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) 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

Types

What the venue disclosed about the accepted handshake.

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.

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

section()

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

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

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.

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 templates require it.

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.