Bourse.WS (bourse v0.5.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)
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 (: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

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.

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

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.

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.

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.

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