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

nautilus_binance/futures/websocket/trading/
client.rs
