nautilus_binance/futures/websocket/trading/
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 Trading API client.
17//!
18//! ## Connection details
19//!
20//! - Endpoint: `ws-fapi.binance.com/ws-fapi/v1` (USD-M only)
21//! - Authentication: HMAC-SHA256 signature per request
22//! - JSON request/response pattern
23//! - Connection validity: 24 hours
24//! - Ping/pong: every 20 seconds
25
26use std::{
27    fmt::Debug,
28    num::NonZeroU32,
29    sync::{
30        Arc, LazyLock,
31        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
32    },
33    time::Duration,
34};
35
36use arc_swap::ArcSwap;
37use nautilus_core::string::secret::{REDACTED, SecretString};
38use nautilus_live::{SocketControl, task::TaskGroup};
39use nautilus_network::{
40    http::create_standard_nautilus_headers,
41    mode::ConnectionMode,
42    ratelimiter::quota::Quota,
43    websocket::{
44        PingHandler, TransportBackend, WebSocketClient, WebSocketConfig, channel_message_handler,
45    },
46};
47use parking_lot::Mutex;
48use tokio_util::sync::CancellationToken;
49use ustr::Ustr;
50
51use super::{
52    error::{BinanceFuturesWsApiError, BinanceFuturesWsApiResult},
53    handler::BinanceFuturesWsTradingHandler,
54    messages::{BinanceFuturesWsTradingCommand, BinanceFuturesWsTradingMessage},
55};
56use crate::{
57    common::{
58        consts::{BINANCE_API_KEY_HEADER, BINANCE_FUTURES_USD_WS_API_URL},
59        credential::SigningCredential,
60    },
61    futures::http::query::{
62        BinanceCancelOrderParams, BinanceModifyOrderParams, BinanceNewOrderParams,
63    },
64};
65
66/// Pre-interned rate limit key for futures order operations (place/cancel/modify).
67///
68/// Binance Futures WebSocket API: 1200 requests per minute per IP (20/sec).
69pub static BINANCE_FUTURES_WS_RATE_LIMIT_KEY_ORDER: LazyLock<[Ustr; 1]> =
70    LazyLock::new(|| [Ustr::from("futures_order")]);
71
72/// Returns the Binance Futures WebSocket API order rate limit quota (1200 per minute).
73// Constant values are provably valid
74#[expect(clippy::missing_panics_doc)]
75#[must_use]
76pub fn binance_futures_ws_order_quota() -> Quota {
77    Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant")
78}
79
80/// Binance Futures WebSocket Trading API client.
81///
82/// Provides order management via WebSocket with JSON responses,
83/// complementing the HTTP client with lower-latency order submission.
84/// Only available for USD-M Futures.
85#[derive(Clone)]
86pub struct BinanceFuturesWsTradingClient {
87    url: String,
88    credential: Arc<SigningCredential>,
89    heartbeat: Option<u64>,
90    signal: Arc<AtomicBool>,
91    connection_mode: Arc<ArcSwap<AtomicU8>>,
92    cmd_tx: Arc<
93        tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsTradingCommand>>,
94    >,
95    out_rx:
96        Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsTradingMessage>>>>,
97    handler_tasks: Arc<TaskGroup>,
98    connect_lock: Arc<tokio::sync::Mutex<()>>,
99    request_id_counter: Arc<AtomicU64>,
100    cancellation_token: Arc<Mutex<CancellationToken>>,
101    transport_backend: TransportBackend,
102    proxy_url: Option<SecretString>,
103    recv_window_ms: Option<u64>,
104    socket_control: Option<SocketControl>,
105}
106
107impl Debug for BinanceFuturesWsTradingClient {
108    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109        f.debug_struct(stringify!(BinanceFuturesWsTradingClient))
110            .field("url", &self.url)
111            .field("credential", &REDACTED)
112            .field("heartbeat", &self.heartbeat)
113            .finish_non_exhaustive()
114    }
115}
116
117impl BinanceFuturesWsTradingClient {
118    /// Creates a new [`BinanceFuturesWsTradingClient`] instance.
119    #[must_use]
120    pub fn new(
121        url: Option<String>,
122        api_key: String,
123        api_secret: String,
124        heartbeat: Option<u64>,
125        transport_backend: TransportBackend,
126    ) -> Self {
127        let url = url.unwrap_or_else(|| BINANCE_FUTURES_USD_WS_API_URL.to_string());
128        let credential = Arc::new(SigningCredential::new(api_key, api_secret));
129
130        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel();
131
132        Self {
133            url,
134            credential,
135            heartbeat,
136            signal: Arc::new(AtomicBool::new(false)),
137            connection_mode: Arc::new(ArcSwap::new(Arc::new(AtomicU8::new(
138                ConnectionMode::Closed as u8,
139            )))),
140            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
141            out_rx: Arc::new(Mutex::new(None)),
142            handler_tasks: Arc::new(TaskGroup::new()),
143            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
144            request_id_counter: Arc::new(AtomicU64::new(1)),
145            cancellation_token: Arc::new(Mutex::new(CancellationToken::new())),
146            transport_backend,
147            proxy_url: None,
148            recv_window_ms: None,
149            socket_control: None,
150        }
151    }
152
153    /// Configures the proxy used by the WebSocket connection.
154    #[must_use]
155    pub fn with_proxy(mut self, proxy_url: Option<String>) -> Self {
156        self.proxy_url = proxy_url.map(SecretString::from);
157        self
158    }
159
160    /// Configures socket state reporting and reconnect control.
161    #[must_use]
162    pub fn with_socket_control(mut self, control: SocketControl) -> Self {
163        self.socket_control = Some(control);
164        self
165    }
166
167    /// Configures the receive window added to signed WebSocket API requests.
168    #[must_use]
169    pub const fn with_recv_window(mut self, recv_window_ms: Option<u64>) -> Self {
170        self.recv_window_ms = recv_window_ms;
171        self
172    }
173
174    /// Returns whether the client is actively connected.
175    #[must_use]
176    pub fn is_active(&self) -> bool {
177        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
178        mode_u8 == ConnectionMode::Active as u8
179    }
180
181    /// Returns whether the client is closed.
182    #[must_use]
183    pub fn is_closed(&self) -> bool {
184        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
185        mode_u8 == ConnectionMode::Closed as u8
186    }
187
188    pub fn next_request_id(&self) -> String {
189        let id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
190        format!("req-{id}")
191    }
192
193    /// Connects to the WebSocket Trading API server.
194    ///
195    /// # Errors
196    ///
197    /// Returns an error if connection fails.
198    pub async fn connect(&mut self) -> BinanceFuturesWsApiResult<()> {
199        let connect_lock = Arc::clone(&self.connect_lock);
200        let _connect_guard = connect_lock.lock().await;
201
202        if !self.handler_tasks.is_open() || !self.handler_tasks.is_empty() {
203            self.disconnect_handler().await?;
204            self.handler_tasks.start_generation().map_err(|e| {
205                BinanceFuturesWsApiError::ClientError(format!(
206                    "failed to start WebSocket handler task generation: {e}"
207                ))
208            })?;
209        }
210        let handler_spawner = self.handler_tasks.spawner().map_err(|e| {
211            BinanceFuturesWsApiError::ClientError(format!(
212                "failed to acquire WebSocket handler task spawner: {e}"
213            ))
214        })?;
215        self.signal.store(false, Ordering::Relaxed);
216        *self.cancellation_token.lock() = CancellationToken::new();
217
218        let (raw_handler, raw_rx) = channel_message_handler();
219        let ping_handler: PingHandler = Arc::new(move |_| {});
220
221        let mut headers = create_standard_nautilus_headers();
222        headers.push((
223            BINANCE_API_KEY_HEADER.to_string(),
224            self.credential.api_key().to_string(),
225        ));
226
227        let config = WebSocketConfig {
228            url: self.url.clone(),
229            headers,
230            heartbeat_interval_secs: self.heartbeat,
231            heartbeat_payload: None,
232            connect_timeout_ms: Some(5_000),
233            reconnect_delay_initial_ms: Some(500),
234            reconnect_delay_max_ms: Some(5_000),
235            reconnect_backoff_factor: Some(2.0),
236            reconnect_jitter_ms: Some(250),
237            reconnect_max_attempts: None,
238            heartbeat_timeout_secs: None,
239            idle_timeout_ms: None,
240            writer_capacity: None,
241            backend: self.transport_backend,
242            proxy_url: self
243                .proxy_url
244                .as_ref()
245                .map(|value| value.expose_secret().to_owned()),
246            max_message_size_bytes: None,
247            max_frame_size_bytes: None,
248        };
249
250        let keyed_quotas = vec![(
251            BINANCE_FUTURES_WS_RATE_LIMIT_KEY_ORDER[0].to_string(),
252            binance_futures_ws_order_quota(),
253        )];
254
255        let client = WebSocketClient::builder()
256            .config(config)
257            .message_handler(raw_handler)
258            .ping_handler(ping_handler)
259            .keyed_quotas(keyed_quotas)
260            .default_quota(binance_futures_ws_order_quota())
261            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
262            .connect()
263            .await
264            .map_err(|e| BinanceFuturesWsApiError::ConnectionError(e.to_string()))?;
265
266        let connection_mode = client.connection_mode_atomic();
267        let reconnect_handle = client.reconnect_handle();
268
269        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
270        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
271
272        cmd_tx
273            .send(BinanceFuturesWsTradingCommand::SetClient(client))
274            .map_err(|e| BinanceFuturesWsApiError::HandlerUnavailable(e.to_string()))?;
275
276        {
277            let mut rx_guard = self.out_rx.lock();
278            *rx_guard = Some(out_rx);
279        }
280
281        {
282            let mut tx_guard = self.cmd_tx.write().await;
283            *tx_guard = cmd_tx;
284            self.connection_mode.store(connection_mode);
285        }
286
287        let signal = self.signal.clone();
288        let credential = self.credential.clone();
289        let mut handler =
290            BinanceFuturesWsTradingHandler::new(signal, cmd_rx, raw_rx, out_tx, credential)
291                .with_recv_window(self.recv_window_ms);
292
293        if let Some(control) = &self.socket_control {
294            control.register(move || reconnect_handle.request_reconnect());
295        }
296
297        let cancellation_token = self.cancellation_token.lock().clone();
298
299        let handler_task = async move {
300            tokio::select! {
301                () = cancellation_token.cancelled() => {
302                    log::debug!("Handler task cancelled");
303                }
304                _ = handler.run() => {
305                    log::debug!("Handler run completed");
306                }
307            }
308        };
309
310        if let Err(e) = handler_spawner.spawn(handler_task) {
311            if let Some(control) = &self.socket_control {
312                control.deregister();
313            }
314            self.out_rx.lock().take();
315            return Err(BinanceFuturesWsApiError::HandlerUnavailable(format!(
316                "failed to register handler task: {e}"
317            )));
318        }
319
320        Ok(())
321    }
322
323    /// Disconnects from the WebSocket Trading API server.
324    ///
325    /// # Errors
326    ///
327    /// Returns an error if the handler task fails or does not stop after abort.
328    pub async fn disconnect(&mut self) -> BinanceFuturesWsApiResult<()> {
329        let connect_lock = Arc::clone(&self.connect_lock);
330        let _connect_guard = connect_lock.lock().await;
331
332        self.disconnect_handler().await
333    }
334
335    pub(crate) fn begin_shutdown(&self) {
336        self.handler_tasks.begin_shutdown();
337        self.signal.store(true, Ordering::Relaxed);
338        self.cancellation_token.lock().cancel();
339    }
340
341    async fn disconnect_handler(&self) -> BinanceFuturesWsApiResult<()> {
342        self.handler_tasks.begin_shutdown();
343        self.signal.store(true, Ordering::Relaxed);
344
345        if let Err(e) = self
346            .cmd_tx
347            .read()
348            .await
349            .send(BinanceFuturesWsTradingCommand::Disconnect)
350        {
351            log::debug!("Failed to send disconnect command: {e}");
352        }
353
354        self.cancellation_token.lock().cancel();
355
356        let result = self
357            .handler_tasks
358            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
359            .await
360            .map_err(|e| {
361                BinanceFuturesWsApiError::ClientError(format!("handler task shutdown failed: {e}"))
362            });
363
364        if let Some(control) = &self.socket_control {
365            control.deregister();
366        }
367        result
368    }
369
370    /// Places a new order via the WebSocket Trading API.
371    ///
372    /// # Errors
373    ///
374    /// Returns an error if the handler is unavailable.
375    pub async fn place_order(
376        &self,
377        params: BinanceNewOrderParams,
378    ) -> BinanceFuturesWsApiResult<String> {
379        let id = self.next_request_id();
380        self.place_order_with_id(id.clone(), params).await?;
381        Ok(id)
382    }
383
384    /// Places a new order via the WebSocket Trading API using a pre-generated request ID.
385    ///
386    /// # Errors
387    ///
388    /// Returns an error if the handler is unavailable.
389    pub async fn place_order_with_id(
390        &self,
391        id: String,
392        params: BinanceNewOrderParams,
393    ) -> BinanceFuturesWsApiResult<()> {
394        let cmd = BinanceFuturesWsTradingCommand::PlaceOrder { id, params };
395        self.send_cmd(cmd).await
396    }
397
398    /// Cancels an order via the WebSocket Trading API.
399    ///
400    /// # Errors
401    ///
402    /// Returns an error if the handler is unavailable.
403    pub async fn cancel_order(
404        &self,
405        params: BinanceCancelOrderParams,
406    ) -> BinanceFuturesWsApiResult<String> {
407        let id = self.next_request_id();
408        self.cancel_order_with_id(id.clone(), params).await?;
409        Ok(id)
410    }
411
412    /// Cancels an order via the WebSocket Trading API using a pre-generated request ID.
413    ///
414    /// # Errors
415    ///
416    /// Returns an error if the handler is unavailable.
417    pub async fn cancel_order_with_id(
418        &self,
419        id: String,
420        params: BinanceCancelOrderParams,
421    ) -> BinanceFuturesWsApiResult<()> {
422        let cmd = BinanceFuturesWsTradingCommand::CancelOrder { id, params };
423        self.send_cmd(cmd).await
424    }
425
426    /// Modifies an order via the WebSocket Trading API (in-place amendment).
427    ///
428    /// # Errors
429    ///
430    /// Returns an error if the handler is unavailable.
431    pub async fn modify_order(
432        &self,
433        params: BinanceModifyOrderParams,
434    ) -> BinanceFuturesWsApiResult<String> {
435        let id = self.next_request_id();
436        self.modify_order_with_id(id.clone(), params).await?;
437        Ok(id)
438    }
439
440    /// Modifies an order via the WebSocket Trading API using a pre-generated request ID.
441    ///
442    /// # Errors
443    ///
444    /// Returns an error if the handler is unavailable.
445    pub async fn modify_order_with_id(
446        &self,
447        id: String,
448        params: BinanceModifyOrderParams,
449    ) -> BinanceFuturesWsApiResult<()> {
450        let cmd = BinanceFuturesWsTradingCommand::ModifyOrder { id, params };
451        self.send_cmd(cmd).await
452    }
453
454    /// Receives the next message from the handler.
455    ///
456    /// Returns `None` if the receiver is closed or not initialized.
457    pub async fn recv(&self) -> Option<BinanceFuturesWsTradingMessage> {
458        let rx_opt = {
459            let mut rx_guard = self.out_rx.lock();
460            rx_guard.take()
461        };
462
463        if let Some(mut rx) = rx_opt {
464            let result = rx.recv().await;
465
466            let mut rx_guard = self.out_rx.lock();
467            *rx_guard = Some(rx);
468            result
469        } else {
470            None
471        }
472    }
473
474    async fn send_cmd(&self, cmd: BinanceFuturesWsTradingCommand) -> BinanceFuturesWsApiResult<()> {
475        self.cmd_tx
476            .read()
477            .await
478            .send(cmd)
479            .map_err(|e| BinanceFuturesWsApiError::HandlerUnavailable(e.to_string()))
480    }
481}
482
483#[cfg(test)]
484mod tests {
485    use futures_util::StreamExt;
486    use nautilus_testkit::connection::race_command_with_publication;
487    use rstest::rstest;
488
489    use super::*;
490
491    #[tokio::test]
492    async fn command_during_sender_publication_reaches_initialized_handler() {
493        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
494        let addr = listener.local_addr().unwrap();
495        let (received_tx, mut received_rx) = tokio::sync::mpsc::unbounded_channel();
496
497        let server = tokio::spawn(async move {
498            let (socket, _) = listener.accept().await.unwrap();
499            let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap();
500            while let Some(Ok(message)) = socket.next().await {
501                if let tokio_tungstenite::tungstenite::Message::Text(text) = message {
502                    let payload: serde_json::Value = serde_json::from_str(text.as_str()).unwrap();
503                    let _ = received_tx.send(payload);
504                }
505            }
506        });
507
508        let url = format!("ws://{addr}");
509
510        let mut ws_client = BinanceFuturesWsTradingClient::new(
511            Some(url),
512            "startup-key".to_string(),
513            "startup-secret".to_string(),
514            None,
515            TransportBackend::default(),
516        );
517        let observer = ws_client.clone();
518
519        let command = BinanceFuturesWsTradingCommand::CancelOrder {
520            id: "startup-74".to_string(),
521            params: BinanceCancelOrderParams {
522                symbol: "ETHUSDT".to_string(),
523                order_id: Some(725),
524                orig_client_order_id: None,
525                recv_window: None,
526            },
527        };
528
529        race_command_with_publication(
530            Arc::clone(&ws_client.cmd_tx),
531            ws_client.connect(),
532            move || command,
533            || observer.is_active(),
534        )
535        .await
536        .expect("connect failed");
537        let received = tokio::time::timeout(Duration::from_secs(2), received_rx.recv()).await;
538        ws_client.disconnect().await.expect("disconnect failed");
539        server.abort();
540
541        let mut payload = received
542            .expect("startup command must reach server")
543            .unwrap();
544        let timestamp = payload["params"]
545            .as_object_mut()
546            .unwrap()
547            .remove("timestamp")
548            .unwrap();
549        let signature = payload["params"]
550            .as_object_mut()
551            .unwrap()
552            .remove("signature")
553            .unwrap();
554        let timestamp = timestamp.as_u64().unwrap();
555        let query = format!("apiKey=startup-key&orderId=725&symbol=ETHUSDT&timestamp={timestamp}");
556        assert_eq!(
557            signature,
558            serde_json::json!(observer.credential.sign(&query))
559        );
560        assert_eq!(
561            payload,
562            serde_json::json!({
563                "id": "startup-74",
564                "method": "order.cancel",
565                "params": {
566                    "apiKey": "startup-key",
567                    "symbol": "ETHUSDT",
568                    "orderId": 725
569                }
570            })
571        );
572    }
573
574    #[rstest]
575    fn test_operational_options_are_preserved() {
576        let client = BinanceFuturesWsTradingClient::new(
577            None,
578            "api-key".to_string(),
579            "hmac-secret".to_string(),
580            None,
581            TransportBackend::default(),
582        )
583        .with_proxy(Some("http://proxy.example:8080".to_string()))
584        .with_recv_window(Some(30_000));
585
586        assert_eq!(
587            client.proxy_url.as_ref().map(SecretString::expose_secret),
588            Some("http://proxy.example:8080")
589        );
590        assert_eq!(client.recv_window_ms, Some(30_000));
591    }
592}
nautilus_binance/futures/websocket/trading/client.rs

nautilus_binance/futures/websocket/trading/
client.rs
