nautilus_binance/futures/websocket/streams/
handler.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 handler for JSON streams.
17//!
18//! The handler is a stateless I/O boundary: it deserializes raw JSON into
19//! venue-specific types and emits them on the output channel. Domain conversion
20//! happens in the data and execution client layers.
21
22use std::{
23    fmt::Debug,
24    sync::{
25        Arc,
26        atomic::{AtomicBool, AtomicU64, Ordering},
27    },
28};
29
30use nautilus_network::{
31    RECONNECTED,
32    websocket::{SubscriptionState, WebSocketClient},
33};
34
35use super::{
36    messages::{
37        BinanceFuturesAccountConfigMsg, BinanceFuturesAccountUpdateMsg, BinanceFuturesAggTradeMsg,
38        BinanceFuturesAlgoUpdateMsg, BinanceFuturesBookTickerMsg, BinanceFuturesDepthUpdateMsg,
39        BinanceFuturesKlineMsg, BinanceFuturesLiquidationMsg, BinanceFuturesListenKeyExpiredMsg,
40        BinanceFuturesMarginCallMsg, BinanceFuturesMarkPriceMsg, BinanceFuturesOrderUpdateMsg,
41        BinanceFuturesTickerMsg, BinanceFuturesTradeLiteMsg, BinanceFuturesTradeMsg,
42        BinanceFuturesWsErrorMsg, BinanceFuturesWsErrorResponse, BinanceFuturesWsStreamsCommand,
43        BinanceFuturesWsStreamsMessage, BinanceFuturesWsSubscribeRequest,
44        BinanceFuturesWsSubscribeResponse,
45    },
46    parse_data::extract_event_type,
47};
48use crate::common::{
49    consts::BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION,
50    enums::{BinanceWsEventType, BinanceWsMethod},
51    websocket::{
52        PendingSubscriptionRequest, PendingSubscriptionRequests, reset_requests_after_reconnect,
53    },
54};
55
56/// Handler for Binance Futures WebSocket JSON streams.
57///
58/// Deserializes raw JSON into venue-specific types without performing
59/// domain conversion. The data and execution client layers own instrument
60/// lookups and Nautilus type construction.
61pub struct BinanceFuturesDataWsFeedHandler {
62    #[allow(dead_code)]
63    signal: Arc<AtomicBool>,
64    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsStreamsCommand>,
65    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
66    #[allow(dead_code)]
67    out_tx: tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsStreamsMessage>,
68    inner: Option<WebSocketClient>,
69    subscriptions_state: SubscriptionState,
70    request_id_counter: Arc<AtomicU64>,
71    pending_requests: PendingSubscriptionRequests,
72}
73
74impl Debug for BinanceFuturesDataWsFeedHandler {
75    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76        f.debug_struct(stringify!(BinanceFuturesDataWsFeedHandler))
77            .field("pending_requests", &self.pending_requests.len())
78            .finish_non_exhaustive()
79    }
80}
81
82impl BinanceFuturesDataWsFeedHandler {
83    /// Creates a new handler instance.
84    pub fn new(
85        signal: Arc<AtomicBool>,
86        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsStreamsCommand>,
87        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
88        out_tx: tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsStreamsMessage>,
89        subscriptions_state: SubscriptionState,
90        request_id_counter: Arc<AtomicU64>,
91    ) -> Self {
92        Self {
93            signal,
94            cmd_rx,
95            raw_rx,
96            out_tx,
97            inner: None,
98            subscriptions_state,
99            request_id_counter,
100            pending_requests: PendingSubscriptionRequests::default(),
101        }
102    }
103
104    /// Returns the next message from the handler.
105    ///
106    /// Processes both commands and raw WebSocket messages.
107    pub async fn next(&mut self) -> Option<BinanceFuturesWsStreamsMessage> {
108        loop {
109            if self.signal.load(Ordering::Relaxed) {
110                return None;
111            }
112
113            tokio::select! {
114                Some(cmd) = self.cmd_rx.recv() => {
115                    self.handle_command(cmd).await;
116                }
117                Some(raw) = self.raw_rx.recv() => {
118                    if let Some(msg) = self.handle_raw_message(raw).await {
119                        return Some(msg);
120                    }
121                }
122                else => {
123                    return None;
124                }
125            }
126        }
127    }
128
129    async fn handle_command(&mut self, cmd: BinanceFuturesWsStreamsCommand) {
130        match cmd {
131            BinanceFuturesWsStreamsCommand::SetClient(client) => {
132                self.inner = Some(client);
133            }
134            BinanceFuturesWsStreamsCommand::Disconnect => {
135                if let Some(client) = &self.inner {
136                    let () = client.disconnect().await;
137                }
138                self.inner = None;
139            }
140            BinanceFuturesWsStreamsCommand::Subscribe { streams } => {
141                self.send_subscribe(streams).await;
142            }
143            BinanceFuturesWsStreamsCommand::Unsubscribe { streams } => {
144                self.send_unsubscribe(streams).await;
145            }
146        }
147    }
148
149    async fn send_subscribe(&mut self, streams: Vec<String>) {
150        for stream in &streams {
151            self.subscriptions_state.mark_subscribe(stream);
152        }
153
154        let Some(client) = &self.inner else {
155            log::warn!("Cannot subscribe: no client connected");
156            return;
157        };
158
159        let request_id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
160
161        let request = BinanceFuturesWsSubscribeRequest {
162            method: BinanceWsMethod::Subscribe,
163            params: streams.clone(),
164            id: request_id,
165        };
166
167        let json = match serde_json::to_string(&request) {
168            Ok(j) => j,
169            Err(e) => {
170                log::error!("Failed to serialize subscribe request: {e}");
171                return;
172            }
173        };
174
175        self.pending_requests
176            .insert(request_id, PendingSubscriptionRequest::subscribe(streams));
177
178        if let Err(e) = client
179            .send_text(json, Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
180            .await
181        {
182            if let Some(request) = self.pending_requests.take(request_id) {
183                request.mark_failure(&self.subscriptions_state);
184            }
185            log::error!("Failed to send subscribe request: {e}");
186        }
187    }
188
189    async fn send_unsubscribe(&mut self, streams: Vec<String>) {
190        for stream in &streams {
191            self.subscriptions_state.mark_unsubscribe(stream);
192        }
193
194        let Some(client) = &self.inner else {
195            log::warn!("Cannot unsubscribe: no client connected");
196            return;
197        };
198
199        let request_id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
200
201        let request = BinanceFuturesWsSubscribeRequest {
202            method: BinanceWsMethod::Unsubscribe,
203            params: streams.clone(),
204            id: request_id,
205        };
206
207        let json = match serde_json::to_string(&request) {
208            Ok(j) => j,
209            Err(e) => {
210                log::error!("Failed to serialize unsubscribe request: {e}");
211                return;
212            }
213        };
214
215        self.pending_requests
216            .insert(request_id, PendingSubscriptionRequest::unsubscribe(streams));
217
218        if let Err(e) = client
219            .send_text(json, Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
220            .await
221        {
222            self.pending_requests.take(request_id);
223            log::error!("Failed to send unsubscribe request: {e}");
224        }
225    }
226
227    async fn handle_raw_message(&mut self, raw: Vec<u8>) -> Option<BinanceFuturesWsStreamsMessage> {
228        if let Ok(text) = std::str::from_utf8(&raw)
229            && text == RECONNECTED
230        {
231            reset_requests_after_reconnect(&mut self.pending_requests, &self.subscriptions_state);
232            log::debug!("WebSocket reconnected signal received");
233            return Some(BinanceFuturesWsStreamsMessage::Reconnected);
234        }
235
236        let json: serde_json::Value = match serde_json::from_slice(&raw) {
237            Ok(j) => j,
238            Err(e) => {
239                log::warn!("Failed to parse JSON message: {e}");
240                return None;
241            }
242        };
243
244        if json.get("result").is_some() || json.get("id").is_some() {
245            self.handle_subscription_response(&json);
246            return None;
247        }
248
249        if let Some(code) = json.get("code")
250            && let Some(code) = code.as_i64()
251        {
252            let msg = json
253                .get("msg")
254                .and_then(|m| m.as_str())
255                .unwrap_or("Unknown error")
256                .to_string();
257            return Some(BinanceFuturesWsStreamsMessage::Error(
258                BinanceFuturesWsErrorMsg { code, msg },
259            ));
260        }
261
262        self.handle_stream_data(&json)
263    }
264
265    fn handle_subscription_response(&mut self, json: &serde_json::Value) {
266        if let Ok(error) = serde_json::from_value::<BinanceFuturesWsErrorResponse>(json.clone()) {
267            if let Some(id) = error.id
268                && let Some(request) = self.pending_requests.take(id)
269            {
270                request.mark_failure(&self.subscriptions_state);
271            }
272            log::warn!(
273                "WebSocket error response: code={}, msg={}",
274                error.code,
275                error.msg
276            );
277        } else if let Ok(response) =
278            serde_json::from_value::<BinanceFuturesWsSubscribeResponse>(json.clone())
279            && let Some(request) = self.pending_requests.take(response.id)
280        {
281            if response.result.is_none() {
282                request.confirm(&self.subscriptions_state);
283                log::debug!("Subscription request confirmed: request={request:?}");
284            } else {
285                request.mark_failure(&self.subscriptions_state);
286                log::warn!(
287                    "Subscription request failed: request={request:?}, result={:?}",
288                    response.result
289                );
290            }
291        }
292    }
293
294    fn handle_stream_data(
295        &self,
296        json: &serde_json::Value,
297    ) -> Option<BinanceFuturesWsStreamsMessage> {
298        let event_type = extract_event_type(json)?;
299
300        match event_type {
301            BinanceWsEventType::AggTrade => {
302                serde_json::from_value::<BinanceFuturesAggTradeMsg>(json.clone())
303                    .map(BinanceFuturesWsStreamsMessage::AggTrade)
304                    .map_err(|e| log::warn!("Failed to parse aggregate trade: {e}"))
305                    .ok()
306            }
307            BinanceWsEventType::Trade => {
308                serde_json::from_value::<BinanceFuturesTradeMsg>(json.clone())
309                    .map(BinanceFuturesWsStreamsMessage::Trade)
310                    .map_err(|e| log::warn!("Failed to parse trade: {e}"))
311                    .ok()
312            }
313            BinanceWsEventType::BookTicker => {
314                serde_json::from_value::<BinanceFuturesBookTickerMsg>(json.clone())
315                    .map(BinanceFuturesWsStreamsMessage::BookTicker)
316                    .map_err(|e| log::warn!("Failed to parse book ticker: {e}"))
317                    .ok()
318            }
319            BinanceWsEventType::DepthUpdate => {
320                serde_json::from_value::<BinanceFuturesDepthUpdateMsg>(json.clone())
321                    .map(BinanceFuturesWsStreamsMessage::DepthUpdate)
322                    .map_err(|e| log::warn!("Failed to parse depth update: {e}"))
323                    .ok()
324            }
325            BinanceWsEventType::MarkPriceUpdate => {
326                serde_json::from_value::<BinanceFuturesMarkPriceMsg>(json.clone())
327                    .map(BinanceFuturesWsStreamsMessage::MarkPrice)
328                    .map_err(|e| log::warn!("Failed to parse mark price: {e}"))
329                    .ok()
330            }
331            BinanceWsEventType::Kline => {
332                serde_json::from_value::<BinanceFuturesKlineMsg>(json.clone())
333                    .map(BinanceFuturesWsStreamsMessage::Kline)
334                    .map_err(|e| log::warn!("Failed to parse kline: {e}"))
335                    .ok()
336            }
337            BinanceWsEventType::ForceOrder => {
338                serde_json::from_value::<BinanceFuturesLiquidationMsg>(json.clone())
339                    .map(BinanceFuturesWsStreamsMessage::ForceOrder)
340                    .map_err(|e| log::warn!("Failed to parse force order: {e}"))
341                    .ok()
342            }
343            BinanceWsEventType::Ticker24Hr => {
344                serde_json::from_value::<BinanceFuturesTickerMsg>(json.clone())
345                    .map(BinanceFuturesWsStreamsMessage::Ticker)
346                    .map_err(|e| log::warn!("Failed to parse ticker: {e}"))
347                    .ok()
348            }
349            BinanceWsEventType::MiniTicker24Hr => {
350                log::debug!("Mini ticker not yet supported, skipping");
351                None
352            }
353            BinanceWsEventType::AccountUpdate => {
354                serde_json::from_value::<BinanceFuturesAccountUpdateMsg>(json.clone())
355                    .map(|msg| {
356                        log::debug!(
357                            "Account update: reason={:?}, balances={}, positions={}",
358                            msg.account.reason,
359                            msg.account.balances.len(),
360                            msg.account.positions.len()
361                        );
362                        BinanceFuturesWsStreamsMessage::AccountUpdate(msg)
363                    })
364                    .map_err(|e| log::warn!("Failed to parse account update: {e}"))
365                    .ok()
366            }
367            BinanceWsEventType::OrderTradeUpdate => {
368                serde_json::from_value::<BinanceFuturesOrderUpdateMsg>(json.clone())
369                    .map(|msg| {
370                        log::debug!(
371                            "Order update: symbol={}, order_id={}, exec={:?}, status={:?}",
372                            msg.order.symbol,
373                            msg.order.order_id,
374                            msg.order.execution_type,
375                            msg.order.order_status
376                        );
377                        BinanceFuturesWsStreamsMessage::OrderUpdate(Box::new(msg))
378                    })
379                    .map_err(|e| log::warn!("Failed to parse order update: {e}"))
380                    .ok()
381            }
382            BinanceWsEventType::TradeLite => {
383                serde_json::from_value::<BinanceFuturesTradeLiteMsg>(json.clone())
384                    .map(|msg| {
385                        log::debug!(
386                            "Trade lite: symbol={}, order_id={}, trade_id={}",
387                            msg.symbol,
388                            msg.order_id,
389                            msg.trade_id
390                        );
391                        BinanceFuturesWsStreamsMessage::TradeLite(Box::new(msg))
392                    })
393                    .map_err(|e| log::warn!("Failed to parse trade lite: {e}"))
394                    .ok()
395            }
396            BinanceWsEventType::AlgoUpdate => {
397                serde_json::from_value::<BinanceFuturesAlgoUpdateMsg>(json.clone())
398                    .map(|msg| {
399                        log::debug!(
400                            "Algo order update: symbol={}, algo_id={}, status={:?}",
401                            msg.algo_order.symbol,
402                            msg.algo_order.algo_id,
403                            msg.algo_order.algo_status
404                        );
405                        BinanceFuturesWsStreamsMessage::AlgoUpdate(Box::new(msg))
406                    })
407                    .map_err(|e| log::warn!("Failed to parse algo order update: {e}"))
408                    .ok()
409            }
410            BinanceWsEventType::MarginCall => {
411                serde_json::from_value::<BinanceFuturesMarginCallMsg>(json.clone())
412                    .map(|msg| {
413                        log::warn!(
414                            "Margin call: cross_wallet_balance={}, positions_at_risk={}",
415                            msg.cross_wallet_balance,
416                            msg.positions.len()
417                        );
418                        BinanceFuturesWsStreamsMessage::MarginCall(msg)
419                    })
420                    .map_err(|e| log::warn!("Failed to parse margin call: {e}"))
421                    .ok()
422            }
423            BinanceWsEventType::AccountConfigUpdate => {
424                serde_json::from_value::<BinanceFuturesAccountConfigMsg>(json.clone())
425                    .map(|msg| {
426                        if let Some(ref lc) = msg.leverage_config {
427                            log::debug!(
428                                "Account config update: symbol={}, leverage={}",
429                                lc.symbol,
430                                lc.leverage
431                            );
432                        }
433                        BinanceFuturesWsStreamsMessage::AccountConfigUpdate(msg)
434                    })
435                    .map_err(|e| log::warn!("Failed to parse account config update: {e}"))
436                    .ok()
437            }
438            BinanceWsEventType::ListenKeyExpired => {
439                if let Ok(msg) =
440                    serde_json::from_value::<BinanceFuturesListenKeyExpiredMsg>(json.clone())
441                {
442                    log::warn!("Listen key expired at {}", msg.event_time);
443                }
444                Some(BinanceFuturesWsStreamsMessage::ListenKeyExpired)
445            }
446            BinanceWsEventType::Unknown => {
447                log::warn!("Unknown event type in message: {json}");
448                None
449            }
450        }
451    }
452}
453
454#[cfg(test)]
455mod tests {
456    use rstest::rstest;
457
458    use super::*;
459
460    #[rstest]
461    #[tokio::test]
462    async fn test_subscription_intent_is_preserved_without_active_client() {
463        let signal = Arc::new(AtomicBool::new(false));
464        let request_id_counter = Arc::new(AtomicU64::new(1));
465        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
466        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
467        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
468        let subscriptions = SubscriptionState::new('@');
469        let subscribe_topic = "btcusdt@aggTrade";
470        let unsubscribe_topic = "ethusdt@aggTrade";
471        subscriptions.mark_subscribe(unsubscribe_topic);
472        subscriptions.confirm_subscribe(unsubscribe_topic);
473
474        let mut handler = BinanceFuturesDataWsFeedHandler::new(
475            signal,
476            cmd_rx,
477            raw_rx,
478            out_tx,
479            subscriptions.clone(),
480            request_id_counter,
481        );
482
483        handler
484            .send_subscribe(vec![subscribe_topic.to_string()])
485            .await;
486        handler
487            .send_unsubscribe(vec![unsubscribe_topic.to_string()])
488            .await;
489
490        assert_eq!(
491            subscriptions.pending_subscribe_topics(),
492            [subscribe_topic.to_string()]
493        );
494        assert_eq!(
495            subscriptions.pending_unsubscribe_topics(),
496            [unsubscribe_topic.to_string()]
497        );
498        assert_eq!(subscriptions.len(), 0);
499        assert_eq!(handler.pending_requests.len(), 0);
500    }
501
502    #[rstest]
503    fn test_error_responses_preserve_subscription_intent() {
504        let signal = Arc::new(AtomicBool::new(false));
505        let request_id_counter = Arc::new(AtomicU64::new(3));
506        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
507        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
508        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
509        let subscriptions = SubscriptionState::new('@');
510        let subscribe_topic = "btcusdt@aggTrade";
511        let unsubscribe_topic = "ethusdt@aggTrade";
512        subscriptions.mark_subscribe(subscribe_topic);
513        subscriptions.mark_subscribe(unsubscribe_topic);
514        subscriptions.confirm_subscribe(unsubscribe_topic);
515        subscriptions.mark_unsubscribe(unsubscribe_topic);
516
517        let mut handler = BinanceFuturesDataWsFeedHandler::new(
518            signal,
519            cmd_rx,
520            raw_rx,
521            out_tx,
522            subscriptions.clone(),
523            request_id_counter,
524        );
525        handler.pending_requests.insert(
526            1,
527            PendingSubscriptionRequest::subscribe(vec![subscribe_topic.to_string()]),
528        );
529        handler.pending_requests.insert(
530            2,
531            PendingSubscriptionRequest::unsubscribe(vec![unsubscribe_topic.to_string()]),
532        );
533        let error = serde_json::json!({"code": 2, "msg": "Invalid request", "id": 1});
534
535        handler.handle_subscription_response(&error);
536
537        assert_eq!(
538            subscriptions.pending_subscribe_topics(),
539            [subscribe_topic.to_string()]
540        );
541        assert_eq!(
542            subscriptions.pending_unsubscribe_topics(),
543            [unsubscribe_topic.to_string()]
544        );
545        assert_eq!(subscriptions.len(), 0);
546        assert_eq!(handler.pending_requests.len(), 1);
547
548        let error = serde_json::json!({"code": 2, "msg": "Invalid request", "id": 2});
549        handler.handle_subscription_response(&error);
550
551        assert_eq!(
552            subscriptions.pending_subscribe_topics(),
553            [subscribe_topic.to_string()]
554        );
555        assert_eq!(
556            subscriptions.pending_unsubscribe_topics(),
557            [unsubscribe_topic.to_string()]
558        );
559        assert_eq!(subscriptions.len(), 0);
560        assert_eq!(handler.pending_requests.len(), 0);
561    }
562}
nautilus_binance/futures/websocket/streams/handler.rs

nautilus_binance/futures/websocket/streams/
handler.rs
