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            writer_capacity: None,
306            backend: self.transport_backend,
307            proxy_url: self
308                .proxy_url
309                .as_ref()
310                .map(|value| value.expose_secret().to_owned()),
311            max_message_size_bytes: None,
312            max_frame_size_bytes: None,
313        };
314
315        // Configure rate limits for order operations
316        let keyed_quotas = vec![(
317            BINANCE_WS_RATE_LIMIT_KEY_ORDER[0].to_string(),
318            binance_ws_order_quota(),
319        )];
320
321        let client = WebSocketClient::builder()
322            .config(config)
323            .message_handler(raw_handler)
324            .ping_handler(ping_handler)
325            .keyed_quotas(keyed_quotas)
326            .default_quota(binance_ws_order_quota())
327            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
328            .connect()
329            .await
330            .map_err(|e| BinanceWsApiError::ConnectionError(e.to_string()))?;
331
332        client.set_auth_tracker(self.user_data_tracker.clone(), true);
333        self.connection_mode.store(client.connection_mode_atomic());
334        let reconnect_handle = client.reconnect_handle();
335
336        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
337        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
338
339        cmd_tx
340            .send(BinanceSpotWsTradingCommand::SetClient(client))
341            .map_err(|e| BinanceWsApiError::HandlerUnavailable(e.to_string()))?;
342
343        {
344            let mut rx_guard = self.out_rx.lock();
345            *rx_guard = Some(out_rx);
346        }
347
348        {
349            let mut tx_guard = self.cmd_tx.write().await;
350            *tx_guard = cmd_tx;
351        }
352
353        let signal = self.signal.clone();
354        let credential = self.credential.clone();
355        let mut handler =
356            BinanceSpotWsTradingHandler::new(signal, cmd_rx, raw_rx, out_tx, credential)
357                .with_recv_window(self.recv_window_ms);
358
359        if let Some(control) = &self.socket_control {
360            control.register(move || reconnect_handle.request_reconnect());
361        }
362
363        let cancellation_token = self.cancellation_token.lock().clone();
364
365        let handler_task = async move {
366            tokio::select! {
367                () = cancellation_token.cancelled() => {
368                    log::debug!("Handler task cancelled");
369                }
370                _ = handler.run() => {
371                    log::debug!("Handler run completed");
372                }
373            }
374        };
375
376        if let Err(e) = handler_spawner.spawn(handler_task) {
377            if let Some(control) = &self.socket_control {
378                control.deregister();
379            }
380            self.out_rx.lock().take();
381            return Err(BinanceWsApiError::HandlerUnavailable(format!(
382                "failed to register handler task: {e}"
383            )));
384        }
385
386        Ok(())
387    }
388
389    /// Disconnects from the WebSocket API server.
390    ///
391    /// # Errors
392    ///
393    /// Returns an error if the handler task fails or does not stop after abort.
394    pub async fn disconnect(&mut self) -> BinanceWsApiResult<()> {
395        let connect_lock = Arc::clone(&self.connect_lock);
396        let _connect_guard = connect_lock.lock().await;
397
398        self.disconnect_handler().await
399    }
400
401    pub(crate) fn begin_shutdown(&self) {
402        self.handler_tasks.begin_shutdown();
403        self.signal.store(true, Ordering::Relaxed);
404        self.cancellation_token.lock().cancel();
405    }
406
407    async fn disconnect_handler(&self) -> BinanceWsApiResult<()> {
408        self.handler_tasks.begin_shutdown();
409        self.signal.store(true, Ordering::Relaxed);
410
411        if let Err(e) = self
412            .cmd_tx
413            .read()
414            .await
415            .send(BinanceSpotWsTradingCommand::Disconnect)
416        {
417            log::debug!("Failed to send disconnect command: {e}");
418        }
419
420        self.cancellation_token.lock().cancel();
421
422        let result = self
423            .handler_tasks
424            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
425            .await
426            .map_err(|e| {
427                BinanceWsApiError::ClientError(format!("handler task shutdown failed: {e}"))
428            });
429
430        if let Some(control) = &self.socket_control {
431            control.deregister();
432        }
433        result
434    }
435
436    /// Places a new order via WebSocket API.
437    ///
438    /// # Errors
439    ///
440    /// Returns an error if the handler is unavailable.
441    pub async fn place_order(&self, params: NewOrderParams) -> BinanceWsApiResult<String> {
442        let id = self.next_request_id();
443        self.place_order_with_id(id.clone(), params).await?;
444        Ok(id)
445    }
446
447    /// Places a new order via WebSocket API using a pre-generated request ID.
448    ///
449    /// # Errors
450    ///
451    /// Returns an error if the handler is unavailable.
452    pub async fn place_order_with_id(
453        &self,
454        id: String,
455        params: NewOrderParams,
456    ) -> BinanceWsApiResult<()> {
457        let cmd = BinanceSpotWsTradingCommand::PlaceOrder { id, params };
458        self.send_cmd(cmd).await
459    }
460
461    /// Cancels an order via WebSocket API.
462    ///
463    /// # Errors
464    ///
465    /// Returns an error if the handler is unavailable.
466    pub async fn cancel_order(&self, params: CancelOrderParams) -> BinanceWsApiResult<String> {
467        let id = self.next_request_id();
468        self.cancel_order_with_id(id.clone(), params).await?;
469        Ok(id)
470    }
471
472    /// Cancels an order via WebSocket API using a pre-generated request ID.
473    ///
474    /// # Errors
475    ///
476    /// Returns an error if the handler is unavailable.
477    pub async fn cancel_order_with_id(
478        &self,
479        id: String,
480        params: CancelOrderParams,
481    ) -> BinanceWsApiResult<()> {
482        let cmd = BinanceSpotWsTradingCommand::CancelOrder { id, params };
483        self.send_cmd(cmd).await
484    }
485
486    /// Cancels and replaces an order atomically via WebSocket API.
487    ///
488    /// # Errors
489    ///
490    /// Returns an error if the handler is unavailable.
491    pub async fn cancel_replace_order(
492        &self,
493        params: CancelReplaceOrderParams,
494    ) -> BinanceWsApiResult<String> {
495        let id = self.next_request_id();
496        self.cancel_replace_order_with_id(id.clone(), params)
497            .await?;
498        Ok(id)
499    }
500
501    /// Cancels and replaces an order atomically via WebSocket API using a pre-generated request ID.
502    ///
503    /// # Errors
504    ///
505    /// Returns an error if the handler is unavailable.
506    pub async fn cancel_replace_order_with_id(
507        &self,
508        id: String,
509        params: CancelReplaceOrderParams,
510    ) -> BinanceWsApiResult<()> {
511        let cmd = BinanceSpotWsTradingCommand::CancelReplaceOrder { id, params };
512        self.send_cmd(cmd).await
513    }
514
515    /// Cancels all open orders for a symbol via WebSocket API.
516    ///
517    /// # Errors
518    ///
519    /// Returns an error if the handler is unavailable.
520    pub async fn cancel_all_orders(&self, symbol: impl Into<String>) -> BinanceWsApiResult<String> {
521        let id = self.next_request_id();
522        let cmd = BinanceSpotWsTradingCommand::CancelAllOrders {
523            id: id.clone(),
524            symbol: symbol.into(),
525        };
526        self.send_cmd(cmd).await?;
527        Ok(id)
528    }
529
530    /// Receives the next message from the handler.
531    ///
532    /// Returns `None` if the receiver is closed or not initialized.
533    pub async fn recv(&self) -> Option<BinanceSpotWsTradingMessage> {
534        // Take the receiver out of the mutex to avoid holding it across await
535        let rx_opt = {
536            let mut rx_guard = self.out_rx.lock();
537            rx_guard.take()
538        };
539
540        if let Some(mut rx) = rx_opt {
541            let result = rx.recv().await;
542
543            let mut rx_guard = self.out_rx.lock();
544            *rx_guard = Some(rx);
545            result
546        } else {
547            None
548        }
549    }
550
551    /// Authenticates the WebSocket session via `session.logon`.
552    ///
553    /// # Errors
554    ///
555    /// Returns an error if the handler is unavailable.
556    pub async fn session_logon(&self) -> BinanceWsApiResult<()> {
557        self.send_cmd(BinanceSpotWsTradingCommand::SessionLogon)
558            .await
559    }
560
561    /// Subscribes to the user data stream via `userDataStream.subscribe`.
562    ///
563    /// # Errors
564    ///
565    /// Returns an error if the handler is unavailable.
566    pub async fn subscribe_user_data(&self) -> BinanceWsApiResult<()> {
567        self.send_cmd(BinanceSpotWsTradingCommand::SubscribeUserData)
568            .await
569    }
570
571    async fn send_cmd(&self, cmd: BinanceSpotWsTradingCommand) -> BinanceWsApiResult<()> {
572        self.cmd_tx
573            .read()
574            .await
575            .send(cmd)
576            .map_err(|e| BinanceWsApiError::HandlerUnavailable(e.to_string()))
577    }
578}
579
580#[cfg(test)]
581mod tests {
582    use rstest::rstest;
583
584    use super::*;
585
586    #[rstest]
587    fn test_operational_options_are_preserved() {
588        let client = BinanceSpotWsTradingClient::new(
589            None,
590            "api-key".to_string(),
591            "hmac-secret".to_string(),
592            None,
593            TransportBackend::default(),
594        )
595        .with_proxy(Some("http://proxy.example:8080".to_string()))
596        .with_recv_window(Some(45_000));
597
598        assert_eq!(
599            client.proxy_url.as_ref().map(SecretString::expose_secret),
600            Some("http://proxy.example:8080")
601        );
602        assert_eq!(client.recv_window_ms, Some(45_000));
603    }
604
605    #[rstest]
606    fn test_url_is_redacted() {
607        let url = "wss://stream.example/ws/private-listen-key";
608        let client = BinanceSpotWsTradingClient::new(
609            Some(url.to_string()),
610            "api-key".to_string(),
611            "hmac-secret".to_string(),
612            None,
613            TransportBackend::default(),
614        );
615
616        let debug = format!("{client:?}");
617
618        assert_eq!(client.url.expose_secret(), url);
619        assert!(debug.contains(REDACTED));
620        assert!(!debug.contains("private-listen-key"));
621    }
622}
nautilus_binance/spot/websocket/trading/client.rs

nautilus_binance/spot/websocket/trading/
client.rs
