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

nautilus_binance/spot/websocket/trading/
client.rs
