Bourse.WS (bourse v0.2.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.

Layer 1+2 (Task 92)

Pure URL resolution (Bourse.WS.URLRouting) + connection lifecycle (this module). Layer 3 (auth state machine, custom reconnection) is deliberately deferred — zen_websocket covers reconnection, backoff, heartbeat, and subscription restoration natively.

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

Functions

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

section()

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

t()

@type t() :: %Bourse.WS{
  exchange: Bourse.Exchange.t(),
  section: section(),
  url: String.t(),
  zen_client: ZenWebsocket.Client.t()
}

Functions

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.

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.

TODO(T94): :rest_token (kraken) and :inline_subscribe (coinbase) auth patterns require per-frame auth injection via Bourse.WS.Auth.build_subscribe_auth/5, which this function does not call. Private subscribes on those exchanges ship unauthenticated until the adapter layer lands (see CHANGELOG T94).

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.