Skip to content

Streaming

The stream package delivers real-time Market Data over MQTT. Subscribe and unsubscribe are HTTP calls made through the core client; the MQTT connection only carries pushes. Requires a valid token (see Authentication).

Prerequisites

Error handling

All SDK functions return error. See Errors for the typed error model, transient vs permanent classification, and retry patterns.

Build a streaming client from a configured *client.Client:

cl, err := client.New(client.WithEnv())
if err != nil {
    return err
}
defer func() { _ = cl.Close() }()

if _, err := cl.EnsureToken(ctx); err != nil {
    return err
}

s, err := stream.New(cl, stream.WithWebSocket(true))
if err != nil {
    return err
}
defer func() { _ = s.Close() }()

The broker address comes from the client's resolved endpoints. WithWebSocket(true) selects the MQTT-over-WebSocket endpoint (wss://...:8883/mqtt); otherwise the plain TCP endpoint (...:1883) is used. WithMQTTURL overrides both.

Connect and subscribe

if err := s.Connect(ctx); err != nil {
    return err
}
if err := s.Subscribe(ctx, stream.SubscribeRequest{
    Symbols:  []string{"AAPL"},
    Category: stream.CategoryUSStock,
    SubTypes: []stream.SubType{stream.SubTypeQuote, stream.SubTypeSnapshot, stream.SubTypeTick},
    Grab:     true,
}); err != nil {
    return err
}

SubscribeRequest fields:

Field Purpose
SessionID Overrides the client session id for this call; rarely needed
Symbols Symbols to subscribe to; at most 100 per request
Category Security type: CategoryUSStock, CategoryUSETF, CategoryHKStock, CategoryCNStock (validated with Category.Valid())
SubTypes Data types: SubTypeQuote, SubTypeSnapshot, SubTypeTick (validated with SubType.Valid())
Grab Request an immediate snapshot push on subscription
Depth Level-2 order-book depth as a string (optional; defaults to "10", US stocks up to "50")
OvernightRequired Include the overnight session for US stocks

Unsubscribe mirrors it and accepts UnsubscribeAll: true to cancel every subscription for the session.

Handlers

Register handlers before or after Connect; each may be called more than once.

Handler Payload Fires on
OnQuote *marketdatav1.Quote Order-book pushes
OnSnapshot *marketdatav1.Snapshot Market snapshots
OnTick *marketdatav1.Tick Tick-by-tick trades
OnNotice []byte (JSON) Server notifications
OnError error Async errors (decode failures, unknown topics, resubscribe failures)
OnConnect none Initial connect and every reconnect
OnDisconnect error A live connection was lost 
s.OnSnapshot(func(snap *marketdatav1.Snapshot) {
    log.Printf("%s %s", snap.GetBasic().GetSymbol(), snap.GetPrice())
})

The marketdatav1 package is github.com/shing1211/webullapi4go/gen/webull/marketdata/v1.

Options

Option Purpose
WithSessionID, WithClientID Set the MQTT client id / session id (a unique id is generated by default)
WithMQTTURL Override the broker address (tcp://, wss://, or bare host:port)
WithWebSocket(bool) Select the WebSocket endpoint
WithAutoReconnect(bool) Enable the background reconnect loop
WithAutoResubscribe(bool) Re-issue active subscriptions after a reconnect (default true)
WithResubscribeTimeout(d) Bound the whole re-subscription sequence
WithKeepAlive(d) MQTT keep-alive interval
WithConnectTimeout(d) Bound a single connection attempt
WithWriteTimeout(d) Bound writing a control packet
WithMessageChannelDepth(n) Inbound message buffer size
WithCleanSession(bool) Request a clean MQTT session (default true)
WithTLSConfig(*tls.Config) Override the TLS configuration

SessionID() returns the MQTT client id used by the client. It is stable for the lifetime of the Client.

Reconnection and limits

Webull does not restore subscriptions after a connection is lost. With WithAutoReconnect, the client detects the drop, reconnects, and re-issues the HTTP subscribe calls for every active subscription before OnConnect fires, so handlers observe a restored session. Re-subscription is idempotent. Reconnecting reports an in-progress reconnect attempt and IsConnected reports the live state.

Webull applies the following rules:

  • An App Key supports at most 5 concurrent MQTT connections. Exceeding the limit fails with Webull error code 105; Connect returns an error explaining the limit. After a disconnect the server retains state for about one minute, so wait roughly a minute before reconnecting.
  • A new connection that reuses an existing session id disconnects the previous one. New generates a unique session id by default; only set WithSessionID when exclusivity is guaranteed.
  • The server pushes at most three messages per second per connection.

Topics

Topic constant Value Encoding
TopicQuote quote protobuf (Quote)
TopicSnapshot snapshot protobuf (Snapshot)
TopicTick tick protobuf (Tick)
TopicNotice notice JSON
TopicEcho echo heartbeat, ignored

Full example

The runnable example in examples/streaming connects over WebSocket, subscribes to AAPL quote/snapshot/tick pushes, prints messages, and handles Ctrl+C.