1use std::{
19 future::Future,
20 sync::{
21 Arc,
22 atomic::{AtomicBool, AtomicU64, Ordering},
23 },
24};
25
26use ahash::AHashMap;
27use anyhow::Context;
28use async_trait::async_trait;
29use nautilus_common::{
30 clients::DataClient,
31 live::{get_data_event_sender, get_runtime},
32 messages::{
33 DataEvent,
34 data::{
35 BarsResponse, BookResponse, DataResponse, FundingRatesResponse, InstrumentResponse,
36 InstrumentsResponse, RequestBars, RequestBookSnapshot, RequestFundingRates,
37 RequestInstrument, RequestInstruments, RequestTrades, SubscribeBars,
38 SubscribeBookDeltas, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
39 SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices, SubscribeQuotes,
40 SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
41 UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrumentStatus,
42 UnsubscribeMarkPrices, UnsubscribeQuotes, UnsubscribeTrades,
43 },
44 },
45};
46use nautilus_core::{
47 AtomicMap, AtomicSet,
48 datetime::datetime_to_unix_nanos,
49 nanos::UnixNanos,
50 time::{AtomicTime, get_atomic_clock_realtime},
51};
52use nautilus_model::{
53 data::{Data, OrderBookDeltas, OrderBookDeltas_API, QuoteTick},
54 enums::BookType,
55 identifiers::{ClientId, InstrumentId, Venue},
56 instruments::{Instrument, InstrumentAny},
57 orderbook::OrderBook,
58};
59use tokio::task::JoinHandle;
60use tokio_util::sync::CancellationToken;
61
62use crate::{
63 common::{consts::KRAKEN_VENUE, lookup_instrument_in_snapshot},
64 config::KrakenDataClientConfig,
65 http::{
66 KrakenFuturesHttpClient, futures::client::KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND,
67 },
68 websocket::futures::{
69 client::KrakenFuturesWebSocketClient,
70 messages::KrakenFuturesWsMessage,
71 parse::{
72 parse_futures_ws_book_delta, parse_futures_ws_book_snapshot_deltas,
73 parse_futures_ws_funding_rate, parse_futures_ws_index_price,
74 parse_futures_ws_mark_price, parse_futures_ws_trade_tick,
75 },
76 },
77};
78
79#[allow(dead_code)]
83#[derive(Debug)]
84pub struct KrakenFuturesDataClient {
85 clock: &'static AtomicTime,
86 client_id: ClientId,
87 config: KrakenDataClientConfig,
88 http: KrakenFuturesHttpClient,
89 ws: KrakenFuturesWebSocketClient,
90 is_connected: AtomicBool,
91 cancellation_token: CancellationToken,
92 tasks: Vec<JoinHandle<()>>,
93 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
94 quote_instruments: Arc<AtomicSet<InstrumentId>>,
95 book_instruments: Arc<AtomicSet<InstrumentId>>,
96 data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
97}
98
99impl KrakenFuturesDataClient {
100 pub fn new(client_id: ClientId, config: KrakenDataClientConfig) -> anyhow::Result<Self> {
102 let cancellation_token = CancellationToken::new();
103
104 let http = KrakenFuturesHttpClient::new(
105 config.environment,
106 config.base_url.clone(),
107 config.timeout_secs,
108 None,
109 None,
110 None,
111 config.proxy_url.clone(),
112 config
113 .max_requests_per_second
114 .unwrap_or(KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND),
115 )?;
116
117 let ws = KrakenFuturesWebSocketClient::with_credentials(
118 config.ws_public_url(),
119 config.heartbeat_interval_secs,
120 None,
121 config.transport_backend,
122 config.proxy_url.clone(),
123 );
124
125 Ok(Self {
126 clock: get_atomic_clock_realtime(),
127 client_id,
128 config,
129 http,
130 ws,
131 is_connected: AtomicBool::new(false),
132 cancellation_token,
133 tasks: Vec::new(),
134 instruments: Arc::new(AtomicMap::new()),
135 quote_instruments: Arc::new(AtomicSet::new()),
136 book_instruments: Arc::new(AtomicSet::new()),
137 data_sender: get_data_event_sender(),
138 })
139 }
140
141 #[must_use]
143 pub fn instruments(&self) -> Vec<InstrumentAny> {
144 self.instruments.load().values().cloned().collect()
145 }
146
147 #[must_use]
149 pub fn get_instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
150 self.instruments.load().get(instrument_id).cloned()
151 }
152
153 async fn load_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
154 let instruments = self
155 .http
156 .request_instruments()
157 .await
158 .context("Failed to load futures instruments")?;
159
160 self.instruments.rcu(|m| {
161 for instrument in &instruments {
162 m.insert(instrument.id(), instrument.clone());
163 }
164 });
165
166 self.http.cache_instruments(&instruments);
167
168 log::debug!(
169 "Loaded instruments: client_id={}, count={}",
170 self.client_id,
171 instruments.len()
172 );
173
174 Ok(instruments)
175 }
176
177 fn spawn_ws<F>(&self, fut: F, context: &'static str)
178 where
179 F: Future<Output = anyhow::Result<()>> + Send + 'static,
180 {
181 get_runtime().spawn(async move {
182 if let Err(e) = fut.await {
183 log::error!("{context}: {e:?}");
184 }
185 });
186 }
187
188 fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
189 let mut rx = self
190 .ws
191 .take_output_rx()
192 .context("Failed to take futures WebSocket output receiver")?;
193 let data_sender = self.data_sender.clone();
194 let instruments = self.instruments.clone();
195 let quote_instruments = self.quote_instruments.clone();
196 let book_instruments = self.book_instruments.clone();
197 let book_sequence = Arc::new(AtomicU64::new(0));
198 let cancellation_token = self.cancellation_token.clone();
199 let clock = self.clock;
200
201 let handle = get_runtime().spawn(async move {
202 let mut order_books: AHashMap<InstrumentId, OrderBook> = AHashMap::new();
203 let mut last_quotes: AHashMap<InstrumentId, QuoteTick> = AHashMap::new();
204
205 loop {
206 tokio::select! {
207 () = cancellation_token.cancelled() => {
208 log::debug!("Futures message handler cancelled");
209 break;
210 }
211 msg = rx.recv() => {
212 match msg {
213 Some(ws_msg) => {
214 Self::handle_ws_message(
215 ws_msg,
216 &data_sender,
217 &instruments,
218 "e_instruments,
219 &book_instruments,
220 &mut order_books,
221 &mut last_quotes,
222 &book_sequence,
223 clock,
224 );
225 }
226 None => {
227 log::debug!("Futures WebSocket stream ended");
228 break;
229 }
230 }
231 }
232 }
233 }
234 });
235
236 self.tasks.push(handle);
237 Ok(())
238 }
239
240 #[expect(clippy::too_many_arguments)]
241 fn handle_ws_message(
242 msg: KrakenFuturesWsMessage,
243 sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
244 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
245 quote_instruments: &Arc<AtomicSet<InstrumentId>>,
246 book_instruments: &Arc<AtomicSet<InstrumentId>>,
247 order_books: &mut AHashMap<InstrumentId, OrderBook>,
248 last_quotes: &mut AHashMap<InstrumentId, QuoteTick>,
249 book_sequence: &Arc<AtomicU64>,
250 clock: &'static AtomicTime,
251 ) {
252 let ts_init = clock.get_time_ns();
253
254 match msg {
255 KrakenFuturesWsMessage::Ticker(ticker) => {
256 let instruments = instruments.load();
257 let Some(instrument) =
258 lookup_instrument_in_snapshot(&instruments, ticker.product_id.as_str())
259 else {
260 log::warn!("No instrument for product_id: {}", ticker.product_id);
261 return;
262 };
263
264 if let Some(mark) = parse_futures_ws_mark_price(&ticker, instrument, ts_init)
265 && let Err(e) = sender.send(DataEvent::Data(Data::MarkPriceUpdate(mark)))
266 {
267 log::error!("Failed to send mark price: {e}");
268 }
269
270 if let Some(index) = parse_futures_ws_index_price(&ticker, instrument, ts_init)
271 && let Err(e) = sender.send(DataEvent::Data(Data::IndexPriceUpdate(index)))
272 {
273 log::error!("Failed to send index price: {e}");
274 }
275
276 if let Some(funding) = parse_futures_ws_funding_rate(&ticker, instrument, ts_init)
277 && let Err(e) = sender.send(DataEvent::FundingRate(funding))
278 {
279 log::error!("Failed to send funding rate: {e}");
280 }
281 }
282 KrakenFuturesWsMessage::Trade(trade) => {
283 let instruments = instruments.load();
284 let Some(instrument) =
285 lookup_instrument_in_snapshot(&instruments, trade.product_id.as_str())
286 else {
287 log::warn!("No instrument for product_id: {}", trade.product_id);
288 return;
289 };
290
291 match parse_futures_ws_trade_tick(&trade, instrument, ts_init) {
292 Ok(tick) => {
293 if let Err(e) = sender.send(DataEvent::Data(Data::Trade(tick))) {
294 log::error!("Failed to send trade: {e}");
295 }
296 }
297 Err(e) => log::error!("Failed to parse futures trade tick: {e}"),
298 }
299 }
300 KrakenFuturesWsMessage::BookSnapshot(snapshot) => {
301 let instruments = instruments.load();
302 let Some(instrument) =
303 lookup_instrument_in_snapshot(&instruments, snapshot.product_id.as_str())
304 else {
305 log::warn!("No instrument for product_id: {}", snapshot.product_id);
306 return;
307 };
308 let instrument_id = instrument.id();
309 let sequence = book_sequence.load(Ordering::Relaxed);
310
311 match parse_futures_ws_book_snapshot_deltas(
312 &snapshot, instrument, sequence, ts_init,
313 ) {
314 Ok(delta_vec) => {
315 if delta_vec.is_empty() {
316 return;
317 }
318 book_sequence.fetch_add(delta_vec.len() as u64, Ordering::Relaxed);
319 let deltas = OrderBookDeltas::new(instrument_id, delta_vec);
320
321 let has_quote_sub = quote_instruments.contains(&instrument_id);
322
323 if has_quote_sub {
324 let book = order_books
325 .entry(instrument_id)
326 .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
327
328 if let Err(e) = book.apply_deltas(&deltas) {
329 log::error!("Failed to apply snapshot deltas to order book: {e}");
330 } else {
331 Self::maybe_emit_quote(
332 book,
333 instrument_id,
334 last_quotes,
335 ts_init,
336 sender,
337 );
338 }
339 }
340
341 let has_book_sub = book_instruments.contains(&instrument_id);
342
343 if has_book_sub {
344 let api_deltas = OrderBookDeltas_API::new(deltas);
345 if let Err(e) = sender.send(DataEvent::Data(Data::Deltas(api_deltas))) {
346 log::error!("Failed to send book snapshot deltas: {e}");
347 }
348 }
349 }
350 Err(e) => log::error!("Failed to parse book snapshot: {e}"),
351 }
352 }
353 KrakenFuturesWsMessage::BookDelta(delta) => {
354 let instruments = instruments.load();
355 let Some(instrument) =
356 lookup_instrument_in_snapshot(&instruments, delta.product_id.as_str())
357 else {
358 log::warn!("No instrument for product_id: {}", delta.product_id);
359 return;
360 };
361 let instrument_id = instrument.id();
362 let sequence = book_sequence.fetch_add(1, Ordering::Relaxed);
363 match parse_futures_ws_book_delta(&delta, instrument, sequence, ts_init) {
364 Ok(book_delta) => {
365 let deltas = OrderBookDeltas::new(instrument_id, vec![book_delta]);
366
367 let has_quote_sub = quote_instruments.contains(&instrument_id);
368
369 if has_quote_sub && let Some(book) = order_books.get_mut(&instrument_id) {
370 if let Err(e) = book.apply_deltas(&deltas) {
371 log::error!("Failed to apply delta to order book: {e}");
372 } else {
373 Self::maybe_emit_quote(
374 book,
375 instrument_id,
376 last_quotes,
377 ts_init,
378 sender,
379 );
380 }
381 }
382
383 let has_book_sub = book_instruments.contains(&instrument_id);
384
385 if has_book_sub {
386 let api_deltas = OrderBookDeltas_API::new(deltas);
387 if let Err(e) = sender.send(DataEvent::Data(Data::Deltas(api_deltas))) {
388 log::error!("Failed to send book delta: {e}");
389 }
390 }
391 }
392 Err(e) => log::error!("Failed to parse book delta: {e}"),
393 }
394 }
395 KrakenFuturesWsMessage::Reconnected => {
396 log::info!("Futures WebSocket reconnected");
397 }
398 KrakenFuturesWsMessage::OpenOrdersCancel(_)
399 | KrakenFuturesWsMessage::OpenOrdersDelta(_)
400 | KrakenFuturesWsMessage::FillsDelta(_)
401 | KrakenFuturesWsMessage::Challenge(_) => {}
402 }
403 }
404
405 fn maybe_emit_quote(
406 book: &OrderBook,
407 instrument_id: InstrumentId,
408 last_quotes: &mut AHashMap<InstrumentId, QuoteTick>,
409 ts_init: UnixNanos,
410 sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
411 ) {
412 let (Some(bid_price), Some(ask_price)) = (book.best_bid_price(), book.best_ask_price())
413 else {
414 return;
415 };
416 let (Some(bid_size), Some(ask_size)) = (book.best_bid_size(), book.best_ask_size()) else {
417 return;
418 };
419
420 let bid = bid_price.as_f64();
421 let ask = ask_price.as_f64();
422 if bid > 0.0 && (ask - bid) / bid > 0.25 {
423 log::debug!("Filtered quote with wide spread: bid={bid}, ask={ask}");
424 return;
425 }
426
427 let quote = QuoteTick::new(
428 instrument_id,
429 bid_price,
430 ask_price,
431 bid_size,
432 ask_size,
433 ts_init,
434 ts_init,
435 );
436
437 if matches!(last_quotes.get(&instrument_id), Some(prev) if *prev == quote) {
438 return;
439 }
440
441 last_quotes.insert(instrument_id, quote);
442
443 if let Err(e) = sender.send(DataEvent::Data(Data::Quote(quote))) {
444 log::error!("Failed to send quote: {e}");
445 }
446 }
447}
448
449#[async_trait(?Send)]
450impl DataClient for KrakenFuturesDataClient {
451 fn client_id(&self) -> ClientId {
452 self.client_id
453 }
454
455 fn venue(&self) -> Option<Venue> {
456 Some(*KRAKEN_VENUE)
457 }
458
459 fn start(&mut self) -> anyhow::Result<()> {
460 log::info!(
461 "Starting Futures data client: client_id={}, environment={:?}",
462 self.client_id,
463 self.config.environment
464 );
465 Ok(())
466 }
467
468 fn stop(&mut self) -> anyhow::Result<()> {
469 log::info!("Stopping Futures data client: {}", self.client_id);
470 self.cancellation_token.cancel();
471 self.is_connected.store(false, Ordering::Relaxed);
472 Ok(())
473 }
474
475 fn reset(&mut self) -> anyhow::Result<()> {
476 log::info!("Resetting Futures data client: {}", self.client_id);
477 self.cancellation_token.cancel();
478
479 for task in self.tasks.drain(..) {
480 task.abort();
481 }
482
483 let mut ws = self.ws.clone();
484 get_runtime().spawn(async move {
485 let _ = ws.close().await;
486 });
487
488 self.instruments.store(ahash::AHashMap::new());
489
490 self.quote_instruments.store(ahash::AHashSet::new());
491
492 self.is_connected.store(false, Ordering::Relaxed);
493 self.cancellation_token = CancellationToken::new();
494 Ok(())
495 }
496
497 fn dispose(&mut self) -> anyhow::Result<()> {
498 log::debug!("Disposing Futures data client: {}", self.client_id);
499 self.stop()
500 }
501
502 fn is_connected(&self) -> bool {
503 self.is_connected.load(Ordering::SeqCst)
504 }
505
506 fn is_disconnected(&self) -> bool {
507 !self.is_connected()
508 }
509
510 async fn connect(&mut self) -> anyhow::Result<()> {
511 if self.is_connected() {
512 return Ok(());
513 }
514
515 let instruments = self.load_instruments().await?;
516
517 self.ws
518 .connect()
519 .await
520 .context("Failed to connect futures WebSocket")?;
521 self.ws
522 .wait_until_active(10.0)
523 .await
524 .context("Futures WebSocket failed to become active")?;
525
526 self.spawn_message_handler()?;
527
528 for instrument in instruments {
529 if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
530 log::error!("Failed to send instrument: {e}");
531 }
532 }
533
534 self.is_connected.store(true, Ordering::Release);
535 log::info!(
536 "Connected: client_id={}, product_type=Futures",
537 self.client_id
538 );
539 Ok(())
540 }
541
542 async fn disconnect(&mut self) -> anyhow::Result<()> {
543 if self.is_disconnected() {
544 return Ok(());
545 }
546
547 self.cancellation_token.cancel();
548 let _ = self.ws.close().await;
549
550 for handle in self.tasks.drain(..) {
551 if let Err(e) = handle.await {
552 log::error!("Error joining WebSocket task: {e:?}");
553 }
554 }
555
556 self.cancellation_token = CancellationToken::new();
557
558 self.quote_instruments.store(ahash::AHashSet::new());
559 self.is_connected.store(false, Ordering::Relaxed);
560
561 log::info!("Disconnected: client_id={}", self.client_id);
562 Ok(())
563 }
564
565 fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
566 log::debug!("subscribe_instruments: Kraken instruments are fetched via HTTP on connect");
567 Ok(())
568 }
569
570 fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
571 log::debug!("subscribe_instrument: Kraken instruments are fetched via HTTP on connect");
572 Ok(())
573 }
574
575 fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
576 let instrument_id = cmd.instrument_id;
577 let depth = cmd.depth;
578
579 if cmd.book_type != BookType::L2_MBP {
580 log::warn!(
581 "Book type {:?} not supported by Kraken, skipping subscription",
582 cmd.book_type
583 );
584 return Ok(());
585 }
586
587 self.book_instruments.insert(instrument_id);
588
589 let ws = self.ws.clone();
590 self.spawn_ws(
591 async move {
592 ws.subscribe_book(instrument_id, depth.map(|d| d.get() as u32))
593 .await
594 .map_err(|e| anyhow::anyhow!("{e}"))
595 },
596 "subscribe book",
597 );
598
599 Ok(())
600 }
601
602 fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
603 let instrument_id = cmd.instrument_id;
604 let ws = self.ws.clone();
605
606 self.quote_instruments.insert(instrument_id);
607
608 self.spawn_ws(
609 async move {
610 ws.subscribe_quotes(instrument_id)
611 .await
612 .map_err(|e| anyhow::anyhow!("{e}"))
613 },
614 "subscribe quotes",
615 );
616
617 Ok(())
618 }
619
620 fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
621 let instrument_id = cmd.instrument_id;
622 let ws = self.ws.clone();
623
624 self.spawn_ws(
625 async move {
626 ws.subscribe_trades(instrument_id)
627 .await
628 .map_err(|e| anyhow::anyhow!("{e}"))
629 },
630 "subscribe trades",
631 );
632
633 Ok(())
634 }
635
636 fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
637 let instrument_id = cmd.instrument_id;
638 let ws = self.ws.clone();
639
640 self.spawn_ws(
641 async move {
642 ws.subscribe_mark_price(instrument_id)
643 .await
644 .map_err(|e| anyhow::anyhow!("{e}"))
645 },
646 "subscribe mark price",
647 );
648
649 Ok(())
650 }
651
652 fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
653 let instrument_id = cmd.instrument_id;
654 let ws = self.ws.clone();
655
656 self.spawn_ws(
657 async move {
658 ws.subscribe_index_price(instrument_id)
659 .await
660 .map_err(|e| anyhow::anyhow!("{e}"))
661 },
662 "subscribe index price",
663 );
664
665 Ok(())
666 }
667
668 fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
669 let instrument_id = cmd.instrument_id;
670 let ws = self.ws.clone();
671
672 self.spawn_ws(
673 async move {
674 ws.subscribe_funding_rate(instrument_id)
675 .await
676 .map_err(|e| anyhow::anyhow!("{e}"))
677 },
678 "subscribe funding rate",
679 );
680
681 Ok(())
682 }
683
684 fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
685 log::warn!(
686 "Cannot subscribe to {} bars: Kraken Futures does not support EXTERNAL bar streaming",
687 cmd.bar_type
688 );
689 Ok(())
690 }
691
692 fn subscribe_instrument_status(
693 &mut self,
694 cmd: SubscribeInstrumentStatus,
695 ) -> anyhow::Result<()> {
696 log::debug!(
697 "subscribe_instrument_status: {} (status changes detected via periodic instrument polling)",
698 cmd.instrument_id,
699 );
700 Ok(())
701 }
702
703 fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
704 let instrument_id = cmd.instrument_id;
705
706 self.book_instruments.remove(&instrument_id);
707
708 let ws = self.ws.clone();
709 self.spawn_ws(
710 async move {
711 ws.unsubscribe_book(instrument_id)
712 .await
713 .map_err(|e| anyhow::anyhow!("{e}"))
714 },
715 "unsubscribe book",
716 );
717
718 Ok(())
719 }
720
721 fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
722 let instrument_id = cmd.instrument_id;
723 let ws = self.ws.clone();
724
725 self.quote_instruments.remove(&instrument_id);
726
727 self.spawn_ws(
728 async move {
729 ws.unsubscribe_quotes(instrument_id)
730 .await
731 .map_err(|e| anyhow::anyhow!("{e}"))
732 },
733 "unsubscribe quotes",
734 );
735
736 Ok(())
737 }
738
739 fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
740 let instrument_id = cmd.instrument_id;
741 let ws = self.ws.clone();
742
743 self.spawn_ws(
744 async move {
745 ws.unsubscribe_trades(instrument_id)
746 .await
747 .map_err(|e| anyhow::anyhow!("{e}"))
748 },
749 "unsubscribe trades",
750 );
751
752 Ok(())
753 }
754
755 fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
756 let instrument_id = cmd.instrument_id;
757 let ws = self.ws.clone();
758
759 self.spawn_ws(
760 async move {
761 ws.unsubscribe_mark_price(instrument_id)
762 .await
763 .map_err(|e| anyhow::anyhow!("{e}"))
764 },
765 "unsubscribe mark price",
766 );
767
768 Ok(())
769 }
770
771 fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
772 let instrument_id = cmd.instrument_id;
773 let ws = self.ws.clone();
774
775 self.spawn_ws(
776 async move {
777 ws.unsubscribe_index_price(instrument_id)
778 .await
779 .map_err(|e| anyhow::anyhow!("{e}"))
780 },
781 "unsubscribe index price",
782 );
783
784 Ok(())
785 }
786
787 fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
788 let instrument_id = cmd.instrument_id;
789 let ws = self.ws.clone();
790
791 self.spawn_ws(
792 async move {
793 ws.unsubscribe_funding_rate(instrument_id)
794 .await
795 .map_err(|e| anyhow::anyhow!("{e}"))
796 },
797 "unsubscribe funding rate",
798 );
799
800 Ok(())
801 }
802
803 fn unsubscribe_bars(&mut self, _cmd: &UnsubscribeBars) -> anyhow::Result<()> {
804 Ok(())
805 }
806
807 fn unsubscribe_instrument_status(
808 &mut self,
809 _cmd: &UnsubscribeInstrumentStatus,
810 ) -> anyhow::Result<()> {
811 Ok(())
812 }
813
814 fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
815 let http = self.http.clone();
816 let sender = self.data_sender.clone();
817 let instruments_cache = self.instruments.clone();
818 let request_id = request.request_id;
819 let client_id = request.client_id.unwrap_or(self.client_id);
820 let venue = *KRAKEN_VENUE;
821 let start_nanos = datetime_to_unix_nanos(request.start);
822 let end_nanos = datetime_to_unix_nanos(request.end);
823 let params = request.params;
824 let clock = self.clock;
825
826 get_runtime().spawn(async move {
827 match http.request_instruments().await {
828 Ok(instruments) => {
829 instruments_cache.rcu(|m| {
830 for instrument in &instruments {
831 m.insert(instrument.id(), instrument.clone());
832 }
833 });
834 http.cache_instruments(&instruments);
835
836 let response = DataResponse::Instruments(InstrumentsResponse::new(
837 request_id,
838 client_id,
839 venue,
840 instruments,
841 start_nanos,
842 end_nanos,
843 clock.get_time_ns(),
844 params,
845 ));
846
847 if let Err(e) = sender.send(DataEvent::Response(response)) {
848 log::error!("Failed to send instruments response: {e}");
849 }
850 }
851 Err(e) => log::error!("Instruments request failed: {e:?}"),
852 }
853 });
854
855 Ok(())
856 }
857
858 fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
859 let http = self.http.clone();
860 let sender = self.data_sender.clone();
861 let instruments = self.instruments.clone();
862 let instrument_id = request.instrument_id;
863 let request_id = request.request_id;
864 let client_id = request.client_id.unwrap_or(self.client_id);
865 let start_nanos = datetime_to_unix_nanos(request.start);
866 let end_nanos = datetime_to_unix_nanos(request.end);
867 let params = request.params;
868 let clock = self.clock;
869
870 get_runtime().spawn(async move {
871 match http.request_instruments().await {
872 Ok(all_instruments) => {
873 instruments.rcu(|m| {
874 for instrument in &all_instruments {
875 m.insert(instrument.id(), instrument.clone());
876 }
877 });
878 http.cache_instruments(&all_instruments);
879
880 let instrument = all_instruments
881 .into_iter()
882 .find(|i| i.id() == instrument_id);
883
884 if let Some(instrument) = instrument {
885 let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
886 request_id,
887 client_id,
888 instrument.id(),
889 instrument,
890 start_nanos,
891 end_nanos,
892 clock.get_time_ns(),
893 params,
894 )));
895
896 if let Err(e) = sender.send(DataEvent::Response(response)) {
897 log::error!("Failed to send instrument response: {e}");
898 }
899 } else {
900 log::error!("Instrument not found: {instrument_id}");
901 }
902 }
903 Err(e) => log::error!("Instrument request failed: {e:?}"),
904 }
905 });
906
907 Ok(())
908 }
909
910 fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
911 let http = self.http.clone();
912 let sender = self.data_sender.clone();
913 let instrument_id = request.instrument_id;
914 let start = request.start;
915 let end = request.end;
916 let limit = request.limit.map(|n| n.get() as u64);
917 let request_id = request.request_id;
918 let client_id = request.client_id.unwrap_or(self.client_id);
919 let params = request.params;
920 let clock = self.clock;
921 let start_nanos = datetime_to_unix_nanos(start);
922 let end_nanos = datetime_to_unix_nanos(end);
923
924 get_runtime().spawn(async move {
925 match http.request_trades(instrument_id, start, end, limit).await {
926 Ok(trades) => {
927 let response = DataResponse::Trades(TradesResponse::new(
928 request_id,
929 client_id,
930 instrument_id,
931 trades,
932 start_nanos,
933 end_nanos,
934 clock.get_time_ns(),
935 params,
936 ));
937
938 if let Err(e) = sender.send(DataEvent::Response(response)) {
939 log::error!("Failed to send trades response: {e}");
940 }
941 }
942 Err(e) => log::error!("Trades request failed: {e:?}"),
943 }
944 });
945
946 Ok(())
947 }
948
949 fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
950 let http = self.http.clone();
951 let sender = self.data_sender.clone();
952 let bar_type = request.bar_type;
953 let start = request.start;
954 let end = request.end;
955 let limit = request.limit.map(|n| n.get() as u64);
956 let request_id = request.request_id;
957 let client_id = request.client_id.unwrap_or(self.client_id);
958 let params = request.params;
959 let clock = self.clock;
960 let start_nanos = datetime_to_unix_nanos(start);
961 let end_nanos = datetime_to_unix_nanos(end);
962
963 get_runtime().spawn(async move {
964 match http.request_bars(bar_type, start, end, limit).await {
965 Ok(bars) => {
966 let response = DataResponse::Bars(BarsResponse::new(
967 request_id,
968 client_id,
969 bar_type,
970 bars,
971 start_nanos,
972 end_nanos,
973 clock.get_time_ns(),
974 params,
975 ));
976
977 if let Err(e) = sender.send(DataEvent::Response(response)) {
978 log::error!("Failed to send bars response: {e}");
979 }
980 }
981 Err(e) => log::error!("Bars request failed: {e:?}"),
982 }
983 });
984
985 Ok(())
986 }
987
988 fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
989 let http = self.http.clone();
990 let sender = self.data_sender.clone();
991 let instrument_id = request.instrument_id;
992 let depth = request.depth.map(|n| n.get() as u32);
993 let request_id = request.request_id;
994 let client_id = request.client_id.unwrap_or(self.client_id);
995 let params = request.params;
996 let clock = self.clock;
997
998 get_runtime().spawn(async move {
999 match http.request_book_snapshot(instrument_id, depth).await {
1000 Ok(book) => {
1001 let response = DataResponse::Book(BookResponse::new(
1002 request_id,
1003 client_id,
1004 instrument_id,
1005 book,
1006 None,
1007 None,
1008 clock.get_time_ns(),
1009 params,
1010 ));
1011
1012 if let Err(e) = sender.send(DataEvent::Response(response)) {
1013 log::error!("Failed to send book snapshot response: {e}");
1014 }
1015 }
1016 Err(e) => log::error!("Book snapshot request failed: {e:?}"),
1017 }
1018 });
1019
1020 Ok(())
1021 }
1022
1023 fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1024 let http = self.http.clone();
1025 let sender = self.data_sender.clone();
1026 let instrument_id = request.instrument_id;
1027 let start = request.start;
1028 let end = request.end;
1029 let limit = request.limit.map(|n| n.get());
1030 let request_id = request.request_id;
1031 let client_id = request.client_id.unwrap_or(self.client_id);
1032 let start_nanos = datetime_to_unix_nanos(start);
1033 let end_nanos = datetime_to_unix_nanos(end);
1034 let params = request.params;
1035 let clock = self.clock;
1036
1037 get_runtime().spawn(async move {
1038 match http
1039 .request_funding_rates(instrument_id, start, end, limit)
1040 .await
1041 {
1042 Ok(rates) => {
1043 let response = DataResponse::FundingRates(FundingRatesResponse::new(
1044 request_id,
1045 client_id,
1046 instrument_id,
1047 rates,
1048 start_nanos,
1049 end_nanos,
1050 clock.get_time_ns(),
1051 params,
1052 ));
1053
1054 if let Err(e) = sender.send(DataEvent::Response(response)) {
1055 log::error!("Failed to send funding rates response: {e}");
1056 }
1057 }
1058 Err(e) => log::error!("Funding rates request failed: {e:?}"),
1059 }
1060 });
1061
1062 Ok(())
1063 }
1064}
1065
1066#[cfg(test)]
1067mod tests {
1068 use nautilus_common::{live::runner::set_data_event_sender, messages::DataEvent};
1069 use rstest::rstest;
1070
1071 use super::*;
1072 use crate::{
1073 common::{consts::KRAKEN_CLIENT_ID, enums::KrakenProductType},
1074 config::KrakenDataClientConfig,
1075 };
1076
1077 fn setup_test_env() {
1078 let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
1079 set_data_event_sender(sender);
1080 }
1081
1082 #[rstest]
1083 fn test_futures_data_client_new() {
1084 setup_test_env();
1085 let config = KrakenDataClientConfig {
1086 product_type: KrakenProductType::Futures,
1087 ..Default::default()
1088 };
1089 let client = KrakenFuturesDataClient::new(*KRAKEN_CLIENT_ID, config);
1090 assert!(client.is_ok());
1091
1092 let client = client.unwrap();
1093 assert_eq!(client.client_id(), *KRAKEN_CLIENT_ID);
1094 assert_eq!(client.venue(), Some(*KRAKEN_VENUE));
1095 assert!(!client.is_connected());
1096 assert!(client.is_disconnected());
1097 assert!(client.instruments().is_empty());
1098 }
1099
1100 #[rstest]
1101 fn test_futures_data_client_start_stop() {
1102 setup_test_env();
1103 let config = KrakenDataClientConfig {
1104 product_type: KrakenProductType::Futures,
1105 ..Default::default()
1106 };
1107 let mut client = KrakenFuturesDataClient::new(*KRAKEN_CLIENT_ID, config).unwrap();
1108
1109 assert!(client.start().is_ok());
1110 assert!(client.stop().is_ok());
1111 assert!(client.is_disconnected());
1112 }
1113}