nautilus_kraken/websocket/futures/
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//! WebSocket message handler for Kraken Futures.
17
18use std::{
19    collections::VecDeque,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26use nautilus_network::{
27    RECONNECTED,
28    websocket::{SubscriptionState, WebSocketClient},
29};
30use serde::Deserialize;
31use serde_json::Value;
32use tokio_tungstenite::tungstenite::Message;
33use ustr::Ustr;
34
35use super::messages::{
36    KrakenFuturesBookDelta, KrakenFuturesBookSnapshot, KrakenFuturesChannel,
37    KrakenFuturesFillsDelta, KrakenFuturesMessageType, KrakenFuturesOpenOrdersCancel,
38    KrakenFuturesOpenOrdersDelta, KrakenFuturesTickerData, KrakenFuturesTradeData,
39    KrakenFuturesWsMessage, classify_futures_message,
40};
41use crate::common::consts::KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION;
42
43/// Commands sent from the outer client to the inner message handler.
44#[derive(Debug)]
45pub enum FuturesHandlerCommand {
46    SetClient(WebSocketClient),
47    Disconnect,
48    Subscribe { payload: String },
49    Unsubscribe { payload: String },
50    RequestChallenge { payload: String },
51}
52
53/// WebSocket message handler for Kraken Futures.
54pub struct FuturesFeedHandler {
55    signal: Arc<AtomicBool>,
56    inner: Option<WebSocketClient>,
57    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<FuturesHandlerCommand>,
58    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
59    subscriptions: SubscriptionState,
60    pending_messages: VecDeque<KrakenFuturesWsMessage>,
61}
62
63impl FuturesFeedHandler {
64    /// Creates a new [`FuturesFeedHandler`] instance.
65    pub fn new(
66        signal: Arc<AtomicBool>,
67        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<FuturesHandlerCommand>,
68        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
69        subscriptions: SubscriptionState,
70    ) -> Self {
71        Self {
72            signal,
73            inner: None,
74            cmd_rx,
75            raw_rx,
76            subscriptions,
77            pending_messages: VecDeque::new(),
78        }
79    }
80
81    pub fn is_stopped(&self) -> bool {
82        self.signal.load(Ordering::Relaxed)
83    }
84
85    fn is_subscribed(&self, channel: KrakenFuturesChannel, symbol: &Ustr) -> bool {
86        let channel_ustr = Ustr::from(channel.as_ref());
87        self.subscriptions.is_subscribed(&channel_ustr, symbol)
88    }
89
90    /// Processes messages and commands, returning when stopped or stream ends.
91    pub async fn next(&mut self) -> Option<KrakenFuturesWsMessage> {
92        if let Some(msg) = self.pending_messages.pop_front() {
93            return Some(msg);
94        }
95
96        loop {
97            tokio::select! {
98                Some(cmd) = self.cmd_rx.recv() => {
99                    match cmd {
100                        FuturesHandlerCommand::SetClient(client) => {
101                            log::debug!("WebSocketClient received by futures handler");
102                            self.inner = Some(client);
103                        }
104                        FuturesHandlerCommand::Disconnect => {
105                            log::debug!("Disconnect command received");
106
107                            if let Some(client) = self.inner.take() {
108                                client.disconnect().await;
109                            }
110                            return None;
111                        }
112                        FuturesHandlerCommand::Subscribe { payload }
113                        | FuturesHandlerCommand::Unsubscribe { payload } => {
114                            if let Some(ref client) = self.inner
115                                && let Err(e) = client.send_text(payload, Some(KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice())).await
116                            {
117                                log::error!("Failed to send text: {e}");
118                            }
119                        }
120                        FuturesHandlerCommand::RequestChallenge { payload } => {
121                            if let Some(ref client) = self.inner
122                                && let Err(e) = client.send_text(payload, None).await
123                            {
124                                log::error!("Failed to send text: {e}");
125                            }
126                        }
127                    }
128                }
129
130                msg = self.raw_rx.recv() => {
131                    let msg = match msg {
132                        Some(msg) => msg,
133                        None => {
134                            log::debug!("WebSocket stream closed");
135                            return None;
136                        }
137                    };
138
139                    if self.signal.load(Ordering::Relaxed) {
140                        log::debug!("Stop signal received");
141                        return None;
142                    }
143
144                    match &msg {
145                        Message::Ping(data) => {
146                            let len = data.len();
147                            log::trace!("Received ping frame with {len} bytes");
148
149                            if let Some(client) = &self.inner
150                                && let Err(e) = client.send_pong(data.to_vec()).await
151                            {
152                                log::warn!("Failed to send pong frame: {e}");
153                            }
154                            continue;
155                        }
156                        Message::Pong(_) => {
157                            log::debug!("Received pong from server");
158                            continue;
159                        }
160                        Message::Close(_) => {
161                            log::debug!("WebSocket connection closed");
162                            return None;
163                        }
164                        Message::Frame(_) => {
165                            log::trace!("Received raw frame");
166                            continue;
167                        }
168                        _ => {}
169                    }
170
171                    let text: &str = match &msg {
172                        Message::Text(text) => text,
173                        Message::Binary(data) => match std::str::from_utf8(data) {
174                            Ok(s) => s,
175                            Err(_) => continue,
176                        },
177                        _ => continue,
178                    };
179
180                    if text == RECONNECTED {
181                        log::debug!("Received WebSocket reconnected signal");
182                        return Some(KrakenFuturesWsMessage::Reconnected);
183                    }
184
185                    self.parse_message(text);
186
187                    if let Some(msg) = self.pending_messages.pop_front() {
188                        return Some(msg);
189                    }
190                }
191            }
192        }
193    }
194
195    fn parse_message(&mut self, text: &str) {
196        let value: Value = match serde_json::from_str(text) {
197            Ok(v) => v,
198            Err(e) => {
199                log::debug!("Failed to parse message as JSON: {e}");
200                return;
201            }
202        };
203
204        match classify_futures_message(&value) {
205            KrakenFuturesMessageType::OpenOrdersSnapshot => {
206                log::debug!(
207                    "Skipping open_orders_snapshot (REST reconciliation handles initial state)"
208                );
209            }
210            KrakenFuturesMessageType::OpenOrdersCancel => {
211                self.handle_open_orders_cancel_value(value);
212            }
213            KrakenFuturesMessageType::OpenOrdersDelta => {
214                self.handle_open_orders_delta_value(value);
215            }
216            KrakenFuturesMessageType::FillsSnapshot => {
217                log::debug!("Skipping fills_snapshot (REST reconciliation handles initial state)");
218            }
219            KrakenFuturesMessageType::FillsDelta => {
220                self.handle_fills_delta_value(value);
221            }
222            KrakenFuturesMessageType::Ticker => {
223                self.handle_ticker_message_value(value);
224            }
225            KrakenFuturesMessageType::TradeSnapshot => {
226                log::debug!("Skipping trade_snapshot (only streaming live trades)");
227            }
228            KrakenFuturesMessageType::Trade => {
229                self.handle_trade_message_value(value);
230            }
231            KrakenFuturesMessageType::BookSnapshot => {
232                self.handle_book_snapshot_value(value);
233            }
234            KrakenFuturesMessageType::BookDelta => {
235                self.handle_book_delta_value(value);
236            }
237            KrakenFuturesMessageType::Info => {
238                log::debug!("Received info message: {text}");
239            }
240            KrakenFuturesMessageType::Pong => {
241                log::debug!("Received text pong response");
242            }
243            KrakenFuturesMessageType::Subscribed => {
244                log::debug!("Subscription confirmed: {text}");
245            }
246            KrakenFuturesMessageType::Unsubscribed => {
247                log::debug!("Unsubscription confirmed: {text}");
248            }
249            KrakenFuturesMessageType::Challenge => {
250                self.handle_challenge_response_value(value);
251            }
252            KrakenFuturesMessageType::Heartbeat => {
253                log::trace!("Heartbeat received");
254            }
255            KrakenFuturesMessageType::Error => {
256                let message = value
257                    .get("message")
258                    .and_then(|v| v.as_str())
259                    .unwrap_or("Unknown error");
260                log::warn!("Kraken Futures WebSocket error: {message}");
261            }
262            KrakenFuturesMessageType::Alert => {
263                let message = value
264                    .get("message")
265                    .and_then(|v| v.as_str())
266                    .unwrap_or("Unknown alert");
267                log::warn!("Kraken Futures WebSocket alert: {message}");
268            }
269            KrakenFuturesMessageType::Unknown => {
270                log::warn!("Unhandled futures message: {text}");
271            }
272        }
273    }
274
275    fn handle_challenge_response_value(&mut self, value: Value) {
276        #[derive(Deserialize)]
277        struct ChallengeResponse {
278            message: String,
279        }
280
281        match serde_json::from_value::<ChallengeResponse>(value) {
282            Ok(response) => {
283                let len = response.message.len();
284                log::debug!("Challenge received, length: {len}");
285
286                self.pending_messages
287                    .push_back(KrakenFuturesWsMessage::Challenge(response.message));
288            }
289            Err(e) => {
290                log::error!("Failed to parse challenge response: {e}");
291            }
292        }
293    }
294
295    fn handle_ticker_message_value(&mut self, value: Value) {
296        let ticker = match serde_json::from_value::<KrakenFuturesTickerData>(value) {
297            Ok(t) => t,
298            Err(e) => {
299                log::debug!("Failed to parse ticker: {e}");
300                return;
301            }
302        };
303
304        self.pending_messages
305            .push_back(KrakenFuturesWsMessage::Ticker(ticker));
306    }
307
308    fn handle_trade_message_value(&mut self, value: Value) {
309        let trade = match serde_json::from_value::<KrakenFuturesTradeData>(value) {
310            Ok(t) => t,
311            Err(e) => {
312                log::warn!("Failed to parse trade: {e}");
313                return;
314            }
315        };
316
317        if !self.is_subscribed(KrakenFuturesChannel::Trades, &trade.product_id) {
318            log::debug!(
319                "Received trade for unsubscribed product: {}",
320                trade.product_id
321            );
322            return;
323        }
324
325        self.pending_messages
326            .push_back(KrakenFuturesWsMessage::Trade(trade));
327    }
328
329    fn handle_book_snapshot_value(&mut self, value: Value) {
330        let snapshot = match serde_json::from_value::<KrakenFuturesBookSnapshot>(value) {
331            Ok(s) => s,
332            Err(e) => {
333                log::warn!("Failed to parse book snapshot: {e}");
334                return;
335            }
336        };
337
338        let has_book = self.is_subscribed(KrakenFuturesChannel::Book, &snapshot.product_id);
339        let has_quotes = self.is_subscribed(KrakenFuturesChannel::Quotes, &snapshot.product_id);
340
341        if !has_book && !has_quotes {
342            log::debug!(
343                "Received book snapshot for unsubscribed product: {}",
344                snapshot.product_id
345            );
346            return;
347        }
348
349        self.pending_messages
350            .push_back(KrakenFuturesWsMessage::BookSnapshot(snapshot));
351    }
352
353    fn handle_book_delta_value(&mut self, value: Value) {
354        let delta = match serde_json::from_value::<KrakenFuturesBookDelta>(value) {
355            Ok(d) => d,
356            Err(e) => {
357                log::warn!("Failed to parse book delta: {e}");
358                return;
359            }
360        };
361
362        let has_book = self.is_subscribed(KrakenFuturesChannel::Book, &delta.product_id);
363        let has_quotes = self.is_subscribed(KrakenFuturesChannel::Quotes, &delta.product_id);
364
365        if !has_book && !has_quotes {
366            log::debug!(
367                "Received book delta for unsubscribed product: {}",
368                delta.product_id
369            );
370            return;
371        }
372
373        self.pending_messages
374            .push_back(KrakenFuturesWsMessage::BookDelta(delta));
375    }
376
377    fn handle_open_orders_delta_value(&mut self, value: Value) {
378        let delta = match serde_json::from_value::<KrakenFuturesOpenOrdersDelta>(value) {
379            Ok(d) => d,
380            Err(e) => {
381                log::error!("Failed to parse open_orders delta: {e}");
382                return;
383            }
384        };
385
386        log::debug!(
387            "Received open_orders delta: order_id={}, is_cancel={}, reason={:?}",
388            delta.order.order_id,
389            delta.is_cancel,
390            delta.reason
391        );
392
393        self.pending_messages
394            .push_back(KrakenFuturesWsMessage::OpenOrdersDelta(delta));
395    }
396
397    fn handle_open_orders_cancel_value(&mut self, value: Value) {
398        let cancel = match serde_json::from_value::<KrakenFuturesOpenOrdersCancel>(value) {
399            Ok(c) => c,
400            Err(e) => {
401                log::error!("Failed to parse open_orders cancel: {e}");
402                return;
403            }
404        };
405
406        log::debug!(
407            "Received open_orders cancel: order_id={}, cli_ord_id={:?}, reason={:?}",
408            cancel.order_id,
409            cancel.cli_ord_id,
410            cancel.reason
411        );
412
413        self.pending_messages
414            .push_back(KrakenFuturesWsMessage::OpenOrdersCancel(cancel));
415    }
416
417    fn handle_fills_delta_value(&mut self, value: Value) {
418        let delta = match serde_json::from_value::<KrakenFuturesFillsDelta>(value) {
419            Ok(d) => d,
420            Err(e) => {
421                log::error!("Failed to parse fills delta: {e}");
422                return;
423            }
424        };
425
426        log::debug!("Received fills delta: fill_count={}", delta.fills.len());
427
428        self.pending_messages
429            .push_back(KrakenFuturesWsMessage::FillsDelta(delta));
430    }
431}
432
433#[cfg(test)]
434mod tests {
435    use rstest::rstest;
436
437    use super::*;
438
439    fn create_test_handler() -> FuturesFeedHandler {
440        let signal = Arc::new(AtomicBool::new(false));
441        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
442        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
443        let subscriptions = SubscriptionState::new(':');
444
445        FuturesFeedHandler::new(signal, cmd_rx, raw_rx, subscriptions)
446    }
447
448    #[rstest]
449    fn test_parse_ticker_emits_ticker_message() {
450        let mut handler = create_test_handler();
451        let json = include_str!("../../../test_data/ws_futures_ticker.json");
452
453        handler.parse_message(json);
454
455        assert_eq!(handler.pending_messages.len(), 1);
456        let msg = handler.pending_messages.pop_front().unwrap();
457        let KrakenFuturesWsMessage::Ticker(ticker) = msg else {
458            panic!("Expected Ticker message, was {msg:?}");
459        };
460        assert_eq!(ticker.product_id, Ustr::from("PI_XBTUSD"));
461        assert_eq!(ticker.bid, Some(21978.5));
462        assert_eq!(ticker.ask, Some(21987.0));
463    }
464
465    #[rstest]
466    fn test_parse_trade_emits_trade_message() {
467        let mut handler = create_test_handler();
468        handler.subscriptions.mark_subscribe("trades:PI_XBTUSD");
469        handler.subscriptions.confirm_subscribe("trades:PI_XBTUSD");
470
471        let json = include_str!("../../../test_data/ws_futures_trade.json");
472
473        handler.parse_message(json);
474
475        assert_eq!(handler.pending_messages.len(), 1);
476        let msg = handler.pending_messages.pop_front().unwrap();
477        let KrakenFuturesWsMessage::Trade(trade) = msg else {
478            panic!("Expected Trade message, was {msg:?}");
479        };
480        assert_eq!(trade.product_id, Ustr::from("PI_XBTUSD"));
481        assert_eq!(trade.price, 34969.5);
482        assert_eq!(trade.qty, 15000.0);
483    }
484
485    #[rstest]
486    fn test_parse_trade_filters_unsubscribed() {
487        let mut handler = create_test_handler();
488        let json = include_str!("../../../test_data/ws_futures_trade.json");
489
490        handler.parse_message(json);
491
492        assert!(
493            handler.pending_messages.is_empty(),
494            "Trade for unsubscribed product should be filtered"
495        );
496    }
497
498    #[rstest]
499    fn test_parse_book_snapshot_emits_book_snapshot() {
500        let mut handler = create_test_handler();
501        handler.subscriptions.mark_subscribe("book:PI_XBTUSD");
502        handler.subscriptions.confirm_subscribe("book:PI_XBTUSD");
503
504        let json = include_str!("../../../test_data/ws_futures_book_snapshot.json");
505
506        handler.parse_message(json);
507
508        assert_eq!(handler.pending_messages.len(), 1);
509        let msg = handler.pending_messages.pop_front().unwrap();
510        let KrakenFuturesWsMessage::BookSnapshot(snapshot) = msg else {
511            panic!("Expected BookSnapshot message, was {msg:?}");
512        };
513        assert_eq!(snapshot.product_id, Ustr::from("PI_XBTUSD"));
514        assert_eq!(snapshot.bids.len(), 2);
515        assert_eq!(snapshot.asks.len(), 2);
516    }
517
518    #[rstest]
519    fn test_parse_book_snapshot_filters_unsubscribed() {
520        let mut handler = create_test_handler();
521        let json = include_str!("../../../test_data/ws_futures_book_snapshot.json");
522
523        handler.parse_message(json);
524
525        assert!(
526            handler.pending_messages.is_empty(),
527            "Book snapshot for unsubscribed product should be filtered"
528        );
529    }
530
531    #[rstest]
532    fn test_parse_book_delta_emits_book_delta() {
533        let mut handler = create_test_handler();
534        handler.subscriptions.mark_subscribe("book:PI_XBTUSD");
535        handler.subscriptions.confirm_subscribe("book:PI_XBTUSD");
536
537        let json = include_str!("../../../test_data/ws_futures_book_delta.json");
538
539        handler.parse_message(json);
540
541        assert_eq!(handler.pending_messages.len(), 1);
542        let msg = handler.pending_messages.pop_front().unwrap();
543        let KrakenFuturesWsMessage::BookDelta(delta) = msg else {
544            panic!("Expected BookDelta message, was {msg:?}");
545        };
546        assert_eq!(delta.product_id, Ustr::from("PI_XBTUSD"));
547        assert_eq!(delta.price, 34981.0);
548    }
549
550    #[rstest]
551    fn test_parse_book_delta_filters_unsubscribed() {
552        let mut handler = create_test_handler();
553        let json = include_str!("../../../test_data/ws_futures_book_delta.json");
554
555        handler.parse_message(json);
556
557        assert!(
558            handler.pending_messages.is_empty(),
559            "Book delta for unsubscribed product should be filtered"
560        );
561    }
562
563    #[rstest]
564    fn test_parse_open_orders_cancel_emits_cancel() {
565        let mut handler = create_test_handler();
566        let json = include_str!("../../../test_data/ws_futures_open_orders_cancel.json");
567
568        handler.parse_message(json);
569
570        assert_eq!(handler.pending_messages.len(), 1);
571        let msg = handler.pending_messages.pop_front().unwrap();
572        let KrakenFuturesWsMessage::OpenOrdersCancel(cancel) = msg else {
573            panic!("Expected OpenOrdersCancel message, was {msg:?}");
574        };
575        assert_eq!(cancel.order_id, "660c6b23-8007-48c1-a7c9-4893f4572e8c");
576        assert!(cancel.is_cancel);
577    }
578
579    #[rstest]
580    fn test_parse_open_orders_delta_emits_delta() {
581        let mut handler = create_test_handler();
582        let json = include_str!("../../../test_data/ws_futures_open_orders_delta.json");
583
584        handler.parse_message(json);
585
586        assert_eq!(handler.pending_messages.len(), 1);
587        let msg = handler.pending_messages.pop_front().unwrap();
588        let KrakenFuturesWsMessage::OpenOrdersDelta(delta) = msg else {
589            panic!("Expected OpenOrdersDelta message, was {msg:?}");
590        };
591        assert_eq!(delta.order.instrument, Ustr::from("PI_XBTUSD"));
592        assert!(!delta.is_cancel);
593    }
594
595    #[rstest]
596    fn test_parse_fills_delta_emits_fills() {
597        let mut handler = create_test_handler();
598        let json = include_str!("../../../test_data/ws_futures_fills_delta.json");
599
600        handler.parse_message(json);
601
602        assert_eq!(handler.pending_messages.len(), 1);
603        let msg = handler.pending_messages.pop_front().unwrap();
604        let KrakenFuturesWsMessage::FillsDelta(fills) = msg else {
605            panic!("Expected FillsDelta message, was {msg:?}");
606        };
607        assert_eq!(fills.fills.len(), 1);
608        assert_eq!(
609            fills.fills[0].fill_id,
610            "6a22a3fb-e18e-4e76-b841-8689735c9158"
611        );
612    }
613
614    #[rstest]
615    fn test_parse_challenge_emits_challenge_message() {
616        let mut handler = create_test_handler();
617        let json = r#"{"event":"challenge","message":"server-challenge-abc"}"#;
618
619        handler.parse_message(json);
620
621        assert_eq!(handler.pending_messages.len(), 1);
622        let msg = handler.pending_messages.pop_front().unwrap();
623        let KrakenFuturesWsMessage::Challenge(challenge) = msg else {
624            panic!("Expected Challenge message, was {msg:?}");
625        };
626        assert_eq!(challenge, "server-challenge-abc");
627    }
628
629    #[rstest]
630    fn test_heartbeat_produces_no_message() {
631        let mut handler = create_test_handler();
632        let json = r#"{"feed":"heartbeat","time":1700000000000}"#;
633
634        handler.parse_message(json);
635
636        assert!(handler.pending_messages.is_empty());
637    }
638
639    #[rstest]
640    fn test_info_event_produces_no_message() {
641        let mut handler = create_test_handler();
642        let json = r#"{"event":"info","version":1}"#;
643
644        handler.parse_message(json);
645
646        assert!(handler.pending_messages.is_empty());
647    }
648}
nautilus_kraken/websocket/futures/handler.rs

nautilus_kraken/websocket/futures/
handler.rs
