nautilus_binance/spot/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 Spot WebSocket API client for SBE trading.
17//!
18//! ## Connection Details
19//!
20//! - Endpoint: `ws-api.binance.com:443/ws-api/v3`
21//! - Authentication: Ed25519 signature per request
22//! - SBE responses: Enabled via `responseFormat=sbe` query parameter
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        AuthTracker, PingHandler, TransportBackend, WebSocketClient, WebSocketConfig,
44        channel_message_handler,
45    },
46};
47use parking_lot::Mutex;
48use tokio_util::sync::CancellationToken;
49use ustr::Ustr;
50
51use super::{
52    error::{BinanceWsApiError, BinanceWsApiResult},
53    handler::BinanceSpotWsTradingHandler,
54    messages::{BinanceSpotWsTradingCommand, BinanceSpotWsTradingMessage},
55};
56use crate::{
57    common::{
58        consts::{BINANCE_API_KEY_HEADER, BINANCE_SPOT_SBE_WS_API_URL},
59        credential::SigningCredential,
60    },
61    spot::http::query::{CancelOrderParams, CancelReplaceOrderParams, NewOrderParams},
62};
63
64/// Environment variable key for Binance API key.
65pub const BINANCE_API_KEY: &str = "BINANCE_API_KEY";
66
67/// Environment variable key for Binance API secret.
68pub const BINANCE_API_SECRET: &str = "BINANCE_API_SECRET";
69
70/// Pre-interned rate limit key for order operations (place/cancel/replace).
71///
72/// Binance WebSocket API: 1200 requests per minute per IP (20/sec).
73pub static BINANCE_WS_RATE_LIMIT_KEY_ORDER: LazyLock<[Ustr; 1]> =
74    LazyLock::new(|| [Ustr::from("order")]);
75
76/// Binance WebSocket API order rate limit: 1200 per minute (20/sec).
77///
78/// Based on Binance documentation for WebSocket API rate limits.
79// Constant values are provably valid
80#[expect(clippy::missing_panics_doc)]
81#[must_use]
82pub fn binance_ws_order_quota() -> Quota {
83    Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant")
84}
85
86/// Binance Spot WebSocket API client for SBE trading.
87///
88/// This client provides order management via WebSocket with SBE-encoded responses,
89/// complementing the HTTP client with lower-latency order submission.
90#[derive(Clone)]
91pub struct BinanceSpotWsTradingClient {
92    url: String,
93    credential: Arc<SigningCredential>,
94    heartbeat: Option<u64>,
95    signal: Arc<AtomicBool>,
96    connection_mode: Arc<ArcSwap<AtomicU8>>,
97    user_data_tracker: AuthTracker,
98    cmd_tx:
99        Arc<tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<BinanceSpotWsTradingCommand>>>,
100    out_rx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsTradingMessage>>>>,
101    handler_tasks: Arc<TaskGroup>,
102    connect_lock: Arc<tokio::sync::Mutex<()>>,
103    request_id_counter: Arc<AtomicU64>,
104    cancellation_token: Arc<Mutex<CancellationToken>>,
105    transport_backend: TransportBackend,
106    proxy_url: Option<String>,
107    recv_window_ms: Option<u64>,
108    socket_control: Option<SocketControl>,
109}
110
111impl Debug for BinanceSpotWsTradingClient {
112    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
113        f.debug_struct(stringify!(BinanceSpotWsTradingClient))
114            .field("url", &self.url)
115            .field("credential", &REDACTED)
116            .field("heartbeat", &self.heartbeat)
117            .finish_non_exhaustive()
118    }
119}
120
121impl BinanceSpotWsTradingClient {
122    /// Creates a new [`BinanceSpotWsTradingClient`] instance.
123    #[must_use]
124    pub fn new(
125        url: Option<String>,
126        api_key: String,
127        api_secret: String,
128        heartbeat: Option<u64>,
129        transport_backend: TransportBackend,
130    ) -> Self {
131        let url = url.unwrap_or_else(|| BINANCE_SPOT_SBE_WS_API_URL.to_string());
132        let credential = Arc::new(SigningCredential::new(api_key, api_secret));
133
134        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel();
135
136        Self {
137            url,
138            credential,
139            heartbeat,
140            signal: Arc::new(AtomicBool::new(false)),
141            connection_mode: Arc::new(ArcSwap::new(Arc::new(AtomicU8::new(
142                ConnectionMode::Closed as u8,
143            )))),
144            user_data_tracker: AuthTracker::new(),
145            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
146            out_rx: Arc::new(Mutex::new(None)),
147            handler_tasks: Arc::new(TaskGroup::new()),
148            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
149            request_id_counter: Arc::new(AtomicU64::new(1)),
150            cancellation_token: Arc::new(Mutex::new(CancellationToken::new())),
151            transport_backend,
152            proxy_url: None,
153            recv_window_ms: None,
154            socket_control: None,
155        }
156    }
157
158    /// Configures the proxy used by the WebSocket connection.
159    #[must_use]
160    pub fn with_proxy(mut self, proxy_url: Option<String>) -> Self {
161        self.proxy_url = proxy_url;
162        self
163    }
164
165    /// Configures socket state reporting and reconnect control.
166    #[must_use]
167    pub fn with_socket_control(mut self, control: SocketControl) -> Self {
168        self.socket_control = Some(control);
169        self
170    }
171
172    /// Configures the receive window added to signed WebSocket API requests.
173    #[must_use]
174    pub const fn with_recv_window(mut self, recv_window_ms: Option<u64>) -> Self {
175        self.recv_window_ms = recv_window_ms;
176        self
177    }
178
179    /// Creates a new client with credentials sourced from environment variables.
180    ///
181    /// Falls back to env vars if `api_key` or `api_secret` are `None`:
182    /// - `BINANCE_API_KEY` for the API key
183    /// - `BINANCE_API_SECRET` for the API secret
184    ///
185    /// # Errors
186    ///
187    /// Returns an error if credentials are missing from environment.
188    pub fn with_env(
189        url: Option<String>,
190        api_key: Option<String>,
191        api_secret: Option<String>,
192        heartbeat: Option<u64>,
193        transport_backend: TransportBackend,
194    ) -> anyhow::Result<Self> {
195        let api_key = nautilus_core::env::get_or_env_var(api_key, BINANCE_API_KEY)?;
196        let api_secret = nautilus_core::env::get_or_env_var(api_secret, BINANCE_API_SECRET)?;
197        Ok(Self::new(
198            url,
199            api_key,
200            api_secret,
201            heartbeat,
202            transport_backend,
203        ))
204    }
205
206    /// Creates a new client with credentials loaded entirely from environment variables.
207    ///
208    /// Reads:
209    /// - `BINANCE_API_KEY` for the API key
210    /// - `BINANCE_API_SECRET` for the API secret
211    ///
212    /// # Errors
213    ///
214    /// Returns an error if environment variables are missing.
215    pub fn from_env(url: Option<String>, heartbeat: Option<u64>) -> anyhow::Result<Self> {
216        Self::with_env(url, None, None, heartbeat, TransportBackend::default())
217    }
218
219    /// Returns whether the client is actively connected.
220    #[must_use]
221    pub fn is_active(&self) -> bool {
222        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
223        mode_u8 == ConnectionMode::Active as u8
224    }
225
226    /// Returns whether the private user data stream is active on the current connection.
227    #[must_use]
228    pub fn is_user_data_active(&self) -> bool {
229        self.is_active() && self.user_data_tracker.is_authenticated()
230    }
231
232    /// Marks the private user data stream active on the current connection.
233    pub fn mark_user_data_active(&self) {
234        self.user_data_tracker.succeed();
235    }
236
237    /// Marks the private user data stream inactive.
238    pub fn mark_user_data_inactive(&self) {
239        self.user_data_tracker.invalidate();
240    }
241
242    /// Returns whether the client is closed.
243    #[must_use]
244    pub fn is_closed(&self) -> bool {
245        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
246        mode_u8 == ConnectionMode::Closed as u8
247    }
248
249    /// Generates the next request ID.
250    pub fn next_request_id(&self) -> String {
251        let id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
252        format!("req-{id}")
253    }
254
255    /// Connects to the WebSocket API server.
256    ///
257    /// # Errors
258    ///
259    /// Returns an error if connection fails.
260    pub async fn connect(&mut self) -> BinanceWsApiResult<()> {
261        let connect_lock = Arc::clone(&self.connect_lock);
262        let _connect_guard = connect_lock.lock().await;
263
264        if !self.handler_tasks.is_open() || !self.handler_tasks.is_empty() {
265            self.disconnect_handler().await?;
266            self.handler_tasks.start_generation().map_err(|e| {
267                BinanceWsApiError::ClientError(format!(
268                    "failed to start WebSocket handler task generation: {e}"
269                ))
270            })?;
271        }
272        let handler_spawner = self.handler_tasks.spawner().map_err(|e| {
273            BinanceWsApiError::ClientError(format!(
274                "failed to acquire WebSocket handler task spawner: {e}"
275            ))
276        })?;
277        self.signal.store(false, Ordering::Relaxed);
278        self.user_data_tracker.invalidate();
279        *self.cancellation_token.lock() = CancellationToken::new();
280
281        let (raw_handler, raw_rx) = channel_message_handler();
282        let ping_handler: PingHandler = Arc::new(move |_| {});
283
284        let headers = vec![(
285            BINANCE_API_KEY_HEADER.to_string(),
286            self.credential.api_key().to_string(),
287        )];
288
289        let config = WebSocketConfig {
290            url: self.url.clone(),
291            headers,
292            heartbeat_interval_secs: self.heartbeat,
293            heartbeat_payload: None,
294            connect_timeout_ms: Some(5_000),
295            reconnect_delay_initial_ms: Some(500),
296            reconnect_delay_max_ms: Some(5_000),
297            reconnect_backoff_factor: Some(2.0),
298            reconnect_jitter_ms: Some(250),
299            reconnect_max_attempts: None,
300            heartbeat_timeout_secs: None,
301            idle_timeout_ms: None,
302            backend: self.transport_backend,
303            proxy_url: self.proxy_url.clone(),
304        };
305
306        // Configure rate limits for order operations
307        let keyed_quotas = vec![(
308            BINANCE_WS_RATE_LIMIT_KEY_ORDER[0].as_str().to_string(),
309            binance_ws_order_quota(),
310        )];
311
312        let client = WebSocketClient::builder()
313            .config(config)
314            .message_handler(raw_handler)
315            .ping_handler(ping_handler)
316            .keyed_quotas(keyed_quotas)
317            .default_quota(binance_ws_order_quota())
318            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
319            .connect()
320            .await
321            .map_err(|e| BinanceWsApiError::ConnectionError(e.to_string()))?;
322
323        client.set_auth_tracker(self.user_data_tracker.clone(), true);
324        self.connection_mode.store(client.connection_mode_atomic());
325        let reconnect_handle = client.reconnect_handle();
326
327        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
328        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
329
330        {
331            let mut rx_guard = self.out_rx.lock();
332            *rx_guard = Some(out_rx);
333        }
334
335        {
336            let mut tx_guard = self.cmd_tx.write().await;
337            *tx_guard = cmd_tx;
338        }
339
340        let signal = self.signal.clone();
341        let credential = self.credential.clone();
342        let mut handler =
343            BinanceSpotWsTradingHandler::new(signal, cmd_rx, raw_rx, out_tx, credential)
344                .with_recv_window(self.recv_window_ms);
345
346        self.cmd_tx
347            .read()
348            .await
349            .send(BinanceSpotWsTradingCommand::SetClient(client))
350            .map_err(|e| BinanceWsApiError::HandlerUnavailable(e.to_string()))?;
351        if let Some(control) = &self.socket_control {
352            control.register(move || reconnect_handle.request_reconnect());
353        }
354
355        let cancellation_token = self.cancellation_token.lock().clone();
356
357        let handler_task = async move {
358            tokio::select! {
359                () = cancellation_token.cancelled() => {
360                    log::debug!("Handler task cancelled");
361                }
362                _ = handler.run() => {
363                    log::debug!("Handler run completed");
364                }
365            }
366        };
367
368        if let Err(e) = handler_spawner.spawn(handler_task) {
369            if let Some(control) = &self.socket_control {
370                control.deregister();
371            }
372            self.out_rx.lock().take();
373            return Err(BinanceWsApiError::HandlerUnavailable(format!(
374                "failed to register handler task: {e}"
375            )));
376        }
377
378        Ok(())
379    }
380
381    /// Disconnects from the WebSocket API server.
382    ///
383    /// # Errors
384    ///
385    /// Returns an error if the handler task fails or does not stop after abort.
386    pub async fn disconnect(&mut self) -> BinanceWsApiResult<()> {
387        let connect_lock = Arc::clone(&self.connect_lock);
388        let _connect_guard = connect_lock.lock().await;
389
390        self.disconnect_handler().await
391    }
392
393    pub(crate) fn begin_shutdown(&self) {
394        self.handler_tasks.begin_shutdown();
395        self.signal.store(true, Ordering::Relaxed);
396        self.cancellation_token.lock().cancel();
397    }
398
399    async fn disconnect_handler(&self) -> BinanceWsApiResult<()> {
400        self.handler_tasks.begin_shutdown();
401        self.signal.store(true, Ordering::Relaxed);
402
403        if let Err(e) = self
404            .cmd_tx
405            .read()
406            .await
407            .send(BinanceSpotWsTradingCommand::Disconnect)
408        {
409            log::debug!("Failed to send disconnect command: {e}");
410        }
411
412        self.cancellation_token.lock().cancel();
413
414        let result = self
415            .handler_tasks
416            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
417            .await
418            .map_err(|e| {
419                BinanceWsApiError::ClientError(format!("handler task shutdown failed: {e}"))
420            });
421
422        if let Some(control) = &self.socket_control {
423            control.deregister();
424        }
425        result
426    }
427
428    /// Places a new order via WebSocket API.
429    ///
430    /// # Errors
431    ///
432    /// Returns an error if the handler is unavailable.
433    pub async fn place_order(&self, params: NewOrderParams) -> BinanceWsApiResult<String> {
434        let id = self.next_request_id();
435        self.place_order_with_id(id.clone(), params).await?;
436        Ok(id)
437    }
438
439    /// Places a new order via WebSocket API using a pre-generated request ID.
440    ///
441    /// # Errors
442    ///
443    /// Returns an error if the handler is unavailable.
444    pub async fn place_order_with_id(
445        &self,
446        id: String,
447        params: NewOrderParams,
448    ) -> BinanceWsApiResult<()> {
449        let cmd = BinanceSpotWsTradingCommand::PlaceOrder { id, params };
450        self.send_cmd(cmd).await
451    }
452
453    /// Cancels an order via WebSocket API.
454    ///
455    /// # Errors
456    ///
457    /// Returns an error if the handler is unavailable.
458    pub async fn cancel_order(&self, params: CancelOrderParams) -> BinanceWsApiResult<String> {
459        let id = self.next_request_id();
460        self.cancel_order_with_id(id.clone(), params).await?;
461        Ok(id)
462    }
463
464    /// Cancels an order via WebSocket API using a pre-generated request ID.
465    ///
466    /// # Errors
467    ///
468    /// Returns an error if the handler is unavailable.
469    pub async fn cancel_order_with_id(
470        &self,
471        id: String,
472        params: CancelOrderParams,
473    ) -> BinanceWsApiResult<()> {
474        let cmd = BinanceSpotWsTradingCommand::CancelOrder { id, params };
475        self.send_cmd(cmd).await
476    }
477
478    /// Cancels and replaces an order atomically via WebSocket API.
479    ///
480    /// # Errors
481    ///
482    /// Returns an error if the handler is unavailable.
483    pub async fn cancel_replace_order(
484        &self,
485        params: CancelReplaceOrderParams,
486    ) -> BinanceWsApiResult<String> {
487        let id = self.next_request_id();
488        self.cancel_replace_order_with_id(id.clone(), params)
489            .await?;
490        Ok(id)
491    }
492
493    /// Cancels and replaces an order atomically via WebSocket API using a pre-generated request ID.
494    ///
495    /// # Errors
496    ///
497    /// Returns an error if the handler is unavailable.
498    pub async fn cancel_replace_order_with_id(
499        &self,
500        id: String,
501        params: CancelReplaceOrderParams,
502    ) -> BinanceWsApiResult<()> {
503        let cmd = BinanceSpotWsTradingCommand::CancelReplaceOrder { id, params };
504        self.send_cmd(cmd).await
505    }
506
507    /// Cancels all open orders for a symbol via WebSocket API.
508    ///
509    /// # Errors
510    ///
511    /// Returns an error if the handler is unavailable.
512    pub async fn cancel_all_orders(&self, symbol: impl Into<String>) -> BinanceWsApiResult<String> {
513        let id = self.next_request_id();
514        let cmd = BinanceSpotWsTradingCommand::CancelAllOrders {
515            id: id.clone(),
516            symbol: symbol.into(),
517        };
518        self.send_cmd(cmd).await?;
519        Ok(id)
520    }
521
522    /// Receives the next message from the handler.
523    ///
524    /// Returns `None` if the receiver is closed or not initialized.
525    pub async fn recv(&self) -> Option<BinanceSpotWsTradingMessage> {
526        // Take the receiver out of the mutex to avoid holding it across await
527        let rx_opt = {
528            let mut rx_guard = self.out_rx.lock();
529            rx_guard.take()
530        };
531
532        if let Some(mut rx) = rx_opt {
533            let result = rx.recv().await;
534
535            let mut rx_guard = self.out_rx.lock();
536            *rx_guard = Some(rx);
537            result
538        } else {
539            None
540        }
541    }
542
543    /// Authenticates the WebSocket session via `session.logon`.
544    ///
545    /// # Errors
546    ///
547    /// Returns an error if the handler is unavailable.
548    pub async fn session_logon(&self) -> BinanceWsApiResult<()> {
549        self.send_cmd(BinanceSpotWsTradingCommand::SessionLogon)
550            .await
551    }
552
553    /// Subscribes to the user data stream via `userDataStream.subscribe`.
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if the handler is unavailable.
558    pub async fn subscribe_user_data(&self) -> BinanceWsApiResult<()> {
559        self.send_cmd(BinanceSpotWsTradingCommand::SubscribeUserData)
560            .await
561    }
562
563    async fn send_cmd(&self, cmd: BinanceSpotWsTradingCommand) -> BinanceWsApiResult<()> {
564        self.cmd_tx
565            .read()
566            .await
567            .send(cmd)
568            .map_err(|e| BinanceWsApiError::HandlerUnavailable(e.to_string()))
569    }
570}
571
572#[cfg(test)]
573mod tests {
574    use rstest::rstest;
575
576    use super::*;
577
578    #[rstest]
579    fn test_operational_options_are_preserved() {
580        let client = BinanceSpotWsTradingClient::new(
581            None,
582            "api-key".to_string(),
583            "hmac-secret".to_string(),
584            None,
585            TransportBackend::default(),
586        )
587        .with_proxy(Some("http://proxy.example:8080".to_string()))
588        .with_recv_window(Some(45_000));
589
590        assert_eq!(
591            client.proxy_url.as_deref(),
592            Some("http://proxy.example:8080")
593        );
594        assert_eq!(client.recv_window_ms, Some(45_000));
595    }
596}
nautilus_binance/spot/websocket/trading/client.rs

nautilus_binance/spot/websocket/trading/
client.rs
