nautilus_kraken/websocket/spot_v2/
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 Spot v2.
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, value::RawValue};
32use tokio_tungstenite::tungstenite::Message;
33
34use super::{
35    enums::{KrakenWsChannel, KrakenWsMessageType},
36    messages::{
37        KrakenSpotWsMessage, KrakenWsBookData, KrakenWsExecutionData, KrakenWsMessage,
38        KrakenWsOhlcData, KrakenWsResponse, KrakenWsTickerData, KrakenWsTradeData,
39    },
40    parse::parse_order_response,
41};
42use crate::{
43    common::consts::{KRAKEN_RATE_LIMIT_KEY_ORDER, KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION},
44    websocket::spot_v2::level_3::messages::{KrakenL3Snapshot, KrakenL3UpdateData},
45};
46
47/// Commands sent from the outer client to the inner message handler.
48#[derive(Debug)]
49pub enum SpotHandlerCommand {
50    SetClient(WebSocketClient),
51    Disconnect,
52    Subscribe { payload: String },
53    Unsubscribe { payload: String },
54    Ping { payload: String },
55    SendOrderRequest { req_id: u64, payload: String },
56}
57
58/// WebSocket message handler for Kraken Spot v2.
59pub(super) struct SpotFeedHandler {
60    signal: Arc<AtomicBool>,
61    inner: Option<WebSocketClient>,
62    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<SpotHandlerCommand>,
63    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
64    subscriptions: SubscriptionState,
65    pending_messages: VecDeque<KrakenSpotWsMessage>,
66}
67
68impl SpotFeedHandler {
69    /// Creates a new [`SpotFeedHandler`] instance.
70    pub(super) fn new(
71        signal: Arc<AtomicBool>,
72        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<SpotHandlerCommand>,
73        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
74        subscriptions: SubscriptionState,
75    ) -> Self {
76        Self {
77            signal,
78            inner: None,
79            cmd_rx,
80            raw_rx,
81            subscriptions,
82            pending_messages: VecDeque::new(),
83        }
84    }
85
86    pub(super) fn is_stopped(&self) -> bool {
87        self.signal.load(Ordering::Relaxed)
88    }
89
90    fn is_subscribed(&self, topic: &str) -> bool {
91        self.subscriptions.all_topics().iter().any(|t| t == topic)
92    }
93
94    /// Processes messages and commands, returning when stopped or stream ends.
95    pub(super) async fn next(&mut self) -> Option<KrakenSpotWsMessage> {
96        if let Some(msg) = self.pending_messages.pop_front() {
97            return Some(msg);
98        }
99
100        loop {
101            tokio::select! {
102                Some(cmd) = self.cmd_rx.recv() => {
103                    match cmd {
104                        SpotHandlerCommand::SetClient(client) => {
105                            log::debug!("WebSocketClient received by handler");
106                            self.inner = Some(client);
107                        }
108                        SpotHandlerCommand::Disconnect => {
109                            log::debug!("Disconnect command received");
110
111                            if let Some(client) = self.inner.take() {
112                                client.disconnect().await;
113                            }
114                        }
115                        SpotHandlerCommand::Subscribe { payload }
116                        | SpotHandlerCommand::Unsubscribe { payload } => {
117                            if let Some(client) = &self.inner
118                                && let Err(e) = client.send_text(payload, Some(KRAKEN_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice())).await
119                            {
120                                log::error!("Failed to send text: {e}");
121                            }
122                        }
123                        SpotHandlerCommand::Ping { payload } => {
124                            if let Some(client) = &self.inner
125                                && let Err(e) = client.send_text(payload, None).await
126                            {
127                                log::error!("Failed to send text: {e}");
128                            }
129                        }
130                        SpotHandlerCommand::SendOrderRequest { req_id, payload } => {
131                            if let Some(client) = &self.inner {
132                                if let Err(e) = client.send_text(payload, Some(KRAKEN_RATE_LIMIT_KEY_ORDER.as_slice())).await {
133                                    log::error!(
134                                        "Kraken WS send_order_request failed req_id={req_id}: {e}"
135                                    );
136                                } else {
137                                    log::debug!(
138                                        "Kraken WS send_order_request enqueued req_id={req_id}"
139                                    );
140                                }
141                            } else {
142                                log::error!(
143                                    "Kraken WS send_order_request without active client req_id={req_id}"
144                                );
145                            }
146                        }
147                    }
148                }
149
150                msg = self.raw_rx.recv() => {
151                    let msg = match msg {
152                        Some(msg) => msg,
153                        None => {
154                            log::debug!("WebSocket stream closed");
155                            return None;
156                        }
157                    };
158
159                    if let Message::Ping(data) = &msg {
160                        log::trace!("Received ping frame with {} bytes", data.len());
161
162                        if let Some(client) = &self.inner
163                            && let Err(e) = client.send_pong(data.to_vec()).await
164                        {
165                            log::warn!("Failed to send pong frame: {e}");
166                        }
167                        continue;
168                    }
169
170                    if self.signal.load(Ordering::Relaxed) {
171                        log::debug!("Stop signal received");
172                        return None;
173                    }
174
175                    let text = match msg {
176                        Message::Text(text) => text.to_string(),
177                        Message::Binary(data) => {
178                            match String::from_utf8(data.to_vec()) {
179                                Ok(text) => text,
180                                Err(e) => {
181                                    log::warn!("Failed to decode binary message: {e}");
182                                    continue;
183                                }
184                            }
185                        }
186                        Message::Pong(_) => {
187                            log::trace!("Received pong");
188                            continue;
189                        }
190                        Message::Close(_) => {
191                            log::debug!("WebSocket connection closed");
192                            return None;
193                        }
194                        Message::Frame(_) => {
195                            log::trace!("Received raw frame");
196                            continue;
197                        }
198                        _ => continue,
199                    };
200
201                    if text == RECONNECTED {
202                        log::debug!("Received WebSocket reconnected signal");
203                        return Some(KrakenSpotWsMessage::Reconnected);
204                    }
205
206                    if let Some(msg) = self.parse_message(&text) {
207                        return Some(msg);
208                    }
209                }
210            }
211        }
212    }
213
214    fn parse_message(&self, text: &str) -> Option<KrakenSpotWsMessage> {
215        // Fast pre-filter for high-frequency control messages (no JSON parsing)
216        if text.len() < 50 && text.starts_with("{\"channel\":\"") {
217            if text.contains("heartbeat") {
218                log::trace!("Received heartbeat");
219                return None;
220            }
221
222            if text.contains("status") {
223                log::debug!("Received status message");
224                return None;
225            }
226        }
227
228        if text.contains("\"level3\"")
229            && let Some(msg) = parse_level3_text(text)
230        {
231            return self.handle_l3_message(msg);
232        }
233
234        let value: Value = match serde_json::from_str(text) {
235            Ok(v) => v,
236            Err(e) => {
237                log::warn!("Failed to parse message: {e}");
238                return None;
239            }
240        };
241
242        if value.get("method").is_some() {
243            match parse_order_response(text) {
244                Ok(Some(msg)) => return Some(msg),
245                Ok(None) => {}
246                Err(e) => log::warn!("Failed to parse order response: {e}"),
247            }
248            self.handle_control_message(value);
249            return None;
250        }
251
252        if value.get("channel").is_some() && value.get("data").is_some() {
253            match serde_json::from_value::<KrakenWsMessage>(value) {
254                Ok(msg) => return self.handle_data_message(msg),
255                Err(e) => {
256                    log::debug!("Failed to parse data message: {e}");
257                    return None;
258                }
259            }
260        }
261
262        log::debug!("Unhandled message structure: {text}");
263        None
264    }
265
266    fn handle_control_message(&self, value: Value) {
267        match serde_json::from_value::<KrakenWsResponse>(value) {
268            Ok(response) => match response {
269                KrakenWsResponse::Subscribe(sub) => {
270                    if sub.success {
271                        if let Some(result) = &sub.result {
272                            log::debug!(
273                                "Subscription confirmed: channel={:?}, req_id={:?}",
274                                result.channel,
275                                sub.req_id
276                            );
277                        } else {
278                            log::debug!("Subscription confirmed: req_id={:?}", sub.req_id);
279                        }
280                    } else {
281                        log::warn!(
282                            "Subscription failed: error={:?}, req_id={:?}",
283                            sub.error,
284                            sub.req_id
285                        );
286                    }
287                }
288                KrakenWsResponse::Unsubscribe(unsub) => {
289                    if unsub.success {
290                        log::debug!("Unsubscription confirmed: req_id={:?}", unsub.req_id);
291                    } else {
292                        log::warn!(
293                            "Unsubscription failed: error={:?}, req_id={:?}",
294                            unsub.error,
295                            unsub.req_id
296                        );
297                    }
298                }
299                KrakenWsResponse::Pong(pong) => {
300                    log::trace!("Received pong: req_id={:?}", pong.req_id);
301                }
302                KrakenWsResponse::Other => {
303                    log::debug!("Received unknown control response");
304                }
305            },
306            Err(_) => {
307                log::debug!("Received control message (failed to parse details)");
308            }
309        }
310    }
311
312    fn handle_data_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
313        match msg.channel {
314            KrakenWsChannel::Book => self.handle_book_message(msg),
315            KrakenWsChannel::Ticker => self.handle_ticker_message(msg),
316            KrakenWsChannel::Trade => self.handle_trade_message(msg),
317            KrakenWsChannel::Ohlc => self.handle_ohlc_message(msg),
318            KrakenWsChannel::Executions => self.handle_executions_message(msg),
319            KrakenWsChannel::Level3 => {
320                unreachable!("level3 messages routed via fast-path in parse_message",)
321            }
322            _ => {
323                log::warn!("Unhandled channel: {:?}", msg.channel);
324                None
325            }
326        }
327    }
328
329    fn handle_book_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
330        let is_snapshot = msg.event_type == KrakenWsMessageType::Snapshot;
331        let mut book_data = Vec::new();
332
333        for data in msg.data {
334            match serde_json::from_value::<KrakenWsBookData>(data) {
335                Ok(bd) => {
336                    if !self.is_subscribed(&format!("book:{}", bd.symbol)) {
337                        continue;
338                    }
339                    book_data.push(bd);
340                }
341                Err(e) => log::error!("Failed to deserialize book data: {e}"),
342            }
343        }
344
345        if book_data.is_empty() {
346            None
347        } else {
348            Some(KrakenSpotWsMessage::Book {
349                data: book_data,
350                is_snapshot,
351            })
352        }
353    }
354
355    fn handle_ticker_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
356        let mut tickers = Vec::new();
357
358        for data in msg.data {
359            match serde_json::from_value::<KrakenWsTickerData>(data) {
360                Ok(td) => {
361                    let symbol = &td.symbol;
362                    let quotes_key = format!("quotes:{symbol}");
363                    let ticker_key = format!("ticker:{symbol}");
364                    if !self.is_subscribed(&quotes_key) && !self.is_subscribed(&ticker_key) {
365                        continue;
366                    }
367                    tickers.push(td);
368                }
369                Err(e) => log::error!("Failed to deserialize ticker data: {e}"),
370            }
371        }
372
373        if tickers.is_empty() {
374            None
375        } else {
376            Some(KrakenSpotWsMessage::Ticker(tickers))
377        }
378    }
379
380    fn handle_trade_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
381        let mut trades = Vec::new();
382
383        for data in msg.data {
384            match serde_json::from_value::<KrakenWsTradeData>(data) {
385                Ok(td) => trades.push(td),
386                Err(e) => log::error!("Failed to deserialize trade data: {e}"),
387            }
388        }
389
390        if trades.is_empty() {
391            None
392        } else {
393            Some(KrakenSpotWsMessage::Trade(trades))
394        }
395    }
396
397    fn handle_ohlc_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
398        let mut ohlc_data = Vec::new();
399
400        for data in msg.data {
401            match serde_json::from_value::<KrakenWsOhlcData>(data) {
402                Ok(od) => ohlc_data.push(od),
403                Err(e) => log::error!("Failed to deserialize OHLC data: {e}"),
404            }
405        }
406
407        if ohlc_data.is_empty() {
408            None
409        } else {
410            Some(KrakenSpotWsMessage::Ohlc(ohlc_data))
411        }
412    }
413
414    fn handle_executions_message(&self, msg: KrakenWsMessage) -> Option<KrakenSpotWsMessage> {
415        let mut executions = Vec::new();
416
417        for data in msg.data {
418            match serde_json::from_value::<KrakenWsExecutionData>(data) {
419                Ok(ed) => executions.push(ed),
420                Err(e) => log::error!("Failed to deserialize execution data: {e}"),
421            }
422        }
423
424        if executions.is_empty() {
425            None
426        } else {
427            Some(KrakenSpotWsMessage::Execution(executions))
428        }
429    }
430}
431
432#[derive(Deserialize)]
433struct Level3RawMessage<'a> {
434    channel: &'a str,
435    #[serde(rename = "type")]
436    msg_type: &'a str,
437    #[serde(borrow)]
438    data: Vec<&'a RawValue>,
439}
440
441fn parse_level3_text(text: &str) -> Option<KrakenSpotWsMessage> {
442    let msg: Level3RawMessage<'_> = serde_json::from_str(text).ok()?;
443    if msg.channel != "level3" {
444        return None;
445    }
446    let first = msg.data.first()?.get();
447
448    match msg.msg_type {
449        "snapshot" => match serde_json::from_str::<KrakenL3Snapshot>(first) {
450            Ok(snap) => Some(KrakenSpotWsMessage::L3Snapshot(snap)),
451            Err(e) => {
452                log::warn!("Failed to deserialize L3 snapshot: {e}");
453                None
454            }
455        },
456        "update" => match serde_json::from_str::<KrakenL3UpdateData>(first) {
457            Ok(update) => Some(KrakenSpotWsMessage::L3Update(update)),
458            Err(e) => {
459                log::warn!("Failed to deserialize L3 update: {e}");
460                None
461            }
462        },
463        _ => None,
464    }
465}
466
467impl SpotFeedHandler {
468    fn handle_l3_message(&self, msg: KrakenSpotWsMessage) -> Option<KrakenSpotWsMessage> {
469        let symbol = match &msg {
470            KrakenSpotWsMessage::L3Snapshot(s) => &s.symbol,
471            KrakenSpotWsMessage::L3Update(u) => &u.symbol,
472            _ => return None,
473        };
474
475        if !self.is_subscribed(&format!("level3:{symbol}")) {
476            return None;
477        }
478        Some(msg)
479    }
480}
481
482#[cfg(test)]
483mod tests {
484    use rstest::rstest;
485
486    use super::*;
487
488    fn create_test_handler() -> SpotFeedHandler {
489        let signal = Arc::new(AtomicBool::new(false));
490        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
491        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
492        let subscriptions = SubscriptionState::new(':');
493
494        SpotFeedHandler::new(signal, cmd_rx, raw_rx, subscriptions)
495    }
496
497    #[rstest]
498    fn test_ticker_message_filtered_without_quotes_subscription() {
499        let handler = create_test_handler();
500
501        let json = r#"{
502            "channel": "ticker",
503            "type": "snapshot",
504            "data": [{
505                "symbol": "BTC/USD",
506                "bid": 105944.20,
507                "bid_qty": 2.5,
508                "ask": 105944.30,
509                "ask_qty": 3.2,
510                "last": 105899.40,
511                "volume": 163.28908096,
512                "vwap": 105904.39279,
513                "low": 104711.00,
514                "high": 106613.10,
515                "change": 250.00,
516                "change_pct": 0.24,
517                "timestamp": "2022-12-25T09:30:59.123456Z"
518            }]
519        }"#;
520
521        let result = handler.parse_message(json);
522        assert!(
523            result.is_none(),
524            "Ticker message should be filtered when no quotes subscription exists"
525        );
526    }
527
528    #[rstest]
529    fn test_ticker_message_passes_with_quotes_subscription() {
530        let handler = create_test_handler();
531        handler.subscriptions.mark_subscribe("quotes:BTC/USD");
532        handler.subscriptions.confirm_subscribe("quotes:BTC/USD");
533
534        let json = r#"{
535            "channel": "ticker",
536            "type": "snapshot",
537            "data": [{
538                "symbol": "BTC/USD",
539                "bid": 105944.20,
540                "bid_qty": 2.5,
541                "ask": 105944.30,
542                "ask_qty": 3.2,
543                "last": 105899.40,
544                "volume": 163.28908096,
545                "vwap": 105904.39279,
546                "low": 104711.00,
547                "high": 106613.10,
548                "change": 250.00,
549                "change_pct": 0.24,
550                "timestamp": "2022-12-25T09:30:59.123456Z"
551            }]
552        }"#;
553
554        let result = handler.parse_message(json);
555        assert!(
556            result.is_some(),
557            "Ticker message should pass with quotes subscription"
558        );
559
560        match result.unwrap() {
561            KrakenSpotWsMessage::Ticker(data) => {
562                assert!(!data.is_empty(), "Should have ticker data");
563            }
564            _ => panic!("Expected Ticker message"),
565        }
566    }
567
568    #[rstest]
569    fn test_ticker_message_passes_with_ticker_subscription() {
570        let handler = create_test_handler();
571        handler.subscriptions.mark_subscribe("ticker:BTC/USD");
572        handler.subscriptions.confirm_subscribe("ticker:BTC/USD");
573
574        let json = r#"{
575            "channel": "ticker",
576            "type": "snapshot",
577            "data": [{
578                "symbol": "BTC/USD",
579                "bid": 105944.20,
580                "bid_qty": 2.5,
581                "ask": 105944.30,
582                "ask_qty": 3.2,
583                "last": 105899.40,
584                "volume": 163.28908096,
585                "vwap": 105904.39279,
586                "low": 104711.00,
587                "high": 106613.10,
588                "change": 250.00,
589                "change_pct": 0.24,
590                "timestamp": "2022-12-25T09:30:59.123456Z"
591            }]
592        }"#;
593
594        let result = handler.parse_message(json);
595        assert!(
596            result.is_some(),
597            "Ticker message should pass with ticker: subscription"
598        );
599
600        match result.unwrap() {
601            KrakenSpotWsMessage::Ticker(data) => {
602                assert!(!data.is_empty(), "Should have ticker data");
603            }
604            _ => panic!("Expected Ticker message"),
605        }
606    }
607
608    #[rstest]
609    fn test_book_message_filtered_without_book_subscription() {
610        let handler = create_test_handler();
611
612        let json = r#"{
613            "channel": "book",
614            "type": "snapshot",
615            "data": [{
616                "symbol": "BTC/USD",
617                "bids": [{"price": 105944.20, "qty": 2.5}],
618                "asks": [{"price": 105944.30, "qty": 3.2}],
619                "checksum": 12345,
620                "timestamp": "2023-10-06T17:35:55.440295Z"
621            }]
622        }"#;
623
624        let result = handler.parse_message(json);
625        assert!(
626            result.is_none(),
627            "Book message should be filtered when no book subscription exists"
628        );
629    }
630
631    #[rstest]
632    fn test_book_message_passes_with_book_subscription() {
633        let handler = create_test_handler();
634        handler.subscriptions.mark_subscribe("book:BTC/USD");
635        handler.subscriptions.confirm_subscribe("book:BTC/USD");
636
637        let json = r#"{
638            "channel": "book",
639            "type": "snapshot",
640            "data": [{
641                "symbol": "BTC/USD",
642                "bids": [{"price": 105944.20, "qty": 2.5}],
643                "asks": [{"price": 105944.30, "qty": 3.2}],
644                "checksum": 12345,
645                "timestamp": "2023-10-06T17:35:55.440295Z"
646            }]
647        }"#;
648
649        let result = handler.parse_message(json);
650        assert!(
651            result.is_some(),
652            "Book message should pass with book subscription"
653        );
654
655        match result.unwrap() {
656            KrakenSpotWsMessage::Book { data, is_snapshot } => {
657                assert!(!data.is_empty());
658                assert!(is_snapshot);
659            }
660            _ => panic!("Expected Book message"),
661        }
662    }
663
664    #[rstest]
665    fn test_send_order_request_variant_construction() {
666        let cmd = SpotHandlerCommand::SendOrderRequest {
667            req_id: 7,
668            payload: r#"{"method":"add_order","req_id":7}"#.to_string(),
669        };
670
671        match cmd {
672            SpotHandlerCommand::SendOrderRequest { req_id, payload } => {
673                assert_eq!(req_id, 7);
674                assert!(payload.contains("add_order"));
675            }
676            _ => panic!("Expected SendOrderRequest, was a different variant"),
677        }
678    }
679
680    #[rstest]
681    #[tokio::test]
682    async fn test_send_order_request_without_active_client_does_not_panic() {
683        let signal = Arc::new(AtomicBool::new(false));
684        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
685        let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel::<Message>();
686        let subscriptions = SubscriptionState::new(':');
687
688        let mut handler = SpotFeedHandler::new(signal.clone(), cmd_rx, raw_rx, subscriptions);
689
690        cmd_tx
691            .send(SpotHandlerCommand::SendOrderRequest {
692                req_id: 42,
693                payload: r#"{"method":"add_order","req_id":42}"#.to_string(),
694            })
695            .unwrap();
696
697        drop(cmd_tx);
698        drop(raw_tx);
699
700        let result = handler.next().await;
701        assert!(
702            result.is_none(),
703            "Handler should return None when streams close"
704        );
705    }
706
707    #[rstest]
708    fn test_quotes_and_book_subscriptions_independent() {
709        let handler = create_test_handler();
710        handler.subscriptions.mark_subscribe("quotes:BTC/USD");
711        handler.subscriptions.confirm_subscribe("quotes:BTC/USD");
712
713        let book_json = r#"{
714            "channel": "book",
715            "type": "snapshot",
716            "data": [{
717                "symbol": "BTC/USD",
718                "bids": [{"price": 105944.20, "qty": 2.5}],
719                "asks": [{"price": 105944.30, "qty": 3.2}],
720                "checksum": 12345,
721                "timestamp": "2023-10-06T17:35:55.440295Z"
722            }]
723        }"#;
724
725        let book_result = handler.parse_message(book_json);
726        assert!(
727            book_result.is_none(),
728            "Book message should be filtered without book: subscription"
729        );
730
731        let ticker_json = r#"{
732            "channel": "ticker",
733            "type": "snapshot",
734            "data": [{
735                "symbol": "BTC/USD",
736                "bid": 105944.20,
737                "bid_qty": 2.5,
738                "ask": 105944.30,
739                "ask_qty": 3.2,
740                "last": 105899.40,
741                "volume": 163.28908096,
742                "vwap": 105904.39279,
743                "low": 104711.00,
744                "high": 106613.10,
745                "change": 250.00,
746                "change_pct": 0.24,
747                "timestamp": "2022-12-25T09:30:59.123456Z"
748            }]
749        }"#;
750
751        let ticker_result = handler.parse_message(ticker_json);
752        assert!(
753            ticker_result.is_some(),
754            "Ticker should pass with quotes subscription"
755        );
756    }
757
758    #[rstest]
759    fn test_parse_message_routes_add_order_response_to_order_response_variant() {
760        use super::super::enums::KrakenWsMethod;
761
762        let handler = create_test_handler();
763        let json = r#"{"method":"add_order","req_id":42,"success":true,"time_in":"2026-05-05T10:00:00.123Z","time_out":"2026-05-05T10:00:00.125Z","result":{"order_id":"OABCDE-12345-FGHIJ","cl_ord_id":"O-20260505-000001","order_userref":0}}"#;
764
765        let result = handler.parse_message(json);
766        match result {
767            Some(KrakenSpotWsMessage::OrderResponse(resp)) => {
768                assert_eq!(resp.method, KrakenWsMethod::AddOrder);
769                assert_eq!(resp.req_id, Some(42));
770                assert!(resp.success);
771            }
772            other => panic!("expected OrderResponse, was {other:?}"),
773        }
774    }
775
776    #[rstest]
777    fn test_parse_level3_snapshot_with_subscription_passes() {
778        let handler = create_test_handler();
779        handler.subscriptions.mark_subscribe("level3:BTC/USD");
780        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
781
782        let json = r#"{
783            "channel": "level3",
784            "type": "snapshot",
785            "data": [{
786                "symbol": "BTC/USD",
787                "bids": [],
788                "asks": [],
789                "checksum": 0,
790                "timestamp": "2024-01-01T00:00:00Z"
791            }]
792        }"#;
793
794        let result = handler.parse_message(json);
795        assert!(matches!(result, Some(KrakenSpotWsMessage::L3Snapshot(_))));
796    }
797
798    #[rstest]
799    fn test_parse_level3_update_without_subscription_filtered() {
800        let handler = create_test_handler();
801        let json = r#"{
802            "channel": "level3",
803            "type": "update",
804            "data": [{
805                "symbol": "BTC/USD",
806                "bids": [],
807                "asks": [],
808                "checksum": 0,
809                "timestamp": "2024-01-01T00:00:00Z"
810            }]
811        }"#;
812
813        let result = handler.parse_message(json);
814        assert!(result.is_none());
815    }
816
817    #[rstest]
818    fn test_parse_level3_snapshot_compact_json() {
819        let handler = create_test_handler();
820        handler.subscriptions.mark_subscribe("level3:BTC/USD");
821        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
822        let json = r#"{"channel":"level3","type":"snapshot","data":[{"symbol":"BTC/USD","bids":[],"asks":[],"checksum":0,"timestamp":"2024-01-01T00:00:00Z"}]}"#;
823        assert!(matches!(
824            handler.parse_message(json),
825            Some(KrakenSpotWsMessage::L3Snapshot(_))
826        ));
827    }
828
829    #[rstest]
830    fn test_parse_level3_snapshot_preserves_raw_decimal() {
831        let handler = create_test_handler();
832        handler.subscriptions.mark_subscribe("level3:BTC/USD");
833        handler.subscriptions.confirm_subscribe("level3:BTC/USD");
834
835        let json = r#"{
836            "channel": "level3",
837            "type": "snapshot",
838            "data": [{
839                "symbol": "BTC/USD",
840                "bids": [{
841                    "order_id": "order-bid-1",
842                    "limit_price": 42000.50000,
843                    "order_qty": 0.01000000,
844                    "timestamp": "2024-01-01T00:00:00Z"
845                }],
846                "asks": [],
847                "checksum": 0,
848                "timestamp": "2024-01-01T00:00:00Z"
849            }]
850        }"#;
851
852        let Some(KrakenSpotWsMessage::L3Snapshot(snap)) = handler.parse_message(json) else {
853            panic!("expected L3 snapshot");
854        };
855        assert_eq!(snap.bids[0].limit_price.raw, "42000.50000");
856        assert_eq!(snap.bids[0].order_qty.raw, "0.01000000");
857    }
858}
nautilus_kraken/websocket/spot_v2/handler.rs

nautilus_kraken/websocket/spot_v2/
handler.rs
