nautilus_binance/futures/websocket/streams/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Binance Futures WebSocket client for JSON market data streams.
17//!
18//! ## Connection Details
19//!
20//! - USD-M Endpoint: `wss://fstream.binance.com/market/ws`
21//! - COIN-M Endpoint: `wss://dstream.binance.com/ws`
22//! - Max streams: 200 per connection
23//! - Max connections: 20 per pool (up to 4,000 total streams)
24//! - Connection validity: 24 hours
25//! - Ping/pong: Every 3 minutes
26
27use std::{
28    fmt::Debug,
29    sync::{
30        Arc, Mutex,
31        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
32    },
33};
34
35use futures_util::Stream;
36use nautilus_common::live::get_runtime;
37use nautilus_core::{AtomicMap, string::secret::REDACTED};
38use nautilus_model::instruments::{Instrument, InstrumentAny};
39use nautilus_network::{
40    mode::ConnectionMode,
41    websocket::{
42        PingHandler, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
43        channel_message_handler,
44    },
45};
46use tokio_tungstenite::tungstenite::Message;
47use tokio_util::sync::CancellationToken;
48use ustr::Ustr;
49
50use super::{
51    error::{BinanceWsError, BinanceWsResult},
52    handler::BinanceFuturesDataWsFeedHandler,
53    messages::{BinanceFuturesWsStreamsCommand, BinanceFuturesWsStreamsMessage},
54};
55use crate::common::{
56    consts::{
57        BINANCE_API_KEY_HEADER, BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION, BINANCE_WS_CONNECTION_QUOTA,
58        BINANCE_WS_SUBSCRIPTION_QUOTA,
59    },
60    credential::SigningCredential,
61    enums::{BinanceEnvironment, BinanceProductType},
62    urls::get_ws_base_url,
63};
64
65/// Maximum streams per WebSocket connection for Futures.
66pub const MAX_STREAMS_PER_CONNECTION: usize = 200;
67
68/// Maximum connections per pool.
69const MAX_CONNECTIONS: usize = 20;
70
71// State for a single WebSocket connection within the pool
72struct ConnectionSlot {
73    cmd_tx: tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsStreamsCommand>,
74    streams: Vec<String>,
75    subscriptions_state: SubscriptionState,
76    handler_task: tokio::task::JoinHandle<()>,
77    bytes_task: tokio::task::JoinHandle<()>,
78    cancellation_token: CancellationToken,
79    connection_mode: Arc<AtomicU8>,
80}
81
82/// Binance Futures WebSocket client for JSON market data streams.
83///
84/// Manages a pool of up to 20 connections, each supporting up to 200 streams.
85/// New connections are created automatically when subscribing exceeds the current
86/// connection's stream limit. All connections feed into a single output stream,
87/// transparent to the data client.
88#[derive(Clone)]
89pub struct BinanceFuturesWebSocketClient {
90    url: String,
91    product_type: BinanceProductType,
92    credential: Option<Arc<SigningCredential>>,
93    heartbeat: Option<u64>,
94    signal: Arc<AtomicBool>,
95    slots: Arc<Mutex<Vec<ConnectionSlot>>>,
96    connect_lock: Arc<tokio::sync::Mutex<()>>,
97    out_tx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsStreamsMessage>>>>,
98    out_rx:
99        Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsStreamsMessage>>>>,
100    request_id_counter: Arc<AtomicU64>,
101    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
102    transport_backend: TransportBackend,
103}
104
105impl Debug for BinanceFuturesWebSocketClient {
106    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        f.debug_struct(stringify!(BinanceFuturesWebSocketClient))
108            .field("url", &self.url)
109            .field("product_type", &self.product_type)
110            .field("credential", &self.credential.as_ref().map(|_| REDACTED))
111            .field("heartbeat", &self.heartbeat)
112            .finish_non_exhaustive()
113    }
114}
115
116impl BinanceFuturesWebSocketClient {
117    /// Creates a new [`BinanceFuturesWebSocketClient`] instance.
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if:
122    /// - `product_type` is not a futures type (UsdM or CoinM).
123    /// - Credential creation fails.
124    pub fn new(
125        product_type: BinanceProductType,
126        environment: BinanceEnvironment,
127        api_key: Option<String>,
128        api_secret: Option<String>,
129        url_override: Option<String>,
130        heartbeat: Option<u64>,
131        transport_backend: TransportBackend,
132    ) -> anyhow::Result<Self> {
133        match product_type {
134            BinanceProductType::UsdM | BinanceProductType::CoinM => {}
135            _ => {
136                anyhow::bail!(
137                    "BinanceFuturesWebSocketClient requires UsdM or CoinM product type, was {product_type:?}"
138                );
139            }
140        }
141
142        let url =
143            url_override.unwrap_or_else(|| get_ws_base_url(product_type, environment).to_string());
144
145        let credential = match (api_key, api_secret) {
146            (Some(key), Some(secret)) => Some(Arc::new(SigningCredential::new(key, secret))),
147            _ => None,
148        };
149
150        Ok(Self {
151            url,
152            product_type,
153            credential,
154            heartbeat,
155            signal: Arc::new(AtomicBool::new(false)),
156            slots: Arc::new(Mutex::new(Vec::new())),
157            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
158            out_tx: Arc::new(Mutex::new(None)),
159            out_rx: Arc::new(Mutex::new(None)),
160            request_id_counter: Arc::new(AtomicU64::new(1)),
161            instruments_cache: Arc::new(AtomicMap::new()),
162            transport_backend,
163        })
164    }
165
166    /// Returns the product type (UsdM or CoinM).
167    #[must_use]
168    pub const fn product_type(&self) -> BinanceProductType {
169        self.product_type
170    }
171
172    /// Returns whether any connection in the pool is active.
173    #[must_use]
174    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
175    pub fn is_active(&self) -> bool {
176        let slots = self.slots.lock().expect("slots lock poisoned");
177        slots
178            .iter()
179            .any(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Active as u8)
180    }
181
182    /// Returns whether all connections in the pool are closed.
183    #[must_use]
184    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
185    pub fn is_closed(&self) -> bool {
186        let slots = self.slots.lock().expect("slots lock poisoned");
187        slots.is_empty()
188            || slots
189                .iter()
190                .all(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Closed as u8)
191    }
192
193    /// Returns the total number of confirmed subscriptions across all connections.
194    #[must_use]
195    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
196    pub fn subscription_count(&self) -> usize {
197        let slots = self.slots.lock().expect("slots lock poisoned");
198        slots.iter().map(|s| s.subscriptions_state.len()).sum()
199    }
200
201    /// Connects the first WebSocket connection in the pool.
202    ///
203    /// # Errors
204    ///
205    /// Returns an error if connection fails.
206    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
207    pub async fn connect(&mut self) -> BinanceWsResult<()> {
208        self.signal.store(false, Ordering::Relaxed);
209
210        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
211        *self.out_tx.lock().expect("out_tx lock poisoned") = Some(out_tx);
212        *self.out_rx.lock().expect("out_rx lock poisoned") = Some(out_rx);
213
214        let slot = self.create_connection().await?;
215        self.slots.lock().expect("slots lock poisoned").push(slot);
216
217        log::debug!(
218            "Connected to Binance Futures stream pool: url={}, product_type={:?}",
219            self.url,
220            self.product_type
221        );
222        Ok(())
223    }
224
225    /// Closes all WebSocket connections in the pool.
226    ///
227    /// # Errors
228    ///
229    /// Returns an error if disconnect fails.
230    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
231    pub async fn close(&mut self) -> BinanceWsResult<()> {
232        self.signal.store(true, Ordering::Relaxed);
233
234        let slots: Vec<ConnectionSlot> = {
235            let mut guard = self.slots.lock().expect("slots lock poisoned");
236            guard.drain(..).collect()
237        };
238
239        for slot in slots {
240            slot.cancellation_token.cancel();
241            let _ = slot.cmd_tx.send(BinanceFuturesWsStreamsCommand::Disconnect);
242            let _ = slot.handler_task.await;
243            slot.bytes_task.abort();
244        }
245
246        *self.out_tx.lock().expect("out_tx lock poisoned") = None;
247        *self.out_rx.lock().expect("out_rx lock poisoned") = None;
248
249        log::debug!("Disconnected from Binance Futures stream pool");
250        Ok(())
251    }
252
253    /// Subscribes to the specified streams.
254    ///
255    /// Streams are distributed across pool connections. New connections are created
256    /// automatically when existing ones reach the 200-stream limit, up to a maximum
257    /// of 20 connections.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the pool is exhausted or command delivery fails.
262    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
263    pub async fn subscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
264        // Serialize all phases so concurrent subscribers see a consistent
265        // pool state and can't trigger spurious `Pool exhausted`.
266        let _connect_guard = self.connect_lock.lock().await;
267
268        // Phase 1: filter already-subscribed streams (brief lock).
269        let new_streams: Vec<String> = {
270            let slots = self.slots.lock().expect("slots lock poisoned");
271            streams
272                .into_iter()
273                .filter(|s| !slots.iter().any(|slot| slot.streams.contains(s)))
274                .collect()
275        };
276
277        if new_streams.is_empty() {
278            return Ok(());
279        }
280
281        // Phase 2: create connections if needed.
282
283        loop {
284            let (remaining_capacity, slot_count) = {
285                let slots = self.slots.lock().expect("slots lock poisoned");
286                let cap: usize = slots
287                    .iter()
288                    .map(|s| MAX_STREAMS_PER_CONNECTION - s.streams.len())
289                    .sum();
290                (cap, slots.len())
291            };
292
293            if remaining_capacity >= new_streams.len() || slot_count >= MAX_CONNECTIONS {
294                break;
295            }
296
297            let new_slot = self.create_connection().await?;
298            let slot_count = {
299                let mut slots = self.slots.lock().expect("slots lock poisoned");
300                slots.push(new_slot);
301                slots.len()
302            };
303            log::debug!(
304                "Pool slot {} connected: url={}, product_type={:?}",
305                slot_count - 1,
306                self.url,
307                self.product_type
308            );
309        }
310
311        // Phase 3: assign streams to slots and send commands (brief lock).
312        // Stage assignments first so a capacity error leaves slots unchanged.
313        let mut slots = self.slots.lock().expect("slots lock poisoned");
314        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
315        let mut slot_counts: Vec<usize> = slots.iter().map(|s| s.streams.len()).collect();
316
317        for stream in &new_streams {
318            let slot_idx = slot_counts
319                .iter()
320                .position(|&count| count < MAX_STREAMS_PER_CONNECTION)
321                .ok_or_else(|| {
322                    let max_total = MAX_CONNECTIONS * MAX_STREAMS_PER_CONNECTION;
323                    BinanceWsError::ClientError(format!(
324                        "Pool exhausted: {max_total} total subscriptions \
325                         ({MAX_CONNECTIONS} connections x {MAX_STREAMS_PER_CONNECTION} streams)"
326                    ))
327                })?;
328
329            slot_counts[slot_idx] += 1;
330
331            if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
332                batch.1.push(stream.clone());
333            } else {
334                slot_batches.push((slot_idx, vec![stream.clone()]));
335            }
336        }
337
338        // Send commands first; only update slot state on success
339        for (slot_idx, batch) in &slot_batches {
340            slots[*slot_idx]
341                .cmd_tx
342                .send(BinanceFuturesWsStreamsCommand::Subscribe {
343                    streams: batch.clone(),
344                })
345                .map_err(|e| {
346                    BinanceWsError::ClientError(format!(
347                        "Handler not available for pool slot {slot_idx}: {e}"
348                    ))
349                })?;
350            slots[*slot_idx].streams.extend(batch.iter().cloned());
351        }
352
353        Ok(())
354    }
355
356    /// Unsubscribes from the specified streams.
357    ///
358    /// # Errors
359    ///
360    /// Returns an error if command delivery fails.
361    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
362    pub async fn unsubscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
363        let mut slots = self.slots.lock().expect("slots lock poisoned");
364        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
365
366        for stream in &streams {
367            if let Some(slot_idx) = slots.iter().position(|s| s.streams.contains(stream)) {
368                if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
369                    batch.1.push(stream.clone());
370                } else {
371                    slot_batches.push((slot_idx, vec![stream.clone()]));
372                }
373            }
374        }
375
376        // Send commands first; only update slot state on success
377        for (slot_idx, batch) in &slot_batches {
378            slots[*slot_idx]
379                .cmd_tx
380                .send(BinanceFuturesWsStreamsCommand::Unsubscribe {
381                    streams: batch.clone(),
382                })
383                .map_err(|e| {
384                    BinanceWsError::ClientError(format!(
385                        "Handler not available for pool slot {slot_idx}: {e}"
386                    ))
387                })?;
388
389            for stream in batch {
390                slots[*slot_idx].streams.retain(|s| s != stream);
391            }
392        }
393
394        Ok(())
395    }
396
397    /// Returns a stream of messages from all WebSocket connections.
398    ///
399    /// This method can only be called once per connection lifecycle. Subsequent calls
400    /// return an empty stream.
401    ///
402    /// # Panics
403    ///
404    /// Panics if the internal output receiver mutex is poisoned.
405    pub fn stream(&self) -> impl Stream<Item = BinanceFuturesWsStreamsMessage> + 'static {
406        let out_rx = self.out_rx.lock().expect("out_rx lock poisoned").take();
407        async_stream::stream! {
408            if let Some(mut rx) = out_rx {
409                while let Some(msg) = rx.recv().await {
410                    yield msg;
411                }
412            }
413        }
414    }
415
416    /// Bulk initialize the instrument cache.
417    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
418        self.instruments_cache.rcu(|m| {
419            for inst in instruments {
420                m.insert(inst.raw_symbol().inner(), inst.clone());
421            }
422        });
423    }
424
425    /// Update a single instrument in the cache.
426    pub fn cache_instrument(&self, instrument: InstrumentAny) {
427        self.instruments_cache
428            .insert(instrument.raw_symbol().inner(), instrument);
429    }
430
431    /// Returns a shared reference to the instruments cache.
432    #[must_use]
433    pub fn instruments_cache(&self) -> Arc<AtomicMap<Ustr, InstrumentAny>> {
434        self.instruments_cache.clone()
435    }
436
437    /// Returns an instrument from the cache by raw symbol.
438    #[must_use]
439    pub fn get_instrument(&self, symbol: &str) -> Option<InstrumentAny> {
440        self.instruments_cache.get_cloned(&Ustr::from(symbol))
441    }
442
443    async fn create_connection(&self) -> BinanceWsResult<ConnectionSlot> {
444        let out_tx = self
445            .out_tx
446            .lock()
447            .expect("out_tx lock poisoned")
448            .clone()
449            .ok_or_else(|| {
450                BinanceWsError::ClientError("Output channel not initialized".to_string())
451            })?;
452
453        let (raw_handler, raw_rx) = channel_message_handler();
454        let ping_handler: PingHandler = Arc::new(move |_| {});
455
456        let headers = if let Some(ref cred) = self.credential {
457            vec![(
458                BINANCE_API_KEY_HEADER.to_string(),
459                cred.api_key().to_string(),
460            )]
461        } else {
462            vec![]
463        };
464
465        let config = WebSocketConfig {
466            url: self.url.clone(),
467            headers,
468            heartbeat: self.heartbeat,
469            heartbeat_msg: None,
470            reconnect_timeout_ms: Some(5_000),
471            reconnect_delay_initial_ms: Some(500),
472            reconnect_delay_max_ms: Some(5_000),
473            reconnect_backoff_factor: Some(2.0),
474            reconnect_jitter_ms: Some(250),
475            reconnect_max_attempts: None,
476            idle_timeout_ms: None,
477            backend: self.transport_backend,
478            proxy_url: None,
479        };
480
481        let keyed_quotas = vec![(
482            BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION[0].as_str().to_string(),
483            *BINANCE_WS_SUBSCRIPTION_QUOTA,
484        )];
485
486        let client = WebSocketClient::connect(
487            config,
488            Some(raw_handler),
489            Some(ping_handler),
490            None,
491            keyed_quotas,
492            Some(*BINANCE_WS_CONNECTION_QUOTA),
493        )
494        .await
495        .map_err(|e| BinanceWsError::NetworkError(e.to_string()))?;
496
497        let connection_mode = client.connection_mode_atomic();
498        let subscriptions_state = SubscriptionState::new('@');
499        let cancellation_token = CancellationToken::new();
500
501        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
502
503        // Convert raw Message frames to Vec<u8> for the JSON handler
504        let (bytes_tx, bytes_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
505
506        let bytes_task = get_runtime().spawn(async move {
507            let mut raw_rx = raw_rx;
508            while let Some(msg) = raw_rx.recv().await {
509                let data = match msg {
510                    Message::Binary(data) => data.to_vec(),
511                    Message::Text(text) => text.as_bytes().to_vec(),
512                    Message::Close(_) => break,
513                    Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
514                };
515
516                if bytes_tx.send(data).is_err() {
517                    break;
518                }
519            }
520        });
521
522        let mut handler = BinanceFuturesDataWsFeedHandler::new(
523            self.signal.clone(),
524            cmd_rx,
525            bytes_rx,
526            out_tx.clone(),
527            subscriptions_state.clone(),
528            self.request_id_counter.clone(),
529        );
530
531        cmd_tx
532            .send(BinanceFuturesWsStreamsCommand::SetClient(client))
533            .map_err(|e| BinanceWsError::ClientError(format!("Failed to set client: {e}")))?;
534
535        let signal = self.signal.clone();
536        let token = cancellation_token.clone();
537        let subs = subscriptions_state.clone();
538        let resubscribe_tx = cmd_tx.clone();
539
540        let handler_task = get_runtime().spawn(async move {
541            loop {
542                tokio::select! {
543                    () = token.cancelled() => {
544                        log::debug!("Handler task cancelled");
545                        break;
546                    }
547                    result = handler.next() => {
548                        match result {
549                            Some(BinanceFuturesWsStreamsMessage::Reconnected) => {
550                                log::info!("WebSocket reconnected, restoring subscriptions");
551                                let all_topics = subs.all_topics();
552                                for topic in &all_topics {
553                                    subs.mark_failure(topic);
554                                }
555
556                                let streams = subs.all_topics();
557                                if !streams.is_empty()
558                                    && let Err(e) = resubscribe_tx.send(BinanceFuturesWsStreamsCommand::Subscribe { streams }) {
559                                        log::error!("Failed to resubscribe after reconnect: {e}");
560                                    }
561
562                                if out_tx.send(BinanceFuturesWsStreamsMessage::Reconnected).is_err() {
563                                    log::debug!("Output channel closed");
564                                    break;
565                                }
566                            }
567                            Some(msg) => {
568                                if out_tx.send(msg).is_err() {
569                                    log::debug!("Output channel closed");
570                                    break;
571                                }
572                            }
573                            None => {
574                                if signal.load(Ordering::Relaxed) {
575                                    log::debug!("Handler received shutdown signal");
576                                } else {
577                                    log::warn!("Handler loop ended unexpectedly");
578                                }
579                                break;
580                            }
581                        }
582                    }
583                }
584            }
585        });
586
587        Ok(ConnectionSlot {
588            cmd_tx,
589            streams: Vec::new(),
590            subscriptions_state,
591            handler_task,
592            bytes_task,
593            cancellation_token,
594            connection_mode,
595        })
596    }
597}
nautilus_binance/futures/websocket/streams/client.rs

nautilus_binance/futures/websocket/streams/
client.rs
