nautilus_binance/futures/websocket/trading/
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 Trading API message handler.
17//!
18//! The handler runs in a dedicated Tokio task as the I/O boundary between the client
19//! orchestrator and the network layer. It exclusively owns the `WebSocketClient` and
20//! processes commands from the client via an unbounded channel.
21//!
22//! Unlike the Spot handler which decodes SBE binary responses, the Futures handler
23//! works with JSON text responses throughout.
24
25use std::{
26    fmt::Debug,
27    sync::{
28        Arc,
29        atomic::{AtomicBool, Ordering},
30    },
31};
32
33use ahash::AHashMap;
34use nautilus_network::{RECONNECTED, websocket::WebSocketClient};
35use tokio_tungstenite::tungstenite::Message;
36
37use super::{
38    client::BINANCE_FUTURES_WS_RATE_LIMIT_KEY_ORDER,
39    error::{BinanceFuturesWsApiError, BinanceFuturesWsApiResult},
40    messages::{
41        BinanceFuturesWsTradingCommand, BinanceFuturesWsTradingMessage,
42        BinanceFuturesWsTradingRequest, BinanceFuturesWsTradingRequestMeta,
43        BinanceFuturesWsTradingResponse, method,
44    },
45};
46use crate::{
47    common::credential::SigningCredential,
48    futures::http::query::{
49        BinanceCancelOrderParams, BinanceModifyOrderParams, BinanceNewOrderParams,
50    },
51};
52
53/// Binance Futures WebSocket Trading API handler.
54///
55/// Runs in a dedicated Tokio task, processing commands from the client
56/// and transforming raw WebSocket JSON messages into Nautilus domain events.
57pub struct BinanceFuturesWsTradingHandler {
58    signal: Arc<AtomicBool>,
59    inner: Option<WebSocketClient>,
60    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsTradingCommand>,
61    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
62    out_tx: tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsTradingMessage>,
63    credential: Arc<SigningCredential>,
64    pending_requests: AHashMap<String, BinanceFuturesWsTradingRequestMeta>,
65}
66
67impl Debug for BinanceFuturesWsTradingHandler {
68    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69        f.debug_struct(stringify!(BinanceFuturesWsTradingHandler))
70            .field("inner", &self.inner.as_ref().map(|_| "<client>"))
71            .field(
72                "pending_requests",
73                &format!("{} pending", self.pending_requests.len()),
74            )
75            .finish_non_exhaustive()
76    }
77}
78
79impl BinanceFuturesWsTradingHandler {
80    /// Creates a new handler instance.
81    #[must_use]
82    pub fn new(
83        signal: Arc<AtomicBool>,
84        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceFuturesWsTradingCommand>,
85        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
86        out_tx: tokio::sync::mpsc::UnboundedSender<BinanceFuturesWsTradingMessage>,
87        credential: Arc<SigningCredential>,
88    ) -> Self {
89        Self {
90            signal,
91            inner: None,
92            cmd_rx,
93            raw_rx,
94            out_tx,
95            credential,
96            pending_requests: AHashMap::new(),
97        }
98    }
99
100    /// Runs the main event loop for commands and raw messages.
101    ///
102    /// Returns `false` when disconnected or the signal is set.
103    pub async fn run(&mut self) -> bool {
104        loop {
105            if self.signal.load(Ordering::Relaxed) {
106                return false;
107            }
108
109            tokio::select! {
110                Some(cmd) = self.cmd_rx.recv() => {
111                    match cmd {
112                        BinanceFuturesWsTradingCommand::SetClient(client) => {
113                            log::debug!("Handler received WebSocket client");
114                            self.inner = Some(client);
115                            self.emit(BinanceFuturesWsTradingMessage::Connected);
116                        }
117                        BinanceFuturesWsTradingCommand::Disconnect => {
118                            log::debug!("Handler disconnecting WebSocket client");
119                            self.inner = None;
120                            return false;
121                        }
122                        BinanceFuturesWsTradingCommand::PlaceOrder { id, params } => {
123                            if let Err(e) = self.handle_place_order(id.clone(), params).await {
124                                log::error!("Failed to handle place order command: {e}");
125                                self.pending_requests.remove(&id);
126                                self.emit(BinanceFuturesWsTradingMessage::RequestFailed {
127                                    request_id: id,
128                                    msg: e.to_string(),
129                                });
130                            }
131                        }
132                        BinanceFuturesWsTradingCommand::CancelOrder { id, params } => {
133                            if let Err(e) = self.handle_cancel_order(id.clone(), params).await {
134                                log::error!("Failed to handle cancel order command: {e}");
135                                self.pending_requests.remove(&id);
136                                self.emit(BinanceFuturesWsTradingMessage::RequestFailed {
137                                    request_id: id,
138                                    msg: e.to_string(),
139                                });
140                            }
141                        }
142                        BinanceFuturesWsTradingCommand::ModifyOrder { id, params } => {
143                            if let Err(e) = self.handle_modify_order(id.clone(), params).await {
144                                log::error!("Failed to handle modify order command: {e}");
145                                self.pending_requests.remove(&id);
146                                self.emit(BinanceFuturesWsTradingMessage::RequestFailed {
147                                    request_id: id,
148                                    msg: e.to_string(),
149                                });
150                            }
151                        }
152                    }
153                }
154                Some(msg) = self.raw_rx.recv() => {
155                    if let Message::Text(ref text) = msg
156                        && text.as_str() == RECONNECTED
157                    {
158                        log::debug!("Handler received reconnection signal");
159                        self.fail_pending_requests();
160                        self.emit(BinanceFuturesWsTradingMessage::Reconnected);
161                        continue;
162                    }
163
164                    self.handle_message(msg);
165                }
166                else => {
167                    return false;
168                }
169            }
170        }
171    }
172
173    fn emit(&self, msg: BinanceFuturesWsTradingMessage) {
174        if let Err(e) = self.out_tx.send(msg) {
175            log::error!("Failed to send message to output channel: {e}");
176        }
177    }
178
179    fn fail_pending_requests(&mut self) {
180        if self.pending_requests.is_empty() {
181            return;
182        }
183
184        let count = self.pending_requests.len();
185        log::warn!("Failing {count} pending requests after reconnection");
186
187        let pending = std::mem::take(&mut self.pending_requests);
188        for (request_id, _meta) in pending {
189            self.emit(BinanceFuturesWsTradingMessage::RequestFailed {
190                request_id,
191                msg: "Connection lost before response received".to_string(),
192            });
193        }
194    }
195
196    async fn handle_place_order(
197        &mut self,
198        id: String,
199        params: BinanceNewOrderParams,
200    ) -> BinanceFuturesWsApiResult<()> {
201        let params_json = serde_json::to_value(&params)
202            .map_err(|e| BinanceFuturesWsApiError::JsonError(e.to_string()))?;
203        let signed_params = self.sign_params(params_json)?;
204
205        let request = BinanceFuturesWsTradingRequest::new(&id, method::ORDER_PLACE, signed_params);
206        self.pending_requests
207            .insert(id.clone(), BinanceFuturesWsTradingRequestMeta::PlaceOrder);
208        self.send_request(request).await
209    }
210
211    async fn handle_cancel_order(
212        &mut self,
213        id: String,
214        params: BinanceCancelOrderParams,
215    ) -> BinanceFuturesWsApiResult<()> {
216        let params_json = serde_json::to_value(&params)
217            .map_err(|e| BinanceFuturesWsApiError::JsonError(e.to_string()))?;
218        let signed_params = self.sign_params(params_json)?;
219
220        let request = BinanceFuturesWsTradingRequest::new(&id, method::ORDER_CANCEL, signed_params);
221        self.pending_requests
222            .insert(id.clone(), BinanceFuturesWsTradingRequestMeta::CancelOrder);
223        self.send_request(request).await
224    }
225
226    async fn handle_modify_order(
227        &mut self,
228        id: String,
229        params: BinanceModifyOrderParams,
230    ) -> BinanceFuturesWsApiResult<()> {
231        let params_json = serde_json::to_value(&params)
232            .map_err(|e| BinanceFuturesWsApiError::JsonError(e.to_string()))?;
233        let signed_params = self.sign_params(params_json)?;
234
235        let request = BinanceFuturesWsTradingRequest::new(&id, method::ORDER_MODIFY, signed_params);
236        self.pending_requests
237            .insert(id.clone(), BinanceFuturesWsTradingRequestMeta::ModifyOrder);
238        self.send_request(request).await
239    }
240
241    fn sign_params(
242        &self,
243        mut params: serde_json::Value,
244    ) -> BinanceFuturesWsApiResult<serde_json::Value> {
245        let timestamp = std::time::SystemTime::now()
246            .duration_since(std::time::UNIX_EPOCH)
247            .map_err(|e| BinanceFuturesWsApiError::ClientError(e.to_string()))?
248            .as_millis() as i64;
249
250        if let Some(obj) = params.as_object_mut() {
251            obj.insert("timestamp".to_string(), serde_json::json!(timestamp));
252            obj.insert(
253                "apiKey".to_string(),
254                serde_json::json!(self.credential.api_key()),
255            );
256        }
257
258        let query_string = serde_urlencoded::to_string(&params)
259            .map_err(|e| BinanceFuturesWsApiError::ClientError(e.to_string()))?;
260        let signature = self.credential.sign(&query_string);
261
262        if let Some(obj) = params.as_object_mut() {
263            obj.insert("signature".to_string(), serde_json::json!(signature));
264        }
265
266        Ok(params)
267    }
268
269    async fn send_request(
270        &mut self,
271        request: BinanceFuturesWsTradingRequest,
272    ) -> BinanceFuturesWsApiResult<()> {
273        let client = self.inner.as_mut().ok_or_else(|| {
274            BinanceFuturesWsApiError::ConnectionError("WebSocket not connected".to_string())
275        })?;
276
277        let json = serde_json::to_string(&request)
278            .map_err(|e| BinanceFuturesWsApiError::JsonError(e.to_string()))?;
279
280        log::debug!(
281            "Sending Futures WS Trading API request id={} method={}",
282            request.id,
283            request.method
284        );
285
286        client
287            .send_text(
288                json,
289                Some(BINANCE_FUTURES_WS_RATE_LIMIT_KEY_ORDER.as_slice()),
290            )
291            .await
292            .map_err(|e| {
293                BinanceFuturesWsApiError::ConnectionError(format!("Failed to send request: {e}"))
294            })?;
295
296        Ok(())
297    }
298
299    fn handle_message(&mut self, msg: Message) {
300        match msg {
301            Message::Text(text) => self.handle_text_response(&text),
302            Message::Ping(_) | Message::Pong(_) => {}
303            Message::Close(frame) => {
304                log::debug!("WebSocket closed: {frame:?}");
305            }
306            Message::Binary(_) | Message::Frame(_) => {}
307        }
308    }
309
310    fn handle_text_response(&mut self, text: &str) {
311        let response: BinanceFuturesWsTradingResponse = match serde_json::from_str(text) {
312            Ok(r) => r,
313            Err(e) => {
314                log::warn!("Failed to parse WS Trading API response: {e}");
315                return;
316            }
317        };
318
319        let Some(meta) = self.pending_requests.remove(&response.id) else {
320            log::warn!("Received response for unknown request ID: {}", response.id);
321            return;
322        };
323
324        if response.status != 200 {
325            let (code, msg) = response.error.map(|e| (e.code, e.msg)).unwrap_or((
326                -1,
327                format!("Request failed with status {}", response.status),
328            ));
329            let rejection = self.create_rejection(response.id, code, msg, meta);
330            self.emit(rejection);
331            return;
332        }
333
334        let Some(result) = response.result else {
335            log::warn!(
336                "Missing result in success response for request {}",
337                response.id
338            );
339            return;
340        };
341
342        match meta {
343            BinanceFuturesWsTradingRequestMeta::PlaceOrder => {
344                match serde_json::from_value(result) {
345                    Ok(order) => {
346                        self.emit(BinanceFuturesWsTradingMessage::OrderAccepted {
347                            request_id: response.id,
348                            response: Box::new(order),
349                        });
350                    }
351                    Err(e) => {
352                        log::error!("Failed to deserialize order response: {e}");
353                        self.emit(BinanceFuturesWsTradingMessage::Error(e.to_string()));
354                    }
355                }
356            }
357            BinanceFuturesWsTradingRequestMeta::CancelOrder => match serde_json::from_value(result)
358            {
359                Ok(order) => {
360                    self.emit(BinanceFuturesWsTradingMessage::OrderCanceled {
361                        request_id: response.id,
362                        response: Box::new(order),
363                    });
364                }
365                Err(e) => {
366                    log::error!("Failed to deserialize cancel response: {e}");
367                    self.emit(BinanceFuturesWsTradingMessage::Error(e.to_string()));
368                }
369            },
370            BinanceFuturesWsTradingRequestMeta::ModifyOrder => match serde_json::from_value(result)
371            {
372                Ok(order) => {
373                    self.emit(BinanceFuturesWsTradingMessage::OrderModified {
374                        request_id: response.id,
375                        response: Box::new(order),
376                    });
377                }
378                Err(e) => {
379                    log::error!("Failed to deserialize modify response: {e}");
380                    self.emit(BinanceFuturesWsTradingMessage::Error(e.to_string()));
381                }
382            },
383        }
384    }
385
386    fn create_rejection(
387        &self,
388        request_id: String,
389        code: i32,
390        msg: String,
391        meta: BinanceFuturesWsTradingRequestMeta,
392    ) -> BinanceFuturesWsTradingMessage {
393        match meta {
394            BinanceFuturesWsTradingRequestMeta::PlaceOrder => {
395                BinanceFuturesWsTradingMessage::OrderRejected {
396                    request_id,
397                    code,
398                    msg,
399                }
400            }
401            BinanceFuturesWsTradingRequestMeta::CancelOrder => {
402                BinanceFuturesWsTradingMessage::CancelRejected {
403                    request_id,
404                    code,
405                    msg,
406                }
407            }
408            BinanceFuturesWsTradingRequestMeta::ModifyOrder => {
409                BinanceFuturesWsTradingMessage::ModifyRejected {
410                    request_id,
411                    code,
412                    msg,
413                }
414            }
415        }
416    }
417}
nautilus_binance/futures/websocket/trading/handler.rs

nautilus_binance/futures/websocket/trading/
handler.rs
