nautilus_binance/futures/websocket/
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/ws`
21//! - COIN-M Endpoint: `wss://dstream.binance.com/ws`
22//! - Max streams: 200 per connection
23//! - Connection validity: 24 hours
24//! - Ping/pong: Every 3 minutes
25
26use std::{
27    fmt::Debug,
28    sync::{
29        Arc,
30        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
31    },
32};
33
34use arc_swap::ArcSwap;
35use dashmap::DashMap;
36use futures_util::Stream;
37use nautilus_common::live::get_runtime;
38use nautilus_model::instruments::{Instrument, InstrumentAny};
39use nautilus_network::{
40    mode::ConnectionMode,
41    websocket::{
42        PingHandler, SubscriptionState, WebSocketClient, WebSocketConfig, channel_message_handler,
43    },
44};
45use tokio_tungstenite::tungstenite::Message;
46use tokio_util::sync::CancellationToken;
47use ustr::Ustr;
48
49use super::{
50    handler::BinanceFuturesWsFeedHandler,
51    messages::{BinanceFuturesHandlerCommand, NautilusFuturesWsMessage},
52};
53use crate::{
54    common::{
55        credential::Credential,
56        enums::{BinanceEnvironment, BinanceProductType},
57        urls::get_ws_base_url,
58    },
59    websocket::error::{BinanceWsError, BinanceWsResult},
60};
61
62/// Maximum streams per WebSocket connection for Futures.
63pub const MAX_STREAMS_PER_CONNECTION: usize = 200;
64
65/// Binance Futures WebSocket client for JSON market data streams.
66#[derive(Clone)]
67#[cfg_attr(
68    feature = "python",
69    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.binance")
70)]
71pub struct BinanceFuturesWebSocketClient {
72    url: String,
73    product_type: BinanceProductType,
74    credential: Option<Arc<Credential>>,
75    heartbeat: Option<u64>,
76    signal: Arc<AtomicBool>,
77    connection_mode: Arc<ArcSwap<AtomicU8>>,
78    cmd_tx:
79        Arc<tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<BinanceFuturesHandlerCommand>>>,
80    out_rx: Arc<
81        std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<NautilusFuturesWsMessage>>>,
82    >,
83    task_handle: Option<Arc<tokio::task::JoinHandle<()>>>,
84    subscriptions_state: SubscriptionState,
85    request_id_counter: Arc<AtomicU64>,
86    instruments_cache: Arc<DashMap<Ustr, InstrumentAny>>,
87    cancellation_token: CancellationToken,
88}
89
90impl Debug for BinanceFuturesWebSocketClient {
91    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92        f.debug_struct(stringify!(BinanceFuturesWebSocketClient))
93            .field("url", &self.url)
94            .field("product_type", &self.product_type)
95            .field(
96                "credential",
97                &self.credential.as_ref().map(|_| "<redacted>"),
98            )
99            .field("heartbeat", &self.heartbeat)
100            .finish_non_exhaustive()
101    }
102}
103
104impl BinanceFuturesWebSocketClient {
105    /// Creates a new [`BinanceFuturesWebSocketClient`] instance.
106    ///
107    /// # Errors
108    ///
109    /// Returns an error if:
110    /// - `product_type` is not a futures type (UsdM or CoinM).
111    /// - Credential creation fails.
112    pub fn new(
113        product_type: BinanceProductType,
114        environment: BinanceEnvironment,
115        api_key: Option<String>,
116        api_secret: Option<String>,
117        url_override: Option<String>,
118        heartbeat: Option<u64>,
119    ) -> anyhow::Result<Self> {
120        match product_type {
121            BinanceProductType::UsdM | BinanceProductType::CoinM => {}
122            _ => {
123                anyhow::bail!(
124                    "BinanceFuturesWebSocketClient requires UsdM or CoinM product type, got {product_type:?}"
125                );
126            }
127        }
128
129        let url =
130            url_override.unwrap_or_else(|| get_ws_base_url(product_type, environment).to_string());
131
132        let credential = match (api_key, api_secret) {
133            (Some(key), Some(secret)) => Some(Arc::new(Credential::new(key, secret))),
134            _ => None,
135        };
136
137        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel();
138
139        Ok(Self {
140            url,
141            product_type,
142            credential,
143            heartbeat,
144            signal: Arc::new(AtomicBool::new(false)),
145            connection_mode: Arc::new(ArcSwap::new(Arc::new(AtomicU8::new(
146                ConnectionMode::Closed as u8,
147            )))),
148            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
149            out_rx: Arc::new(std::sync::Mutex::new(None)),
150            task_handle: None,
151            subscriptions_state: SubscriptionState::new('@'),
152            request_id_counter: Arc::new(AtomicU64::new(1)),
153            instruments_cache: Arc::new(DashMap::new()),
154            cancellation_token: CancellationToken::new(),
155        })
156    }
157
158    /// Returns the product type (UsdM or CoinM).
159    #[must_use]
160    pub const fn product_type(&self) -> BinanceProductType {
161        self.product_type
162    }
163
164    /// Returns whether the client is actively connected.
165    #[must_use]
166    pub fn is_active(&self) -> bool {
167        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
168        mode_u8 == ConnectionMode::Active as u8
169    }
170
171    /// Returns whether the client is closed.
172    #[must_use]
173    pub fn is_closed(&self) -> bool {
174        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
175        mode_u8 == ConnectionMode::Closed as u8
176    }
177
178    /// Returns the number of confirmed subscriptions.
179    #[must_use]
180    pub fn subscription_count(&self) -> usize {
181        self.subscriptions_state.len()
182    }
183
184    /// Connects to the WebSocket server.
185    ///
186    /// # Errors
187    ///
188    /// Returns an error if connection fails.
189    ///
190    /// # Panics
191    ///
192    /// Panics if the internal output receiver mutex is poisoned.
193    pub async fn connect(&mut self) -> BinanceWsResult<()> {
194        self.signal.store(false, Ordering::Relaxed);
195
196        let (raw_handler, raw_rx) = channel_message_handler();
197        let ping_handler: PingHandler = Arc::new(move |_| {});
198
199        // Build headers for HMAC authentication (if needed for user data streams)
200        let headers = if let Some(ref cred) = self.credential {
201            vec![("X-MBX-APIKEY".to_string(), cred.api_key().to_string())]
202        } else {
203            vec![]
204        };
205
206        let config = WebSocketConfig {
207            url: self.url.clone(),
208            headers,
209            heartbeat: self.heartbeat,
210            heartbeat_msg: None,
211            reconnect_timeout_ms: Some(5_000),
212            reconnect_delay_initial_ms: Some(500),
213            reconnect_delay_max_ms: Some(5_000),
214            reconnect_backoff_factor: Some(2.0),
215            reconnect_jitter_ms: Some(250),
216            reconnect_max_attempts: None,
217        };
218
219        let client = WebSocketClient::connect(
220            config,
221            Some(raw_handler),
222            Some(ping_handler),
223            None,
224            vec![],
225            None,
226        )
227        .await
228        .map_err(|e| BinanceWsError::NetworkError(e.to_string()))?;
229
230        self.connection_mode.store(client.connection_mode_atomic());
231
232        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
233        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
234        *self.cmd_tx.write().await = cmd_tx;
235        *self.out_rx.lock().expect("out_rx lock poisoned") = Some(out_rx);
236
237        let (bytes_tx, bytes_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
238
239        let bytes_task = get_runtime().spawn(async move {
240            let mut raw_rx = raw_rx;
241            while let Some(msg) = raw_rx.recv().await {
242                let data = match msg {
243                    Message::Binary(data) => data.to_vec(),
244                    Message::Text(text) => text.as_bytes().to_vec(),
245                    Message::Close(_) => break,
246                    Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
247                };
248                if bytes_tx.send(data).is_err() {
249                    break;
250                }
251            }
252        });
253
254        let mut handler = BinanceFuturesWsFeedHandler::new(
255            self.signal.clone(),
256            cmd_rx,
257            bytes_rx,
258            out_tx.clone(),
259            self.subscriptions_state.clone(),
260            self.request_id_counter.clone(),
261        );
262
263        self.cmd_tx
264            .read()
265            .await
266            .send(BinanceFuturesHandlerCommand::SetClient(client))
267            .map_err(|e| BinanceWsError::ClientError(format!("Failed to set client: {e}")))?;
268
269        let instruments: Vec<InstrumentAny> = self
270            .instruments_cache
271            .iter()
272            .map(|entry| entry.value().clone())
273            .collect();
274
275        if !instruments.is_empty() {
276            self.cmd_tx
277                .read()
278                .await
279                .send(BinanceFuturesHandlerCommand::InitializeInstruments(
280                    instruments,
281                ))
282                .map_err(|e| {
283                    BinanceWsError::ClientError(format!("Failed to initialize instruments: {e}"))
284                })?;
285        }
286
287        let signal = self.signal.clone();
288        let cancellation_token = self.cancellation_token.clone();
289        let subscriptions_state = self.subscriptions_state.clone();
290        let cmd_tx = self.cmd_tx.clone();
291
292        let task_handle = get_runtime().spawn(async move {
293            loop {
294                tokio::select! {
295                    _ = cancellation_token.cancelled() => {
296                        tracing::debug!("Handler task cancelled");
297                        break;
298                    }
299                    result = handler.next() => {
300                        match result {
301                            Some(NautilusFuturesWsMessage::Reconnected) => {
302                                tracing::info!("WebSocket reconnected, restoring subscriptions");
303                                // Mark all confirmed subscriptions as pending
304                                let all_topics = subscriptions_state.all_topics();
305                                for topic in &all_topics {
306                                    subscriptions_state.mark_failure(topic);
307                                }
308
309                                // Resubscribe using tracked subscription state
310                                let streams = subscriptions_state.all_topics();
311                                if !streams.is_empty()
312                                    && let Err(e) = cmd_tx.read().await.send(BinanceFuturesHandlerCommand::Subscribe { streams }) {
313                                        tracing::error!(error = %e, "Failed to resubscribe after reconnect");
314                                    }
315
316                                if out_tx.send(NautilusFuturesWsMessage::Reconnected).is_err() {
317                                    tracing::debug!("Output channel closed");
318                                    break;
319                                }
320                            }
321                            Some(msg) => {
322                                if out_tx.send(msg).is_err() {
323                                    tracing::debug!("Output channel closed");
324                                    break;
325                                }
326                            }
327                            None => {
328                                if signal.load(Ordering::Relaxed) {
329                                    tracing::debug!("Handler received shutdown signal");
330                                } else {
331                                    tracing::warn!("Handler loop ended unexpectedly");
332                                }
333                                break;
334                            }
335                        }
336                    }
337                }
338            }
339            bytes_task.abort();
340        });
341
342        self.task_handle = Some(Arc::new(task_handle));
343
344        tracing::info!(url = %self.url, product_type = ?self.product_type, "Connected to Binance Futures stream");
345        Ok(())
346    }
347
348    /// Closes the WebSocket connection.
349    ///
350    /// # Errors
351    ///
352    /// Returns an error if disconnect fails.
353    ///
354    /// # Panics
355    ///
356    /// Panics if the internal output receiver mutex is poisoned.
357    pub async fn close(&mut self) -> BinanceWsResult<()> {
358        self.signal.store(true, Ordering::Relaxed);
359        self.cancellation_token.cancel();
360
361        let _ = self
362            .cmd_tx
363            .read()
364            .await
365            .send(BinanceFuturesHandlerCommand::Disconnect);
366
367        if let Some(handle) = self.task_handle.take()
368            && let Ok(handle) = Arc::try_unwrap(handle)
369        {
370            let _ = handle.await;
371        }
372
373        *self.out_rx.lock().expect("out_rx lock poisoned") = None;
374
375        tracing::info!("Disconnected from Binance Futures stream");
376        Ok(())
377    }
378
379    /// Subscribes to the specified streams.
380    ///
381    /// # Errors
382    ///
383    /// Returns an error if subscription fails or would exceed stream limit.
384    pub async fn subscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
385        let current_count = self.subscriptions_state.len();
386        if current_count + streams.len() > MAX_STREAMS_PER_CONNECTION {
387            return Err(BinanceWsError::ClientError(format!(
388                "Would exceed max streams: {} + {} > {}",
389                current_count,
390                streams.len(),
391                MAX_STREAMS_PER_CONNECTION
392            )));
393        }
394
395        self.cmd_tx
396            .read()
397            .await
398            .send(BinanceFuturesHandlerCommand::Subscribe { streams })
399            .map_err(|e| BinanceWsError::ClientError(format!("Handler not available: {e}")))?;
400
401        Ok(())
402    }
403
404    /// Unsubscribes from the specified streams.
405    ///
406    /// # Errors
407    ///
408    /// Returns an error if unsubscription fails.
409    pub async fn unsubscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
410        self.cmd_tx
411            .read()
412            .await
413            .send(BinanceFuturesHandlerCommand::Unsubscribe { streams })
414            .map_err(|e| BinanceWsError::ClientError(format!("Handler not available: {e}")))?;
415
416        Ok(())
417    }
418
419    /// Returns a stream of messages from the WebSocket.
420    ///
421    /// This method can only be called once per connection. Subsequent calls
422    /// will return an empty stream. If you need to consume messages from
423    /// multiple tasks, clone the client before connecting.
424    ///
425    /// # Panics
426    ///
427    /// Panics if the internal output receiver mutex is poisoned.
428    pub fn stream(&self) -> impl Stream<Item = NautilusFuturesWsMessage> + 'static {
429        let out_rx = self.out_rx.lock().expect("out_rx lock poisoned").take();
430        async_stream::stream! {
431            if let Some(mut rx) = out_rx {
432                while let Some(msg) = rx.recv().await {
433                    yield msg;
434                }
435            }
436        }
437    }
438
439    /// Bulk initialize the instrument cache.
440    ///
441    /// Instruments are cached by their raw symbol (e.g., "BTCUSDT") to match
442    /// the symbol format sent in WebSocket messages.
443    pub fn cache_instruments(&self, instruments: Vec<InstrumentAny>) {
444        for inst in &instruments {
445            self.instruments_cache
446                .insert(inst.raw_symbol().inner(), inst.clone());
447        }
448
449        if self.is_active() {
450            let cmd_tx = self.cmd_tx.clone();
451            let instruments_clone = instruments;
452            get_runtime().spawn(async move {
453                let _ =
454                    cmd_tx
455                        .read()
456                        .await
457                        .send(BinanceFuturesHandlerCommand::InitializeInstruments(
458                            instruments_clone,
459                        ));
460            });
461        }
462    }
463
464    /// Update a single instrument in the cache.
465    ///
466    /// Instruments are cached by their raw symbol (e.g., "BTCUSDT") to match
467    /// the symbol format sent in WebSocket messages.
468    pub fn cache_instrument(&self, instrument: InstrumentAny) {
469        self.instruments_cache
470            .insert(instrument.raw_symbol().inner(), instrument.clone());
471
472        if self.is_active() {
473            let cmd_tx = self.cmd_tx.clone();
474            get_runtime().spawn(async move {
475                let _ = cmd_tx
476                    .read()
477                    .await
478                    .send(BinanceFuturesHandlerCommand::UpdateInstrument(instrument));
479            });
480        }
481    }
482
483    /// Get an instrument from the cache.
484    #[must_use]
485    pub fn get_instrument(&self, symbol: &str) -> Option<InstrumentAny> {
486        self.instruments_cache
487            .get(&Ustr::from(symbol))
488            .map(|entry| entry.value().clone())
489    }
490}
nautilus_binance/futures/websocket/client.rs

nautilus_binance/futures/websocket/
client.rs
