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

nautilus_kraken/data/
futures.rs
