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

nautilus_binance/futures/websocket/trading/
client.rs
