nautilus_kraken/data/
futures.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//! Kraken Futures data client implementation.
17
18use std::{
19    future::Future,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, AtomicU64, Ordering},
23    },
24    time::Duration,
25};
26
27use ahash::AHashMap;
28use anyhow::Context;
29use async_trait::async_trait;
30use dashmap::DashMap;
31use nautilus_common::{
32    clients::DataClient,
33    live::{get_data_event_sender, sender::EventSender},
34    messages::{
35        DataEvent,
36        data::{
37            BarsResponse, BookResponse, DataResponse, FundingRatesResponse, InstrumentResponse,
38            InstrumentsResponse, RequestBars, RequestBookSnapshot, RequestFundingRates,
39            RequestInstrument, RequestInstruments, RequestTrades, SubscribeBars,
40            SubscribeBookDeltas, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
41            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices, SubscribeQuotes,
42            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
43            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrumentStatus,
44            UnsubscribeMarkPrices, UnsubscribeQuotes, UnsubscribeTrades,
45        },
46    },
47};
48use nautilus_core::{
49    AtomicMap, AtomicSet,
50    datetime::datetime_to_unix_nanos,
51    nanos::UnixNanos,
52    time::{AtomicTime, get_atomic_clock_realtime},
53};
54use nautilus_live::{SocketControl, task::TaskGroup};
55use nautilus_model::{
56    data::{Data, OrderBookDeltas, QuoteTick},
57    enums::BookType,
58    identifiers::{ClientId, InstrumentId, Venue},
59    instruments::{Instrument, InstrumentAny},
60    orderbook::OrderBook,
61};
62use rust_decimal_macros::dec;
63use tokio_util::sync::CancellationToken;
64
65use crate::{
66    common::{consts::KRAKEN_VENUE, lookup_instrument_in_snapshot},
67    config::KrakenDataClientConfig,
68    http::{
69        KrakenFuturesHttpClient, futures::client::KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND,
70    },
71    websocket::futures::{
72        client::KrakenFuturesWebSocketClient,
73        messages::KrakenFuturesWsMessage,
74        parse::{
75            parse_futures_ws_book_delta, parse_futures_ws_book_snapshot_deltas,
76            parse_futures_ws_funding_rate, parse_futures_ws_index_price,
77            parse_futures_ws_mark_price, parse_futures_ws_trade_tick,
78        },
79    },
80};
81
82/// Kraken Futures data client.
83///
84/// Provides real-time market data from Kraken Futures markets.
85#[allow(dead_code)]
86#[derive(Debug)]
87pub struct KrakenFuturesDataClient {
88    clock: &'static AtomicTime,
89    client_id: ClientId,
90    config: KrakenDataClientConfig,
91    http: KrakenFuturesHttpClient,
92    ws: KrakenFuturesWebSocketClient,
93    is_connected: AtomicBool,
94    cancellation_token: CancellationToken,
95    session_tasks: TaskGroup,
96    command_tasks: TaskGroup,
97    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
98    quote_instruments: Arc<AtomicSet<InstrumentId>>,
99    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
100    book_instruments: Arc<AtomicSet<InstrumentId>>,
101    data_sender: EventSender<DataEvent>,
102}
103
104impl KrakenFuturesDataClient {
105    /// Creates a new [`KrakenFuturesDataClient`] instance.
106    pub fn new(client_id: ClientId, config: KrakenDataClientConfig) -> anyhow::Result<Self> {
107        let session_tasks = TaskGroup::new();
108        let cancellation_token = session_tasks.cancellation_token();
109        let command_tasks = TaskGroup::new();
110        let proxy_url = config
111            .proxy_url
112            .as_ref()
113            .map(|value| value.expose_secret().to_owned());
114
115        let http = KrakenFuturesHttpClient::new(
116            config.environment,
117            config.base_url.clone(),
118            config.timeout_secs,
119            None,
120            None,
121            None,
122            proxy_url.clone(),
123            config
124                .max_requests_per_second
125                .unwrap_or(KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND),
126        )?;
127
128        let ws = KrakenFuturesWebSocketClient::with_credentials(
129            config.ws_public_url(),
130            config.heartbeat_interval_secs,
131            None,
132            None,
133            config.transport_backend,
134            proxy_url,
135        )
136        .with_socket_control(SocketControl::new(
137            client_id,
138            Some(*KRAKEN_VENUE),
139            "kraken-futures-data-streams",
140        ));
141
142        Ok(Self {
143            clock: get_atomic_clock_realtime(),
144            client_id,
145            config,
146            http,
147            ws,
148            is_connected: AtomicBool::new(false),
149            cancellation_token,
150            session_tasks,
151            command_tasks,
152            instruments: Arc::new(AtomicMap::new()),
153            quote_instruments: Arc::new(AtomicSet::new()),
154            last_quotes: Arc::new(DashMap::new()),
155            book_instruments: Arc::new(AtomicSet::new()),
156            data_sender: get_data_event_sender(),
157        })
158    }
159
160    /// Returns the cached instruments.
161    #[must_use]
162    pub fn instruments(&self) -> Vec<InstrumentAny> {
163        self.instruments.load().values().cloned().collect()
164    }
165
166    /// Returns a cached instrument by ID.
167    #[must_use]
168    pub fn get_instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
169        self.instruments.load().get(instrument_id).cloned()
170    }
171
172    async fn load_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
173        let instruments = self
174            .http
175            .request_instruments()
176            .await
177            .context("Failed to load futures instruments")?;
178
179        self.instruments.rcu(|m| {
180            for instrument in &instruments {
181                m.insert(instrument.id(), instrument.clone());
182            }
183        });
184
185        self.http.cache_instruments(&instruments);
186
187        log::debug!(
188            "Loaded instruments: client_id={}, count={}",
189            self.client_id,
190            instruments.len()
191        );
192
193        Ok(instruments)
194    }
195
196    fn spawn_ws<F>(&self, fut: F, context: &'static str)
197    where
198        F: Future<Output = anyhow::Result<()>> + Send + 'static,
199    {
200        let future = async move {
201            if let Err(e) = fut.await {
202                log::error!("{context}: {e:?}");
203            }
204        };
205
206        if let Err(e) = self.command_tasks.spawn(future) {
207            log::warn!("Skipping Kraken Futures {context} after shutdown began: {e}");
208        }
209    }
210
211    fn spawn_command<F>(&self, future: F)
212    where
213        F: Future<Output = ()> + Send + 'static,
214    {
215        if let Err(e) = self.command_tasks.spawn(future) {
216            log::warn!("Skipping Kraken Futures data command after shutdown began: {e}");
217        }
218    }
219
220    async fn finish_tasks(&self) -> anyhow::Result<()> {
221        let (session_result, command_result) = tokio::join!(
222            self.session_tasks
223                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
224            self.command_tasks
225                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
226        );
227        session_result.context("failed to finish Kraken Futures data session tasks")?;
228        command_result.context("failed to finish Kraken Futures data command tasks")?;
229        Ok(())
230    }
231
232    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
233        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
234            self.session_tasks.begin_shutdown();
235            self.command_tasks.begin_shutdown();
236            let _ = self.ws.close().await;
237            self.finish_tasks().await?;
238            self.session_tasks
239                .start_generation()
240                .context("failed to start Kraken Futures data session task generation")?;
241            self.command_tasks
242                .start_generation()
243                .context("failed to start Kraken Futures data command task generation")?;
244            self.cancellation_token = self.session_tasks.cancellation_token();
245            self.ws = KrakenFuturesWebSocketClient::with_credentials(
246                self.config.ws_public_url(),
247                self.config.heartbeat_interval_secs,
248                None,
249                None,
250                self.config.transport_backend,
251                self.config
252                    .proxy_url
253                    .as_ref()
254                    .map(|value| value.expose_secret().to_owned()),
255            )
256            .with_socket_control(SocketControl::new(
257                self.client_id,
258                Some(*KRAKEN_VENUE),
259                "kraken-futures-data-streams",
260            ));
261        }
262        Ok(())
263    }
264
265    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
266        self.session_tasks.begin_shutdown();
267        self.command_tasks.begin_shutdown();
268        let ws_result = self.ws.close().await;
269        let tasks_result = self.finish_tasks().await;
270        self.is_connected.store(false, Ordering::Release);
271        tasks_result?;
272        Ok(ws_result?)
273    }
274
275    fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
276        let mut rx = self
277            .ws
278            .take_output_rx()
279            .context("Failed to take futures WebSocket output receiver")?;
280        let data_sender = self.data_sender.clone();
281        let instruments = self.instruments.clone();
282        let quote_instruments = self.quote_instruments.clone();
283        let last_quotes = self.last_quotes.clone();
284        let book_instruments = self.book_instruments.clone();
285        let book_sequence = Arc::new(AtomicU64::new(0));
286        let cancellation_token = self.cancellation_token.clone();
287        let clock = self.clock;
288
289        let future = async move {
290            let mut order_books: AHashMap<InstrumentId, OrderBook> = AHashMap::new();
291
292            loop {
293                tokio::select! {
294                    () = cancellation_token.cancelled() => {
295                        log::debug!("Futures message handler cancelled");
296                        break;
297                    }
298                    msg = rx.recv() => {
299                        match msg {
300                            Some(ws_msg) => {
301                                Self::handle_ws_message(
302                                    ws_msg,
303                                    &data_sender,
304                                    &instruments,
305                                    &quote_instruments,
306                                    &book_instruments,
307                                    &mut order_books,
308                                    &last_quotes,
309                                    &book_sequence,
310                                    clock,
311                                );
312                            }
313                            None => {
314                                log::debug!("Futures WebSocket stream ended");
315                                break;
316                            }
317                        }
318                    }
319                }
320            }
321        };
322
323        self.session_tasks
324            .spawn(future)
325            .context("failed to register Kraken Futures message handler")
326    }
327
328    #[expect(clippy::too_many_arguments)]
329    fn handle_ws_message(
330        msg: KrakenFuturesWsMessage,
331        sender: &EventSender<DataEvent>,
332        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
333        quote_instruments: &Arc<AtomicSet<InstrumentId>>,
334        book_instruments: &Arc<AtomicSet<InstrumentId>>,
335        order_books: &mut AHashMap<InstrumentId, OrderBook>,
336        last_quotes: &DashMap<InstrumentId, QuoteTick>,
337        book_sequence: &Arc<AtomicU64>,
338        clock: &'static AtomicTime,
339    ) {
340        let ts_init = clock.get_time_ns();
341
342        match msg {
343            KrakenFuturesWsMessage::Ticker(ticker) => {
344                let instruments = instruments.load();
345                let Some(instrument) =
346                    lookup_instrument_in_snapshot(&instruments, ticker.product_id.as_str())
347                else {
348                    log::warn!("No instrument for product_id: {}", ticker.product_id);
349                    return;
350                };
351
352                if let Some(mark) = parse_futures_ws_mark_price(&ticker, instrument, ts_init)
353                    && let Err(e) = sender.send(DataEvent::Data(Data::MarkPrice(mark)))
354                {
355                    log::error!("Failed to send mark price: {e}");
356                }
357
358                if let Some(index) = parse_futures_ws_index_price(&ticker, instrument, ts_init)
359                    && let Err(e) = sender.send(DataEvent::Data(Data::IndexPrice(index)))
360                {
361                    log::error!("Failed to send index price: {e}");
362                }
363
364                if let Some(funding) = parse_futures_ws_funding_rate(&ticker, instrument, ts_init)
365                    && let Err(e) = sender.send(DataEvent::FundingRate(funding))
366                {
367                    log::error!("Failed to send funding rate: {e}");
368                }
369            }
370            KrakenFuturesWsMessage::Trade(trade) => {
371                let instruments = instruments.load();
372                let Some(instrument) =
373                    lookup_instrument_in_snapshot(&instruments, trade.product_id.as_str())
374                else {
375                    log::warn!("No instrument for product_id: {}", trade.product_id);
376                    return;
377                };
378
379                match parse_futures_ws_trade_tick(&trade, instrument, ts_init) {
380                    Ok(tick) => {
381                        if let Err(e) = sender.send(DataEvent::Data(Data::Trade(tick))) {
382                            log::error!("Failed to send trade: {e}");
383                        }
384                    }
385                    Err(e) => log::error!("Failed to parse futures trade tick: {e}"),
386                }
387            }
388            KrakenFuturesWsMessage::BookSnapshot(snapshot) => {
389                let instruments = instruments.load();
390                let Some(instrument) =
391                    lookup_instrument_in_snapshot(&instruments, snapshot.product_id.as_str())
392                else {
393                    log::warn!("No instrument for product_id: {}", snapshot.product_id);
394                    return;
395                };
396                let instrument_id = instrument.id();
397                let sequence = book_sequence.load(Ordering::Relaxed);
398
399                match parse_futures_ws_book_snapshot_deltas(
400                    &snapshot, instrument, sequence, ts_init,
401                ) {
402                    Ok(delta_vec) => {
403                        if delta_vec.is_empty() {
404                            return;
405                        }
406                        book_sequence.fetch_add(delta_vec.len() as u64, Ordering::Relaxed);
407                        let deltas = OrderBookDeltas::new(instrument_id, delta_vec);
408
409                        let has_quote_sub = quote_instruments.contains(&instrument_id);
410                        let book = order_books
411                            .entry(instrument_id)
412                            .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
413
414                        if let Err(e) = book.apply_deltas(&deltas) {
415                            log::error!("Failed to apply snapshot deltas to order book: {e}");
416                        } else if has_quote_sub {
417                            Self::maybe_emit_quote(
418                                book,
419                                instrument_id,
420                                last_quotes,
421                                ts_init,
422                                sender,
423                            );
424                        }
425
426                        let has_book_sub = book_instruments.contains(&instrument_id);
427
428                        if has_book_sub
429                            && let Err(e) =
430                                sender.send(DataEvent::Data(Data::BookDeltas(Box::new(deltas))))
431                        {
432                            log::error!("Failed to send book snapshot deltas: {e}");
433                        }
434                    }
435                    Err(e) => log::error!("Failed to parse book snapshot: {e}"),
436                }
437            }
438            KrakenFuturesWsMessage::BookDelta(delta) => {
439                let instruments = instruments.load();
440                let Some(instrument) =
441                    lookup_instrument_in_snapshot(&instruments, delta.product_id.as_str())
442                else {
443                    log::warn!("No instrument for product_id: {}", delta.product_id);
444                    return;
445                };
446                let instrument_id = instrument.id();
447                let sequence = book_sequence.fetch_add(1, Ordering::Relaxed);
448                match parse_futures_ws_book_delta(&delta, instrument, sequence, ts_init) {
449                    Ok(book_delta) => {
450                        let deltas = OrderBookDeltas::new(instrument_id, vec![book_delta]);
451
452                        let has_quote_sub = quote_instruments.contains(&instrument_id);
453
454                        if let Some(book) = order_books.get_mut(&instrument_id) {
455                            if let Err(e) = book.apply_deltas(&deltas) {
456                                log::error!("Failed to apply delta to order book: {e}");
457                            } else if has_quote_sub {
458                                Self::maybe_emit_quote(
459                                    book,
460                                    instrument_id,
461                                    last_quotes,
462                                    ts_init,
463                                    sender,
464                                );
465                            }
466                        }
467
468                        let has_book_sub = book_instruments.contains(&instrument_id);
469
470                        if has_book_sub
471                            && let Err(e) =
472                                sender.send(DataEvent::Data(Data::BookDeltas(Box::new(deltas))))
473                        {
474                            log::error!("Failed to send book delta: {e}");
475                        }
476                    }
477                    Err(e) => log::error!("Failed to parse book delta: {e}"),
478                }
479            }
480            KrakenFuturesWsMessage::Reconnected => {
481                last_quotes.clear();
482                log::info!("Futures WebSocket reconnected");
483            }
484            KrakenFuturesWsMessage::OpenOrdersCancel(_)
485            | KrakenFuturesWsMessage::OpenOrdersDelta(_)
486            | KrakenFuturesWsMessage::FillsDelta(_)
487            | KrakenFuturesWsMessage::Challenge(_) => {}
488        }
489    }
490
491    fn maybe_emit_quote(
492        book: &OrderBook,
493        instrument_id: InstrumentId,
494        last_quotes: &DashMap<InstrumentId, QuoteTick>,
495        ts_init: UnixNanos,
496        sender: &EventSender<DataEvent>,
497    ) {
498        let (Some(bid_price), Some(ask_price)) = (book.best_bid_price(), book.best_ask_price())
499        else {
500            return;
501        };
502        let (Some(bid_size), Some(ask_size)) = (book.best_bid_size(), book.best_ask_size()) else {
503            return;
504        };
505
506        let bid = bid_price.as_decimal();
507        let ask = ask_price.as_decimal();
508        if bid > dec!(0) && (ask - bid) / bid > dec!(0.25) {
509            log::debug!("Filtered quote with wide spread: bid={bid}, ask={ask}");
510            return;
511        }
512
513        let quote = QuoteTick::new(
514            instrument_id,
515            bid_price,
516            ask_price,
517            bid_size,
518            ask_size,
519            ts_init,
520            ts_init,
521        );
522
523        if matches!(
524            last_quotes.get(&instrument_id),
525            Some(prev) if prev.has_same_bid_ask(&quote)
526        ) {
527            return;
528        }
529
530        last_quotes.insert(instrument_id, quote);
531
532        if let Err(e) = sender.send(DataEvent::Data(Data::Quote(quote))) {
533            log::error!("Failed to send quote: {e}");
534        }
535    }
536}
537
538#[async_trait(?Send)]
539impl DataClient for KrakenFuturesDataClient {
540    fn client_id(&self) -> ClientId {
541        self.client_id
542    }
543
544    fn venue(&self) -> Option<Venue> {
545        Some(*KRAKEN_VENUE)
546    }
547
548    fn start(&mut self) -> anyhow::Result<()> {
549        log::info!(
550            "Starting Futures data client: client_id={}, environment={:?}",
551            self.client_id,
552            self.config.environment
553        );
554        Ok(())
555    }
556
557    fn stop(&mut self) -> anyhow::Result<()> {
558        log::info!("Stopping Futures data client: {}", self.client_id);
559        self.session_tasks.begin_shutdown();
560        self.command_tasks.begin_shutdown();
561        self.ws.begin_shutdown();
562        self.is_connected.store(false, Ordering::Relaxed);
563        Ok(())
564    }
565
566    fn reset(&mut self) -> anyhow::Result<()> {
567        log::info!("Resetting Futures data client: {}", self.client_id);
568        self.session_tasks.begin_shutdown();
569        self.command_tasks.begin_shutdown();
570        self.ws.begin_shutdown();
571        self.is_connected.store(false, Ordering::Relaxed);
572
573        self.instruments.store(ahash::AHashMap::new());
574
575        self.last_quotes.clear();
576        self.quote_instruments.store(ahash::AHashSet::new());
577
578        Ok(())
579    }
580
581    fn dispose(&mut self) -> anyhow::Result<()> {
582        log::debug!("Disposing Futures data client: {}", self.client_id);
583        self.stop()
584    }
585
586    fn is_connected(&self) -> bool {
587        self.is_connected.load(Ordering::SeqCst)
588    }
589
590    fn is_disconnected(&self) -> bool {
591        !self.is_connected()
592    }
593
594    async fn connect(&mut self) -> anyhow::Result<()> {
595        if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
596            return Ok(());
597        }
598
599        self.prepare_task_groups().await?;
600        self.last_quotes.clear();
601
602        let instruments = self.load_instruments().await?;
603
604        let session_result = async {
605            self.ws
606                .connect()
607                .await
608                .context("Failed to connect futures WebSocket")?;
609            self.ws
610                .wait_until_active(10.0)
611                .await
612                .context("Futures WebSocket failed to become active")?;
613
614            self.spawn_message_handler()?;
615
616            Ok::<(), anyhow::Error>(())
617        }
618        .await;
619
620        if let Err(e) = session_result {
621            if let Err(teardown_error) = self.teardown_partial_connect().await {
622                return Err(e.context(format!(
623                    "Kraken Futures data startup teardown failed: {teardown_error}"
624                )));
625            }
626            return Err(e);
627        }
628
629        for instrument in instruments {
630            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
631                log::error!("Failed to send instrument: {e}");
632            }
633        }
634
635        self.is_connected.store(true, Ordering::Release);
636        log::info!(
637            "Connected: client_id={}, product_type=Futures",
638            self.client_id
639        );
640        Ok(())
641    }
642
643    async fn disconnect(&mut self) -> anyhow::Result<()> {
644        self.teardown_partial_connect().await?;
645
646        self.last_quotes.clear();
647        self.quote_instruments.store(ahash::AHashSet::new());
648        self.is_connected.store(false, Ordering::Relaxed);
649
650        log::info!("Disconnected: client_id={}", self.client_id);
651        Ok(())
652    }
653
654    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
655        log::debug!("subscribe_instruments: Kraken instruments are fetched via HTTP on connect");
656        Ok(())
657    }
658
659    fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
660        log::debug!("subscribe_instrument: Kraken instruments are fetched via HTTP on connect");
661        Ok(())
662    }
663
664    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
665        let instrument_id = cmd.instrument_id;
666        let depth = cmd.depth;
667
668        if cmd.book_type != BookType::L2_MBP {
669            log::warn!(
670                "Book type {:?} not supported by Kraken, skipping subscription",
671                cmd.book_type
672            );
673            return Ok(());
674        }
675
676        self.book_instruments.insert(instrument_id);
677
678        let ws = self.ws.clone();
679        self.spawn_ws(
680            async move {
681                ws.subscribe_book(instrument_id, depth.map(|d| d.get() as u32))
682                    .await
683                    .map_err(|e| anyhow::anyhow!("{e}"))
684            },
685            "subscribe book",
686        );
687
688        Ok(())
689    }
690
691    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
692        let instrument_id = cmd.instrument_id;
693        let ws = self.ws.clone();
694
695        self.last_quotes.remove(&instrument_id);
696        self.quote_instruments.insert(instrument_id);
697
698        self.spawn_ws(
699            async move {
700                ws.subscribe_quotes(instrument_id)
701                    .await
702                    .map_err(|e| anyhow::anyhow!("{e}"))
703            },
704            "subscribe quotes",
705        );
706
707        Ok(())
708    }
709
710    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
711        let instrument_id = cmd.instrument_id;
712        let ws = self.ws.clone();
713
714        self.spawn_ws(
715            async move {
716                ws.subscribe_trades(instrument_id)
717                    .await
718                    .map_err(|e| anyhow::anyhow!("{e}"))
719            },
720            "subscribe trades",
721        );
722
723        Ok(())
724    }
725
726    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
727        let instrument_id = cmd.instrument_id;
728        let ws = self.ws.clone();
729
730        self.spawn_ws(
731            async move {
732                ws.subscribe_mark_price(instrument_id)
733                    .await
734                    .map_err(|e| anyhow::anyhow!("{e}"))
735            },
736            "subscribe mark price",
737        );
738
739        Ok(())
740    }
741
742    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
743        let instrument_id = cmd.instrument_id;
744        let ws = self.ws.clone();
745
746        self.spawn_ws(
747            async move {
748                ws.subscribe_index_price(instrument_id)
749                    .await
750                    .map_err(|e| anyhow::anyhow!("{e}"))
751            },
752            "subscribe index price",
753        );
754
755        Ok(())
756    }
757
758    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
759        let instrument_id = cmd.instrument_id;
760        let ws = self.ws.clone();
761
762        self.spawn_ws(
763            async move {
764                ws.subscribe_funding_rate(instrument_id)
765                    .await
766                    .map_err(|e| anyhow::anyhow!("{e}"))
767            },
768            "subscribe funding rate",
769        );
770
771        Ok(())
772    }
773
774    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
775        log::warn!(
776            "Cannot subscribe to {} bars: Kraken Futures does not support EXTERNAL bar streaming",
777            cmd.bar_type
778        );
779        Ok(())
780    }
781
782    fn subscribe_instrument_status(
783        &mut self,
784        cmd: SubscribeInstrumentStatus,
785    ) -> anyhow::Result<()> {
786        log::debug!(
787            "subscribe_instrument_status: {} (status changes detected via periodic instrument polling)",
788            cmd.instrument_id,
789        );
790        Ok(())
791    }
792
793    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
794        let instrument_id = cmd.instrument_id;
795
796        self.book_instruments.remove(&instrument_id);
797
798        let ws = self.ws.clone();
799        self.spawn_ws(
800            async move {
801                ws.unsubscribe_book(instrument_id)
802                    .await
803                    .map_err(|e| anyhow::anyhow!("{e}"))
804            },
805            "unsubscribe book",
806        );
807
808        Ok(())
809    }
810
811    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
812        let instrument_id = cmd.instrument_id;
813        let ws = self.ws.clone();
814
815        self.quote_instruments.remove(&instrument_id);
816        self.last_quotes.remove(&instrument_id);
817        self.spawn_ws(
818            async move {
819                ws.unsubscribe_quotes(instrument_id)
820                    .await
821                    .map_err(|e| anyhow::anyhow!("{e}"))
822            },
823            "unsubscribe quotes",
824        );
825
826        Ok(())
827    }
828
829    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
830        let instrument_id = cmd.instrument_id;
831        let ws = self.ws.clone();
832
833        self.spawn_ws(
834            async move {
835                ws.unsubscribe_trades(instrument_id)
836                    .await
837                    .map_err(|e| anyhow::anyhow!("{e}"))
838            },
839            "unsubscribe trades",
840        );
841
842        Ok(())
843    }
844
845    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
846        let instrument_id = cmd.instrument_id;
847        let ws = self.ws.clone();
848
849        self.spawn_ws(
850            async move {
851                ws.unsubscribe_mark_price(instrument_id)
852                    .await
853                    .map_err(|e| anyhow::anyhow!("{e}"))
854            },
855            "unsubscribe mark price",
856        );
857
858        Ok(())
859    }
860
861    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
862        let instrument_id = cmd.instrument_id;
863        let ws = self.ws.clone();
864
865        self.spawn_ws(
866            async move {
867                ws.unsubscribe_index_price(instrument_id)
868                    .await
869                    .map_err(|e| anyhow::anyhow!("{e}"))
870            },
871            "unsubscribe index price",
872        );
873
874        Ok(())
875    }
876
877    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
878        let instrument_id = cmd.instrument_id;
879        let ws = self.ws.clone();
880
881        self.spawn_ws(
882            async move {
883                ws.unsubscribe_funding_rate(instrument_id)
884                    .await
885                    .map_err(|e| anyhow::anyhow!("{e}"))
886            },
887            "unsubscribe funding rate",
888        );
889
890        Ok(())
891    }
892
893    fn unsubscribe_bars(&mut self, _cmd: &UnsubscribeBars) -> anyhow::Result<()> {
894        Ok(())
895    }
896
897    fn unsubscribe_instrument_status(
898        &mut self,
899        _cmd: &UnsubscribeInstrumentStatus,
900    ) -> anyhow::Result<()> {
901        Ok(())
902    }
903
904    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
905        let http = self.http.clone();
906        let sender = self.data_sender.clone();
907        let instruments_cache = self.instruments.clone();
908        let request_id = request.request_id;
909        let client_id = request.client_id.unwrap_or(self.client_id);
910        let venue = *KRAKEN_VENUE;
911        let start_nanos = datetime_to_unix_nanos(request.start);
912        let end_nanos = datetime_to_unix_nanos(request.end);
913        let params = request.params;
914        let clock = self.clock;
915
916        self.spawn_command(async move {
917            match http.request_instruments().await {
918                Ok(instruments) => {
919                    instruments_cache.rcu(|m| {
920                        for instrument in &instruments {
921                            m.insert(instrument.id(), instrument.clone());
922                        }
923                    });
924                    http.cache_instruments(&instruments);
925
926                    let response = DataResponse::Instruments(InstrumentsResponse::new(
927                        request_id,
928                        client_id,
929                        venue,
930                        instruments,
931                        start_nanos,
932                        end_nanos,
933                        clock.get_time_ns(),
934                        params,
935                    ));
936
937                    if let Err(e) = sender.send(DataEvent::Response(response)) {
938                        log::error!("Failed to send instruments response: {e}");
939                    }
940                }
941                Err(e) => log::error!("Instruments request failed: {e:?}"),
942            }
943        });
944
945        Ok(())
946    }
947
948    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
949        let http = self.http.clone();
950        let sender = self.data_sender.clone();
951        let instruments = self.instruments.clone();
952        let instrument_id = request.instrument_id;
953        let request_id = request.request_id;
954        let client_id = request.client_id.unwrap_or(self.client_id);
955        let start_nanos = datetime_to_unix_nanos(request.start);
956        let end_nanos = datetime_to_unix_nanos(request.end);
957        let params = request.params;
958        let clock = self.clock;
959
960        self.spawn_command(async move {
961            match http.request_instruments().await {
962                Ok(all_instruments) => {
963                    instruments.rcu(|m| {
964                        for instrument in &all_instruments {
965                            m.insert(instrument.id(), instrument.clone());
966                        }
967                    });
968                    http.cache_instruments(&all_instruments);
969
970                    let instrument = all_instruments
971                        .into_iter()
972                        .find(|i| i.id() == instrument_id);
973
974                    if let Some(instrument) = instrument {
975                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
976                            request_id,
977                            client_id,
978                            instrument.id(),
979                            instrument,
980                            start_nanos,
981                            end_nanos,
982                            clock.get_time_ns(),
983                            params,
984                        )));
985
986                        if let Err(e) = sender.send(DataEvent::Response(response)) {
987                            log::error!("Failed to send instrument response: {e}");
988                        }
989                    } else {
990                        log::error!("Instrument not found: {instrument_id}");
991                    }
992                }
993                Err(e) => log::error!("Instrument request failed: {e:?}"),
994            }
995        });
996
997        Ok(())
998    }
999
1000    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1001        let http = self.http.clone();
1002        let sender = self.data_sender.clone();
1003        let instrument_id = request.instrument_id;
1004        let start = request.start;
1005        let end = request.end;
1006        let limit = request.limit.map(|n| n.get() as u64);
1007        let request_id = request.request_id;
1008        let client_id = request.client_id.unwrap_or(self.client_id);
1009        let params = request.params;
1010        let clock = self.clock;
1011        let start_nanos = datetime_to_unix_nanos(start);
1012        let end_nanos = datetime_to_unix_nanos(end);
1013
1014        self.spawn_command(async move {
1015            match http.request_trades(instrument_id, start, end, limit).await {
1016                Ok(trades) => {
1017                    let response = DataResponse::Trades(TradesResponse::new(
1018                        request_id,
1019                        client_id,
1020                        instrument_id,
1021                        trades,
1022                        start_nanos,
1023                        end_nanos,
1024                        clock.get_time_ns(),
1025                        params,
1026                    ));
1027
1028                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1029                        log::error!("Failed to send trades response: {e}");
1030                    }
1031                }
1032                Err(e) => log::error!("Trades request failed: {e:?}"),
1033            }
1034        });
1035
1036        Ok(())
1037    }
1038
1039    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1040        let http = self.http.clone();
1041        let sender = self.data_sender.clone();
1042        let bar_type = request.bar_type;
1043        let start = request.start;
1044        let end = request.end;
1045        let limit = request.limit.map(|n| n.get() as u64);
1046        let request_id = request.request_id;
1047        let client_id = request.client_id.unwrap_or(self.client_id);
1048        let params = request.params;
1049        let clock = self.clock;
1050        let start_nanos = datetime_to_unix_nanos(start);
1051        let end_nanos = datetime_to_unix_nanos(end);
1052
1053        self.spawn_command(async move {
1054            match http.request_bars(bar_type, start, end, limit).await {
1055                Ok(bars) => {
1056                    let response = DataResponse::Bars(BarsResponse::new(
1057                        request_id,
1058                        client_id,
1059                        bar_type,
1060                        bars,
1061                        start_nanos,
1062                        end_nanos,
1063                        clock.get_time_ns(),
1064                        params,
1065                    ));
1066
1067                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1068                        log::error!("Failed to send bars response: {e}");
1069                    }
1070                }
1071                Err(e) => log::error!("Bars request failed: {e:?}"),
1072            }
1073        });
1074
1075        Ok(())
1076    }
1077
1078    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
1079        let http = self.http.clone();
1080        let sender = self.data_sender.clone();
1081        let instrument_id = request.instrument_id;
1082        let depth = request.depth.map(|n| n.get() as u32);
1083        let request_id = request.request_id;
1084        let client_id = request.client_id.unwrap_or(self.client_id);
1085        let params = request.params;
1086        let clock = self.clock;
1087
1088        self.spawn_command(async move {
1089            match http.request_book_snapshot(instrument_id, depth).await {
1090                Ok(book) => {
1091                    let response = DataResponse::Book(BookResponse::new(
1092                        request_id,
1093                        client_id,
1094                        instrument_id,
1095                        book,
1096                        None,
1097                        None,
1098                        clock.get_time_ns(),
1099                        params,
1100                    ));
1101
1102                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1103                        log::error!("Failed to send book snapshot response: {e}");
1104                    }
1105                }
1106                Err(e) => log::error!("Book snapshot request failed: {e:?}"),
1107            }
1108        });
1109
1110        Ok(())
1111    }
1112
1113    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1114        let http = self.http.clone();
1115        let sender = self.data_sender.clone();
1116        let instrument_id = request.instrument_id;
1117        let start = request.start;
1118        let end = request.end;
1119        let limit = request.limit.map(|n| n.get());
1120        let request_id = request.request_id;
1121        let client_id = request.client_id.unwrap_or(self.client_id);
1122        let start_nanos = datetime_to_unix_nanos(start);
1123        let end_nanos = datetime_to_unix_nanos(end);
1124        let params = request.params;
1125        let clock = self.clock;
1126
1127        self.spawn_command(async move {
1128            match http
1129                .request_funding_rates(instrument_id, start, end, limit)
1130                .await
1131            {
1132                Ok(rates) => {
1133                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
1134                        request_id,
1135                        client_id,
1136                        instrument_id,
1137                        rates,
1138                        start_nanos,
1139                        end_nanos,
1140                        clock.get_time_ns(),
1141                        params,
1142                    ));
1143
1144                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1145                        log::error!("Failed to send funding rates response: {e}");
1146                    }
1147                }
1148                Err(e) => log::error!("Funding rates request failed: {e:?}"),
1149            }
1150        });
1151
1152        Ok(())
1153    }
1154}
1155
1156#[cfg(test)]
1157#[path = "../../benches"]
1158pub(crate) mod tests {
1159    // These routines are exercised by Criterion, not the unit-test harness
1160    #[allow(dead_code)]
1161    #[path = "quotes/mod.rs"]
1162    pub(crate) mod benchmarks;
1163
1164    use nautilus_common::{live::runner::set_data_event_sender, messages::DataEvent};
1165    use nautilus_core::UUID4;
1166    use nautilus_model::types::{Price, Quantity};
1167    use rstest::rstest;
1168    use rust_decimal::Decimal;
1169    use ustr::Ustr;
1170
1171    use super::*;
1172    use crate::{
1173        common::{
1174            consts::KRAKEN_CLIENT_ID,
1175            enums::{KrakenOrderSide, KrakenProductType},
1176            parse::parse_futures_instrument,
1177        },
1178        config::KrakenDataClientConfig,
1179        http::models::FuturesInstrumentsResponse,
1180        websocket::futures::messages::{
1181            KrakenFuturesBookDelta, KrakenFuturesBookSnapshot, KrakenFuturesFeed,
1182        },
1183    };
1184
1185    fn setup_test_env() {
1186        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1187        set_data_event_sender(sender);
1188    }
1189
1190    #[rstest]
1191    fn test_futures_data_client_new() {
1192        setup_test_env();
1193        let config = KrakenDataClientConfig {
1194            product_type: KrakenProductType::Futures,
1195            ..Default::default()
1196        };
1197        let client = KrakenFuturesDataClient::new(*KRAKEN_CLIENT_ID, config);
1198        assert!(client.is_ok());
1199
1200        let client = client.unwrap();
1201        assert_eq!(client.client_id(), *KRAKEN_CLIENT_ID);
1202        assert_eq!(client.venue(), Some(*KRAKEN_VENUE));
1203        assert!(!client.is_connected());
1204        assert!(client.is_disconnected());
1205        assert!(client.instruments().is_empty());
1206    }
1207
1208    #[rstest]
1209    fn test_futures_data_client_start_stop() {
1210        setup_test_env();
1211        let config = KrakenDataClientConfig {
1212            product_type: KrakenProductType::Futures,
1213            ..Default::default()
1214        };
1215        let mut client = KrakenFuturesDataClient::new(*KRAKEN_CLIENT_ID, config).unwrap();
1216
1217        assert!(client.start().is_ok());
1218        assert!(client.stop().is_ok());
1219        assert!(client.is_disconnected());
1220    }
1221
1222    #[rstest]
1223    #[case::bid_price(KrakenOrderSide::Buy, dec!(34893.0), dec!(6385))]
1224    #[case::ask_price(KrakenOrderSide::Sell, dec!(34911.0), dec!(20598))]
1225    #[case::bid_size(KrakenOrderSide::Buy, dec!(34892.5), dec!(7000))]
1226    #[case::ask_size(KrakenOrderSide::Sell, dec!(34911.5), dec!(21000))]
1227    fn test_handle_ws_message_emits_quote_only_on_top_of_book_change(
1228        #[case] side: KrakenOrderSide,
1229        #[case] price: Decimal,
1230        #[case] qty: Decimal,
1231    ) {
1232        let instrument = futures_instrument();
1233        let instrument_id = instrument.id();
1234        let instruments = Arc::new(AtomicMap::new());
1235        instruments.insert(instrument_id, instrument.clone());
1236        let quote_instruments = Arc::new(AtomicSet::new());
1237        quote_instruments.insert(instrument_id);
1238        let book_instruments = Arc::new(AtomicSet::new());
1239        book_instruments.insert(instrument_id);
1240        let mut order_books = AHashMap::new();
1241        let last_quotes = DashMap::new();
1242        let book_sequence = Arc::new(AtomicU64::new(0));
1243        let ts1 = UnixNanos::new(1_700_000_000_000_000_000);
1244        let ts2 = UnixNanos::new(1_700_000_000_100_000_000);
1245        let ts3 = UnixNanos::new(1_700_000_000_200_000_000);
1246        let ts4 = UnixNanos::new(1_700_000_000_300_000_000);
1247        let clock: &'static AtomicTime = Box::leak(Box::new(AtomicTime::new(false, ts1)));
1248        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1249        let sender: EventSender<DataEvent> = tx.into();
1250        let json = include_str!("../../test_data/ws_futures_book_snapshot.json");
1251        let snapshot: KrakenFuturesBookSnapshot = serde_json::from_str(json).unwrap();
1252
1253        let delta = |side, price, qty, seq, timestamp| {
1254            KrakenFuturesWsMessage::BookDelta(KrakenFuturesBookDelta {
1255                feed: KrakenFuturesFeed::Book,
1256                product_id: Ustr::from("PI_XBTUSD"),
1257                side,
1258                seq,
1259                price,
1260                qty,
1261                timestamp,
1262            })
1263        };
1264
1265        // The first delta changes depth only; the second changes one top-of-book field
1266        let messages = [
1267            (ts1, KrakenFuturesWsMessage::BookSnapshot(snapshot)),
1268            (
1269                ts2,
1270                delta(
1271                    KrakenOrderSide::Buy,
1272                    dec!(34892),
1273                    dec!(12000),
1274                    326_072_250,
1275                    1_612_269_825_900,
1276                ),
1277            ),
1278            (ts3, delta(side, price, qty, 326_072_251, 1_612_269_826_000)),
1279            (ts4, delta(side, price, qty, 326_072_252, 1_612_269_826_100)),
1280        ];
1281        let mut expected_deltas = Vec::new();
1282        let mut sequence = 0;
1283
1284        for (ts_init, msg) in &messages {
1285            let deltas = match msg {
1286                KrakenFuturesWsMessage::BookSnapshot(snapshot) => {
1287                    parse_futures_ws_book_snapshot_deltas(snapshot, &instrument, sequence, *ts_init)
1288                        .unwrap()
1289                }
1290
1291                KrakenFuturesWsMessage::BookDelta(delta) => {
1292                    let delta = parse_futures_ws_book_delta(delta, &instrument, sequence, *ts_init)
1293                        .unwrap();
1294                    vec![delta]
1295                }
1296
1297                other => panic!("Unexpected message: {other:?}"),
1298            };
1299
1300            sequence += deltas.len() as u64;
1301            expected_deltas.push(OrderBookDeltas::new(instrument_id, deltas));
1302        }
1303
1304        for (ts_init, msg) in messages {
1305            clock.set_time(ts_init);
1306            KrakenFuturesDataClient::handle_ws_message(
1307                msg,
1308                &sender,
1309                &instruments,
1310                &quote_instruments,
1311                &book_instruments,
1312                &mut order_books,
1313                &last_quotes,
1314                &book_sequence,
1315                clock,
1316            );
1317        }
1318
1319        let mut quotes = Vec::new();
1320        let mut book_deltas = Vec::new();
1321
1322        while let Ok(event) = rx.try_recv() {
1323            match event {
1324                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
1325                DataEvent::Data(Data::BookDeltas(deltas)) => book_deltas.push(*deltas),
1326                other => panic!("Unexpected event: {other:?}"),
1327            }
1328        }
1329
1330        let initial = QuoteTick::new(
1331            instrument_id,
1332            Price::from("34892.5"),
1333            Price::from("34911.5"),
1334            Quantity::from("6385"),
1335            Quantity::from("20598"),
1336            ts1,
1337            ts1,
1338        );
1339
1340        let mut changed = QuoteTick {
1341            ts_event: ts3,
1342            ts_init: ts3,
1343            ..initial
1344        };
1345
1346        match side {
1347            KrakenOrderSide::Buy => {
1348                changed.bid_price = Price::from_decimal_dp(price, 1).unwrap();
1349                changed.bid_size = Quantity::from_decimal_dp(qty, 0).unwrap();
1350            }
1351
1352            KrakenOrderSide::Sell => {
1353                changed.ask_price = Price::from_decimal_dp(price, 1).unwrap();
1354                changed.ask_size = Quantity::from_decimal_dp(qty, 0).unwrap();
1355            }
1356        }
1357
1358        assert_eq!(quotes, vec![initial, changed]);
1359
1360        // PartialEq omits batch content and nested order values
1361        assert_eq!(
1362            serde_json::to_value(&book_deltas).unwrap(),
1363            serde_json::to_value(&expected_deltas).unwrap(),
1364        );
1365    }
1366
1367    #[rstest]
1368    #[case::spread_boundary(dec!(125.0), true)]
1369    #[case::wide_spread(dec!(125.5), false)]
1370    fn test_quote_spread_filter_preserves_cached_quote_and_book_delivery(
1371        #[case] ask_price: Decimal,
1372        #[case] accepted: bool,
1373    ) {
1374        let instrument = futures_instrument();
1375        let instrument_id = instrument.id();
1376        let instruments = Arc::new(AtomicMap::new());
1377        instruments.insert(instrument_id, instrument.clone());
1378        let quote_instruments = Arc::new(AtomicSet::new());
1379        quote_instruments.insert(instrument_id);
1380        let book_instruments = Arc::new(AtomicSet::new());
1381        book_instruments.insert(instrument_id);
1382        let mut order_books = AHashMap::new();
1383        let last_quotes = DashMap::new();
1384        let sequence = Arc::new(AtomicU64::new(0));
1385        let clock = Box::leak(Box::new(AtomicTime::new(false, UnixNanos::new(10))));
1386        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1387        let sender = tx.into();
1388        let mut initial: KrakenFuturesBookSnapshot = serde_json::from_str(include_str!(
1389            "../../test_data/ws_futures_book_snapshot.json"
1390        ))
1391        .unwrap();
1392        initial.bids[0].price = dec!(100.0);
1393        initial.bids[1].price = dec!(99.5);
1394        initial.asks[0].price = dec!(101.0);
1395        initial.asks[1].price = dec!(101.5);
1396        let mut filtered = initial.clone();
1397        filtered.asks[0].price = ask_price;
1398        filtered.asks[1].price = ask_price + dec!(0.5);
1399        let mut changed = initial.clone();
1400        changed.bids[0].qty = dec!(7000);
1401        let mut expected_deltas = Vec::new();
1402        let mut next_sequence = 0;
1403
1404        for (snapshot, timestamp) in [(initial, 10), (filtered, 20), (changed, 30)] {
1405            let timestamp = UnixNanos::new(timestamp);
1406            let deltas = parse_futures_ws_book_snapshot_deltas(
1407                &snapshot,
1408                &instrument,
1409                next_sequence,
1410                timestamp,
1411            )
1412            .unwrap();
1413            next_sequence += deltas.len() as u64;
1414            expected_deltas.push(OrderBookDeltas::new(instrument_id, deltas));
1415            clock.set_time(timestamp);
1416            KrakenFuturesDataClient::handle_ws_message(
1417                KrakenFuturesWsMessage::BookSnapshot(snapshot),
1418                &sender,
1419                &instruments,
1420                &quote_instruments,
1421                &book_instruments,
1422                &mut order_books,
1423                &last_quotes,
1424                &sequence,
1425                clock,
1426            );
1427        }
1428
1429        let mut quotes = Vec::new();
1430        let mut deltas = Vec::new();
1431
1432        while let Ok(event) = rx.try_recv() {
1433            match event {
1434                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
1435                DataEvent::Data(Data::BookDeltas(batch)) => deltas.push(*batch),
1436                other => panic!("Unexpected event: {other:?}"),
1437            }
1438        }
1439
1440        let initial = QuoteTick::new(
1441            instrument_id,
1442            Price::from("100.0"),
1443            Price::from("101.0"),
1444            Quantity::from("6385"),
1445            Quantity::from("20598"),
1446            UnixNanos::new(10),
1447            UnixNanos::new(10),
1448        );
1449        let mut expected_quotes = vec![initial];
1450        if accepted {
1451            expected_quotes.push(QuoteTick {
1452                ask_price: Price::from_decimal_dp(ask_price, 1).unwrap(),
1453                ts_event: UnixNanos::new(20),
1454                ts_init: UnixNanos::new(20),
1455                ..initial
1456            });
1457        }
1458
1459        let changed = QuoteTick {
1460            bid_size: Quantity::from("7000"),
1461            ts_event: UnixNanos::new(30),
1462            ts_init: UnixNanos::new(30),
1463            ..initial
1464        };
1465
1466        expected_quotes.push(changed);
1467        assert_eq!(quotes, expected_quotes);
1468        assert_eq!(
1469            last_quotes.get(&instrument_id).map(|quote| *quote),
1470            Some(changed)
1471        );
1472        assert_eq!(
1473            serde_json::to_value(&deltas).unwrap(),
1474            serde_json::to_value(&expected_deltas).unwrap()
1475        );
1476    }
1477
1478    #[rstest]
1479    #[case::resubscribe(false, false)]
1480    #[case::resubscribe_after_late_quote(false, true)]
1481    #[case::reconnect(true, false)]
1482    #[tokio::test]
1483    async fn test_quote_subscription_emits_unchanged_initial_quote(
1484        #[case] reconnect: bool,
1485        #[case] late_quote: bool,
1486    ) {
1487        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1488        set_data_event_sender(tx);
1489
1490        let mut client = KrakenFuturesDataClient::new(
1491            *KRAKEN_CLIENT_ID,
1492            KrakenDataClientConfig {
1493                product_type: KrakenProductType::Futures,
1494                ..Default::default()
1495            },
1496        )
1497        .unwrap();
1498
1499        let instrument = futures_instrument();
1500        let instrument_id = instrument.id();
1501        client.instruments.insert(instrument_id, instrument);
1502        let ts1 = UnixNanos::new(1_700_000_000_000_000_000);
1503        let ts2 = UnixNanos::new(1_700_000_000_100_000_000);
1504        let ts3 = UnixNanos::new(1_700_000_000_200_000_000);
1505        let clock = Box::leak(Box::new(AtomicTime::new(false, ts1)));
1506        let mut order_books = AHashMap::new();
1507        let sequence = Arc::new(AtomicU64::new(0));
1508        let snapshot: KrakenFuturesBookSnapshot = serde_json::from_str(include_str!(
1509            "../../test_data/ws_futures_book_snapshot.json"
1510        ))
1511        .unwrap();
1512
1513        let initial = QuoteTick::new(
1514            instrument_id,
1515            Price::from("34892.5"),
1516            Price::from("34911.5"),
1517            Quantity::from("6385"),
1518            Quantity::from("20598"),
1519            ts1,
1520            ts1,
1521        );
1522
1523        let subscribe = || {
1524            SubscribeQuotes::new(
1525                instrument_id,
1526                Some(*KRAKEN_CLIENT_ID),
1527                None,
1528                UUID4::new(),
1529                UnixNanos::default(),
1530                None,
1531                None,
1532            )
1533        };
1534
1535        let mut handle = |client: &KrakenFuturesDataClient, msg| {
1536            KrakenFuturesDataClient::handle_ws_message(
1537                msg,
1538                &client.data_sender,
1539                &client.instruments,
1540                &client.quote_instruments,
1541                &client.book_instruments,
1542                &mut order_books,
1543                &client.last_quotes,
1544                &sequence,
1545                clock,
1546            );
1547        };
1548
1549        client.subscribe_quotes(subscribe()).unwrap();
1550        handle(
1551            &client,
1552            KrakenFuturesWsMessage::BookSnapshot(snapshot.clone()),
1553        );
1554
1555        if reconnect {
1556            handle(&client, KrakenFuturesWsMessage::Reconnected);
1557        } else {
1558            client
1559                .unsubscribe_quotes(&UnsubscribeQuotes::new(
1560                    instrument_id,
1561                    Some(*KRAKEN_CLIENT_ID),
1562                    None,
1563                    UUID4::new(),
1564                    UnixNanos::default(),
1565                    None,
1566                    None,
1567                ))
1568                .unwrap();
1569            clock.set_time(ts2);
1570            handle(
1571                &client,
1572                KrakenFuturesWsMessage::BookSnapshot(snapshot.clone()),
1573            );
1574
1575            if late_quote {
1576                client.last_quotes.insert(instrument_id, initial);
1577            }
1578
1579            client.subscribe_quotes(subscribe()).unwrap();
1580        }
1581
1582        clock.set_time(ts3);
1583        handle(&client, KrakenFuturesWsMessage::BookSnapshot(snapshot));
1584        let mut quotes = Vec::new();
1585
1586        while let Ok(event) = rx.try_recv() {
1587            match event {
1588                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
1589                other => panic!("Unexpected event: {other:?}"),
1590            }
1591        }
1592
1593        assert_eq!(
1594            quotes,
1595            vec![
1596                initial,
1597                QuoteTick {
1598                    ts_event: ts3,
1599                    ts_init: ts3,
1600                    ..initial
1601                }
1602            ]
1603        );
1604    }
1605
1606    #[rstest]
1607    #[case::joining_book_subscription(false)]
1608    #[case::returning_to_book_subscription(true)]
1609    #[tokio::test]
1610    async fn test_quote_subscription_uses_current_book(#[case] initial_quotes: bool) {
1611        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
1612        set_data_event_sender(tx);
1613
1614        let mut client = KrakenFuturesDataClient::new(
1615            *KRAKEN_CLIENT_ID,
1616            KrakenDataClientConfig {
1617                product_type: KrakenProductType::Futures,
1618                ..Default::default()
1619            },
1620        )
1621        .unwrap();
1622
1623        let instrument = futures_instrument();
1624        let instrument_id = instrument.id();
1625        client.instruments.insert(instrument_id, instrument.clone());
1626        client.book_instruments.insert(instrument_id);
1627        let ts1 = UnixNanos::new(1_700_000_000_000_000_000);
1628        let ts2 = UnixNanos::new(1_700_000_000_100_000_000);
1629        let ts3 = UnixNanos::new(1_700_000_000_200_000_000);
1630        let clock = Box::leak(Box::new(AtomicTime::new(false, ts1)));
1631        let mut order_books = AHashMap::new();
1632        let sequence = Arc::new(AtomicU64::new(0));
1633        let snapshot: KrakenFuturesBookSnapshot = serde_json::from_str(include_str!(
1634            "../../test_data/ws_futures_book_snapshot.json"
1635        ))
1636        .unwrap();
1637
1638        let messages = [
1639            KrakenFuturesWsMessage::BookSnapshot(snapshot),
1640            KrakenFuturesWsMessage::BookDelta(KrakenFuturesBookDelta {
1641                feed: KrakenFuturesFeed::Book,
1642                product_id: Ustr::from("PI_XBTUSD"),
1643                side: KrakenOrderSide::Buy,
1644                seq: 326_072_250,
1645                price: dec!(34892.5),
1646                qty: dec!(7000),
1647                timestamp: 1_612_269_825_900,
1648            }),
1649            KrakenFuturesWsMessage::BookDelta(KrakenFuturesBookDelta {
1650                feed: KrakenFuturesFeed::Book,
1651                product_id: Ustr::from("PI_XBTUSD"),
1652                side: KrakenOrderSide::Buy,
1653                seq: 326_072_251,
1654                price: dec!(34892),
1655                qty: dec!(12000),
1656                timestamp: 1_612_269_826_000,
1657            }),
1658        ];
1659
1660        let mut expected_deltas = Vec::new();
1661        let mut next_sequence = 0;
1662
1663        for (message, ts) in messages.iter().zip([ts1, ts2, ts3]) {
1664            let deltas = match message {
1665                KrakenFuturesWsMessage::BookSnapshot(snapshot) => {
1666                    parse_futures_ws_book_snapshot_deltas(snapshot, &instrument, next_sequence, ts)
1667                        .unwrap()
1668                }
1669
1670                KrakenFuturesWsMessage::BookDelta(delta) => vec![
1671                    parse_futures_ws_book_delta(delta, &instrument, next_sequence, ts).unwrap(),
1672                ],
1673                other => panic!("Unexpected message: {other:?}"),
1674            };
1675
1676            next_sequence += deltas.len() as u64;
1677            expected_deltas.push(OrderBookDeltas::new(instrument_id, deltas));
1678        }
1679
1680        let subscribe = || {
1681            SubscribeQuotes::new(
1682                instrument_id,
1683                Some(*KRAKEN_CLIENT_ID),
1684                None,
1685                UUID4::new(),
1686                UnixNanos::default(),
1687                None,
1688                None,
1689            )
1690        };
1691
1692        if initial_quotes {
1693            client.subscribe_quotes(subscribe()).unwrap();
1694        }
1695
1696        for (index, message) in messages.into_iter().enumerate() {
1697            if index == 1 && initial_quotes {
1698                client
1699                    .unsubscribe_quotes(&UnsubscribeQuotes::new(
1700                        instrument_id,
1701                        Some(*KRAKEN_CLIENT_ID),
1702                        None,
1703                        UUID4::new(),
1704                        UnixNanos::default(),
1705                        None,
1706                        None,
1707                    ))
1708                    .unwrap();
1709            }
1710
1711            if index == 2 {
1712                client.subscribe_quotes(subscribe()).unwrap();
1713            }
1714
1715            clock.set_time([ts1, ts2, ts3][index]);
1716            KrakenFuturesDataClient::handle_ws_message(
1717                message,
1718                &client.data_sender,
1719                &client.instruments,
1720                &client.quote_instruments,
1721                &client.book_instruments,
1722                &mut order_books,
1723                &client.last_quotes,
1724                &sequence,
1725                clock,
1726            );
1727        }
1728
1729        let mut quotes = Vec::new();
1730        let mut deltas = Vec::new();
1731
1732        while let Ok(event) = rx.try_recv() {
1733            match event {
1734                DataEvent::Data(Data::Quote(quote)) => quotes.push(quote),
1735                DataEvent::Data(Data::BookDeltas(batch)) => deltas.push(*batch),
1736                other => panic!("Unexpected event: {other:?}"),
1737            }
1738        }
1739
1740        let initial = QuoteTick::new(
1741            instrument_id,
1742            Price::from("34892.5"),
1743            Price::from("34911.5"),
1744            Quantity::from("6385"),
1745            Quantity::from("20598"),
1746            ts1,
1747            ts1,
1748        );
1749
1750        let current = QuoteTick {
1751            bid_size: Quantity::from("7000"),
1752            ts_event: ts3,
1753            ts_init: ts3,
1754            ..initial
1755        };
1756
1757        let expected_quotes = if initial_quotes {
1758            vec![initial, current]
1759        } else {
1760            vec![current]
1761        };
1762
1763        assert_eq!(quotes, expected_quotes);
1764        assert_eq!(
1765            serde_json::to_value(&deltas).unwrap(),
1766            serde_json::to_value(&expected_deltas).unwrap()
1767        );
1768    }
1769
1770    fn futures_instrument() -> InstrumentAny {
1771        let response: FuturesInstrumentsResponse = serde_json::from_str(include_str!(
1772            "../../test_data/http_futures_instruments.json"
1773        ))
1774        .unwrap();
1775        parse_futures_instrument(
1776            &response.instruments[0],
1777            UnixNanos::default(),
1778            UnixNanos::default(),
1779        )
1780        .unwrap()
1781    }
1782}
nautilus_kraken/data/futures.rs

nautilus_kraken/data/
futures.rs
