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            backend: self.transport_backend,
241            proxy_url: self
242                .proxy_url
243                .as_ref()
244                .map(|value| value.expose_secret().to_owned()),
245        };
246
247        let keyed_quotas = vec![(
248            BINANCE_FUTURES_WS_RATE_LIMIT_KEY_ORDER[0].to_string(),
249            binance_futures_ws_order_quota(),
250        )];
251
252        let client = WebSocketClient::builder()
253            .config(config)
254            .message_handler(raw_handler)
255            .ping_handler(ping_handler)
256            .keyed_quotas(keyed_quotas)
257            .default_quota(binance_futures_ws_order_quota())
258            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
259            .connect()
260            .await
261            .map_err(|e| BinanceFuturesWsApiError::ConnectionError(e.to_string()))?;
262
263        self.connection_mode.store(client.connection_mode_atomic());
264        let reconnect_handle = client.reconnect_handle();
265
266        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
267        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
268
269        cmd_tx
270            .send(BinanceFuturesWsTradingCommand::SetClient(client))
271            .map_err(|e| BinanceFuturesWsApiError::HandlerUnavailable(e.to_string()))?;
272
273        {
274            let mut rx_guard = self.out_rx.lock();
275            *rx_guard = Some(out_rx);
276        }
277
278        {
279            let mut tx_guard = self.cmd_tx.write().await;
280            *tx_guard = cmd_tx;
281        }
282
283        let signal = self.signal.clone();
284        let credential = self.credential.clone();
285        let mut handler =
286            BinanceFuturesWsTradingHandler::new(signal, cmd_rx, raw_rx, out_tx, credential)
287                .with_recv_window(self.recv_window_ms);
288
289        if let Some(control) = &self.socket_control {
290            control.register(move || reconnect_handle.request_reconnect());
291        }
292
293        let cancellation_token = self.cancellation_token.lock().clone();
294
295        let handler_task = async move {
296            tokio::select! {
297                () = cancellation_token.cancelled() => {
298                    log::debug!("Handler task cancelled");
299                }
300                _ = handler.run() => {
301                    log::debug!("Handler run completed");
302                }
303            }
304        };
305
306        if let Err(e) = handler_spawner.spawn(handler_task) {
307            if let Some(control) = &self.socket_control {
308                control.deregister();
309            }
310            self.out_rx.lock().take();
311            return Err(BinanceFuturesWsApiError::HandlerUnavailable(format!(
312                "failed to register handler task: {e}"
313            )));
314        }
315
316        Ok(())
317    }
318
319    /// Disconnects from the WebSocket Trading API server.
320    ///
321    /// # Errors
322    ///
323    /// Returns an error if the handler task fails or does not stop after abort.
324    pub async fn disconnect(&mut self) -> BinanceFuturesWsApiResult<()> {
325        let connect_lock = Arc::clone(&self.connect_lock);
326        let _connect_guard = connect_lock.lock().await;
327
328        self.disconnect_handler().await
329    }
330
331    pub(crate) fn begin_shutdown(&self) {
332        self.handler_tasks.begin_shutdown();
333        self.signal.store(true, Ordering::Relaxed);
334        self.cancellation_token.lock().cancel();
335    }
336
337    async fn disconnect_handler(&self) -> BinanceFuturesWsApiResult<()> {
338        self.handler_tasks.begin_shutdown();
339        self.signal.store(true, Ordering::Relaxed);
340
341        if let Err(e) = self
342            .cmd_tx
343            .read()
344            .await
345            .send(BinanceFuturesWsTradingCommand::Disconnect)
346        {
347            log::debug!("Failed to send disconnect command: {e}");
348        }
349
350        self.cancellation_token.lock().cancel();
351
352        let result = self
353            .handler_tasks
354            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
355            .await
356            .map_err(|e| {
357                BinanceFuturesWsApiError::ClientError(format!("handler task shutdown failed: {e}"))
358            });
359
360        if let Some(control) = &self.socket_control {
361            control.deregister();
362        }
363        result
364    }
365
366    /// Places a new order via the WebSocket Trading API.
367    ///
368    /// # Errors
369    ///
370    /// Returns an error if the handler is unavailable.
371    pub async fn place_order(
372        &self,
373        params: BinanceNewOrderParams,
374    ) -> BinanceFuturesWsApiResult<String> {
375        let id = self.next_request_id();
376        self.place_order_with_id(id.clone(), params).await?;
377        Ok(id)
378    }
379
380    /// Places a new order via the WebSocket Trading API using a pre-generated request ID.
381    ///
382    /// # Errors
383    ///
384    /// Returns an error if the handler is unavailable.
385    pub async fn place_order_with_id(
386        &self,
387        id: String,
388        params: BinanceNewOrderParams,
389    ) -> BinanceFuturesWsApiResult<()> {
390        let cmd = BinanceFuturesWsTradingCommand::PlaceOrder { id, params };
391        self.send_cmd(cmd).await
392    }
393
394    /// Cancels an order via the WebSocket Trading API.
395    ///
396    /// # Errors
397    ///
398    /// Returns an error if the handler is unavailable.
399    pub async fn cancel_order(
400        &self,
401        params: BinanceCancelOrderParams,
402    ) -> BinanceFuturesWsApiResult<String> {
403        let id = self.next_request_id();
404        self.cancel_order_with_id(id.clone(), params).await?;
405        Ok(id)
406    }
407
408    /// Cancels an order via the WebSocket Trading API using a pre-generated request ID.
409    ///
410    /// # Errors
411    ///
412    /// Returns an error if the handler is unavailable.
413    pub async fn cancel_order_with_id(
414        &self,
415        id: String,
416        params: BinanceCancelOrderParams,
417    ) -> BinanceFuturesWsApiResult<()> {
418        let cmd = BinanceFuturesWsTradingCommand::CancelOrder { id, params };
419        self.send_cmd(cmd).await
420    }
421
422    /// Modifies an order via the WebSocket Trading API (in-place amendment).
423    ///
424    /// # Errors
425    ///
426    /// Returns an error if the handler is unavailable.
427    pub async fn modify_order(
428        &self,
429        params: BinanceModifyOrderParams,
430    ) -> BinanceFuturesWsApiResult<String> {
431        let id = self.next_request_id();
432        self.modify_order_with_id(id.clone(), params).await?;
433        Ok(id)
434    }
435
436    /// Modifies an order via the WebSocket Trading API using a pre-generated request ID.
437    ///
438    /// # Errors
439    ///
440    /// Returns an error if the handler is unavailable.
441    pub async fn modify_order_with_id(
442        &self,
443        id: String,
444        params: BinanceModifyOrderParams,
445    ) -> BinanceFuturesWsApiResult<()> {
446        let cmd = BinanceFuturesWsTradingCommand::ModifyOrder { id, params };
447        self.send_cmd(cmd).await
448    }
449
450    /// Receives the next message from the handler.
451    ///
452    /// Returns `None` if the receiver is closed or not initialized.
453    pub async fn recv(&self) -> Option<BinanceFuturesWsTradingMessage> {
454        let rx_opt = {
455            let mut rx_guard = self.out_rx.lock();
456            rx_guard.take()
457        };
458
459        if let Some(mut rx) = rx_opt {
460            let result = rx.recv().await;
461
462            let mut rx_guard = self.out_rx.lock();
463            *rx_guard = Some(rx);
464            result
465        } else {
466            None
467        }
468    }
469
470    async fn send_cmd(&self, cmd: BinanceFuturesWsTradingCommand) -> BinanceFuturesWsApiResult<()> {
471        self.cmd_tx
472            .read()
473            .await
474            .send(cmd)
475            .map_err(|e| BinanceFuturesWsApiError::HandlerUnavailable(e.to_string()))
476    }
477}
478
479#[cfg(test)]
480mod tests {
481    use rstest::rstest;
482
483    use super::*;
484
485    #[rstest]
486    fn test_operational_options_are_preserved() {
487        let client = BinanceFuturesWsTradingClient::new(
488            None,
489            "api-key".to_string(),
490            "hmac-secret".to_string(),
491            None,
492            TransportBackend::default(),
493        )
494        .with_proxy(Some("http://proxy.example:8080".to_string()))
495        .with_recv_window(Some(30_000));
496
497        assert_eq!(
498            client.proxy_url.as_ref().map(SecretString::expose_secret),
499            Some("http://proxy.example:8080")
500        );
501        assert_eq!(client.recv_window_ms, Some(30_000));
502    }
503}
nautilus_binance/futures/websocket/trading/client.rs

nautilus_binance/futures/websocket/trading/
client.rs
