nautilus_deribit/
data.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//! Live market data client implementation for the Deribit adapter.
17
18use std::{
19    sync::{
20        Arc,
21        atomic::{AtomicBool, Ordering},
22    },
23    time::Duration,
24};
25
26use ahash::{AHashMap, AHashSet};
27use anyhow::Context;
28use async_trait::async_trait;
29use futures_util::StreamExt;
30use nautilus_common::{
31    clients::DataClient,
32    live::runner::get_data_event_sender,
33    log_debug, log_info,
34    messages::{
35        DataEvent, DataResponse,
36        data::{
37            BarsResponse, BookResponse, CustomDataResponse, ForwardPricesResponse,
38            InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
39            RequestCustomData, RequestForwardPrices, RequestInstrument, RequestInstruments,
40            RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeBookDepth10,
41            SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
42            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
43            SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, TradesResponse,
44            UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth10, UnsubscribeCustomData,
45            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
46            UnsubscribeInstrumentStatus, UnsubscribeInstruments, UnsubscribeMarkPrices,
47            UnsubscribeOptionGreeks, UnsubscribeQuotes, UnsubscribeTrades,
48        },
49    },
50};
51use nautilus_core::{
52    AtomicMap, AtomicSet, Params,
53    datetime::datetime_to_unix_nanos,
54    time::{AtomicTime, get_atomic_clock_realtime},
55};
56use nautilus_live::{
57    SocketControl,
58    task::{TaskGroup, TaskGroupGuard},
59};
60use nautilus_model::{
61    data::{CustomData, Data, DataType, ForwardPrice},
62    enums::BookType,
63    identifiers::{ClientId, InstrumentId, Symbol, Venue},
64    instruments::{Instrument, InstrumentAny},
65};
66use tokio_util::sync::CancellationToken;
67
68use crate::{
69    common::{
70        consts::{
71            DERIBIT_BOOK_DEFAULT_DEPTH, DERIBIT_BOOK_DEFAULT_GROUP, DERIBIT_BOOK_VALID_DEPTHS,
72            DERIBIT_VENUE,
73        },
74        parse::{bar_spec_to_resolution, parse_instrument_kind_currency},
75    },
76    config::DeribitDataClientConfig,
77    data_types::{DeribitBookSummary, register_deribit_custom_data},
78    http::{
79        client::DeribitHttpClient,
80        models::{DeribitCurrency, DeribitProductType},
81    },
82    websocket::{
83        auth::DERIBIT_DATA_SESSION_NAME, client::DeribitWebSocketClient,
84        enums::DeribitUpdateInterval, messages::NautilusWsMessage,
85    },
86};
87
88/// Deribit live data client.
89#[derive(Debug)]
90pub struct DeribitDataClient {
91    client_id: ClientId,
92    config: DeribitDataClientConfig,
93    http_client: DeribitHttpClient,
94    ws_client: Option<DeribitWebSocketClient>,
95    is_connected: AtomicBool,
96    cancellation_token: CancellationToken,
97    session_tasks: TaskGroup,
98    command_tasks: TaskGroup,
99    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
100    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
101    mark_price_subs: Arc<AtomicSet<InstrumentId>>,
102    index_price_subs: Arc<AtomicSet<InstrumentId>>,
103    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
104    combo_leg_trade_subs: Arc<AtomicMap<InstrumentId, AHashMap<InstrumentId, usize>>>,
105    clock: &'static AtomicTime,
106}
107
108impl DeribitDataClient {
109    const BOOK_SUMMARY_TYPE_NAME: &'static str = "DeribitBookSummary";
110
111    /// Creates a new [`DeribitDataClient`] instance.
112    ///
113    /// # Errors
114    ///
115    /// Returns an error if the client fails to initialize.
116    pub fn new(client_id: ClientId, config: DeribitDataClientConfig) -> anyhow::Result<Self> {
117        let clock = get_atomic_clock_realtime();
118        let data_sender = get_data_event_sender();
119
120        let http_client = if config.has_api_credentials() {
121            DeribitHttpClient::new_with_env(
122                config.api_key.clone(),
123                config.api_secret.clone(),
124                config.base_url_http.clone(),
125                config.environment,
126                config.http_timeout_secs,
127                config.max_retries,
128                config.retry_delay_initial_ms,
129                config.retry_delay_max_ms,
130                config.proxy_url.clone(),
131            )?
132        } else {
133            DeribitHttpClient::new(
134                config.base_url_http.clone(),
135                config.environment,
136                config.http_timeout_secs,
137                config.max_retries,
138                config.retry_delay_initial_ms,
139                config.retry_delay_max_ms,
140                config.proxy_url.clone(),
141            )?
142        };
143
144        let ws_client = DeribitWebSocketClient::new(
145            Some(config.ws_url()),
146            config.api_key.clone(),
147            config.api_secret.clone(),
148            config.heartbeat_interval_secs,
149            config.auth_timeout_secs,
150            config.environment,
151            config.transport_backend,
152            config.proxy_url.clone(),
153        )?
154        .with_socket_control(SocketControl::new(
155            client_id,
156            Some(*DERIBIT_VENUE),
157            "deribit-data-streams",
158        ));
159
160        let session_tasks = TaskGroup::new();
161        let command_tasks = TaskGroup::new();
162
163        Ok(Self {
164            client_id,
165            config,
166            http_client,
167            ws_client: Some(ws_client),
168            is_connected: AtomicBool::new(false),
169            cancellation_token: session_tasks.cancellation_token(),
170            session_tasks,
171            command_tasks,
172            data_sender,
173            instruments: Arc::new(AtomicMap::new()),
174            mark_price_subs: Arc::new(AtomicSet::new()),
175            index_price_subs: Arc::new(AtomicSet::new()),
176            option_greeks_subs: Arc::new(AtomicSet::new()),
177            combo_leg_trade_subs: Arc::new(AtomicMap::new()),
178            clock,
179        })
180    }
181
182    /// Returns a mutable reference to the WebSocket client.
183    fn ws_client_mut(&mut self) -> anyhow::Result<&mut DeribitWebSocketClient> {
184        self.ws_client
185            .as_mut()
186            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))
187    }
188
189    fn spawn_command<F>(&self, future: F)
190    where
191        F: std::future::Future<Output = ()> + Send + 'static,
192    {
193        if let Err(e) = self.command_tasks.spawn(future) {
194            log::warn!("Skipping Deribit data command after shutdown began: {e}");
195        }
196    }
197
198    async fn finish_tasks(&self) -> anyhow::Result<()> {
199        let (session_result, command_result) = tokio::join!(
200            self.session_tasks
201                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
202            self.command_tasks
203                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
204        );
205        session_result.context("failed to finish Deribit data session tasks")?;
206        command_result.context("failed to finish Deribit data command tasks")?;
207        Ok(())
208    }
209
210    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
211        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
212            self.session_tasks.begin_shutdown();
213            self.command_tasks.begin_shutdown();
214            self.finish_tasks().await?;
215            self.session_tasks
216                .start_generation()
217                .context("failed to start Deribit data session task generation")?;
218            self.command_tasks
219                .start_generation()
220                .context("failed to start Deribit data command task generation")?;
221            self.cancellation_token = self.session_tasks.cancellation_token();
222        }
223        Ok(())
224    }
225
226    async fn teardown_partial_connect(&self) -> anyhow::Result<()> {
227        self.session_tasks.begin_shutdown();
228        self.command_tasks.begin_shutdown();
229        if let Some(ws) = self.ws_client.as_ref() {
230            ws.begin_shutdown();
231        }
232
233        let mut errors = Vec::new();
234
235        if let Some(ws) = self.ws_client.as_ref()
236            && let Err(e) = ws.close().await
237        {
238            errors.push(format!("WebSocket shutdown failed: {e}"));
239        }
240
241        if let Err(e) = self.finish_tasks().await {
242            errors.push(e.to_string());
243        }
244        self.is_connected.store(false, Ordering::Release);
245
246        if errors.is_empty() {
247            Ok(())
248        } else {
249            anyhow::bail!(errors.join("; "))
250        }
251    }
252
253    /// Gets the interval from params, defaulting to Raw if authenticated.
254    ///
255    /// If authenticated, we prefer Raw interval for best data quality.
256    /// Users can still override via params if they want 100ms or agg2.
257    fn get_interval(&self, params: &Option<Params>) -> Option<DeribitUpdateInterval> {
258        if let Some(interval) = params
259            .as_ref()
260            .and_then(|p| p.get_str("interval"))
261            .and_then(|s| s.parse::<DeribitUpdateInterval>().ok())
262        {
263            return Some(interval);
264        }
265
266        // Default to Raw if authenticated, otherwise None (100ms default)
267        if let Some(ws) = self.ws_client.as_ref()
268            && ws.is_authenticated()
269        {
270            return Some(DeribitUpdateInterval::Raw);
271        }
272        None
273    }
274
275    /// Spawns a task to process WebSocket messages.
276    fn spawn_stream_task(
277        &self,
278        stream: impl futures_util::Stream<Item = NautilusWsMessage> + Send + 'static,
279    ) -> anyhow::Result<()> {
280        let data_sender = self.data_sender.clone();
281        let instruments = Arc::clone(&self.instruments);
282        let cancellation = self.cancellation_token.clone();
283
284        let future = async move {
285            tokio::pin!(stream);
286
287            loop {
288                tokio::select! {
289                    maybe_msg = stream.next() => {
290                        match maybe_msg {
291                            Some(msg) => Self::handle_ws_message(msg, &data_sender, &instruments),
292                            None => {
293                                log::debug!("WebSocket stream ended");
294                                break;
295                            }
296                        }
297                    }
298                    () = cancellation.cancelled() => {
299                        log::debug!("WebSocket stream task cancelled");
300                        break;
301                    }
302                }
303            }
304        };
305
306        self.session_tasks
307            .spawn(future)
308            .context("failed to register Deribit WebSocket stream task")?;
309        Ok(())
310    }
311
312    /// Handles incoming WebSocket messages.
313    fn handle_ws_message(
314        message: NautilusWsMessage,
315        sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
316        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
317    ) {
318        match message {
319            NautilusWsMessage::Data(payloads) => {
320                for data in payloads {
321                    Self::send_data(sender, data);
322                }
323            }
324            NautilusWsMessage::Deltas(deltas) => {
325                Self::send_data(sender, Data::Deltas(Box::new(deltas)));
326            }
327            NautilusWsMessage::Instrument(instrument) => {
328                let instrument_any = *instrument;
329                instruments.insert(instrument_any.id(), instrument_any.clone());
330
331                if let Err(e) = sender.send(DataEvent::Instrument(instrument_any)) {
332                    log::warn!("Failed to send instrument update: {e}");
333                }
334            }
335            NautilusWsMessage::OptionGreeks(greeks) => {
336                if let Err(e) = sender.send(DataEvent::OptionGreeks(greeks)) {
337                    log::error!("Failed to send option greeks: {e}");
338                }
339            }
340            NautilusWsMessage::Error(e) => {
341                log::warn!("WebSocket error: {e:?}");
342            }
343            NautilusWsMessage::Raw(value) => {
344                log::debug!("Unhandled raw message: {value}");
345            }
346            NautilusWsMessage::Reconnected => {
347                log::info!("WebSocket reconnected");
348            }
349            NautilusWsMessage::Authenticated(auth) => {
350                log::debug!("WebSocket authenticated: expires_in={}s", auth.expires_in);
351            }
352            NautilusWsMessage::FundingRates(funding_rates) => {
353                for funding_rate in funding_rates {
354                    if let Err(e) = sender.send(DataEvent::FundingRate(funding_rate)) {
355                        log::error!("Failed to send funding rate: {e}");
356                    }
357                }
358            }
359            NautilusWsMessage::InstrumentStatus(status) => {
360                if let Err(e) = sender.send(DataEvent::InstrumentStatus(status)) {
361                    log::error!("Failed to send instrument status event: {e}");
362                }
363            }
364            NautilusWsMessage::OrderStatusReports(reports) => {
365                log::warn!(
366                    "Data client received OrderStatusReports message (should be handled by execution client): {} reports",
367                    reports.len()
368                );
369            }
370            NautilusWsMessage::FillReports(reports) => {
371                log::warn!(
372                    "Data client received FillReports message (should be handled by execution client): {} reports",
373                    reports.len()
374                );
375            }
376            NautilusWsMessage::OrderFilled(order) => {
377                log::warn!(
378                    "Data client received OrderFilled message (should be handled by execution client): {order:?}"
379                );
380            }
381            NautilusWsMessage::OrderRejected(order) => {
382                log::warn!(
383                    "Data client received OrderRejected message (should be handled by execution client): {order:?}"
384                );
385            }
386            NautilusWsMessage::OrderAccepted(order) => {
387                log::warn!(
388                    "Data client received OrderAccepted message (should be handled by execution client): {order:?}"
389                );
390            }
391            NautilusWsMessage::OrderCanceled(order) => {
392                log::warn!(
393                    "Data client received OrderCanceled message (should be handled by execution client): {order:?}"
394                );
395            }
396            NautilusWsMessage::OrderExpired(order) => {
397                log::warn!(
398                    "Data client received OrderExpired message (should be handled by execution client): {order:?}"
399                );
400            }
401            NautilusWsMessage::OrderUpdated(order) => {
402                log::warn!(
403                    "Data client received OrderUpdated message (should be handled by execution client): {order:?}"
404                );
405            }
406            NautilusWsMessage::OrderCancelRejected(order) => {
407                log::warn!(
408                    "Data client received OrderCancelRejected message (should be handled by execution client): {order:?}"
409                );
410            }
411            NautilusWsMessage::OrderModifyRejected(order) => {
412                log::warn!(
413                    "Data client received OrderModifyRejected message (should be handled by execution client): {order:?}"
414                );
415            }
416            NautilusWsMessage::AccountState(state) => {
417                log::warn!(
418                    "Data client received AccountState message (should be handled by execution client): {state:?}"
419                );
420            }
421            NautilusWsMessage::AuthenticationFailed(reason) => {
422                log::error!("Authentication failed in data client: {reason}");
423            }
424        }
425    }
426
427    /// Sends data to the data channel.
428    fn send_data(sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>, data: Data) {
429        if let Err(e) = sender.send(DataEvent::Data(data)) {
430            log::error!("Failed to send data: {e}");
431        }
432    }
433
434    // Returns whether a subscribe should lazy-load the instrument before sending,
435    // erroring up front when the instrument is missing and the flag is disabled
436    // (so the WebSocket handler does not silently drop later frames).
437    fn prepare_subscribe(&self, instrument_id: InstrumentId) -> anyhow::Result<bool> {
438        if self.instruments.contains_key(&instrument_id) {
439            return Ok(false);
440        }
441
442        if !self.config.auto_load_missing_instruments {
443            anyhow::bail!(
444                "Instrument {instrument_id} not found and `auto_load_missing_instruments` is disabled"
445            );
446        }
447        Ok(true)
448    }
449
450    // Fetches an instrument over HTTP and seeds the local, HTTP, and WebSocket caches.
451    async fn lazy_load_instrument(
452        http_client: &DeribitHttpClient,
453        ws: &DeribitWebSocketClient,
454        instruments: &AtomicMap<InstrumentId, InstrumentAny>,
455        instrument_id: InstrumentId,
456    ) -> anyhow::Result<()> {
457        let instrument = http_client
458            .request_instrument(instrument_id)
459            .await
460            .with_context(|| format!("failed to lazy-load instrument {instrument_id}"))?;
461        instruments.insert(instrument.id(), instrument.clone());
462        http_client.cache_instruments(std::slice::from_ref(&instrument));
463        ws.cache_instruments(std::slice::from_ref(&instrument));
464        Ok(())
465    }
466
467    fn subscribe_combo_legs(params: &Option<Params>) -> bool {
468        params
469            .as_ref()
470            .and_then(|params| params.get_bool("subscribe_combo_legs"))
471            .unwrap_or(false)
472    }
473
474    fn book_summary_metadata_currency(data_type: &DataType) -> anyhow::Result<String> {
475        data_type
476            .metadata()
477            .and_then(|m| m.get("currency"))
478            .and_then(|v| v.as_str())
479            .map(str::trim)
480            .filter(|value| !value.is_empty())
481            .map(str::to_ascii_uppercase)
482            .ok_or_else(|| {
483                anyhow::anyhow!("DeribitBookSummary requests require metadata['currency']")
484            })
485    }
486
487    fn book_summary_metadata_kind(data_type: &DataType) -> Option<String> {
488        data_type
489            .metadata()
490            .and_then(|m| m.get("kind"))
491            .and_then(|v| v.as_str())
492            .map(str::trim)
493            .filter(|value| !value.is_empty())
494            .map(str::to_ascii_lowercase)
495    }
496
497    fn book_summary_data_type(currency: &str, kind: Option<&str>) -> DataType {
498        let mut metadata = Params::new();
499        metadata.insert(
500            "currency".to_string(),
501            serde_json::Value::String(currency.to_string()),
502        );
503        let kind = kind.unwrap_or("option");
504        metadata.insert(
505            "kind".to_string(),
506            serde_json::Value::String(kind.to_string()),
507        );
508        DataType::new(
509            Self::BOOK_SUMMARY_TYPE_NAME,
510            Some(metadata),
511            Some(format!("{currency}:{kind}")),
512        )
513    }
514
515    fn combo_leg_trade_ids(
516        instruments: &AtomicMap<InstrumentId, InstrumentAny>,
517        instrument_id: InstrumentId,
518    ) -> Vec<InstrumentId> {
519        let Some(instrument) = instruments.get_cloned(&instrument_id) else {
520            log::warn!("Cannot expand Deribit combo legs for missing instrument {instrument_id}");
521            return Vec::new();
522        };
523
524        let info = match instrument {
525            InstrumentAny::CryptoOptionSpread(spread) => spread.info,
526            InstrumentAny::CryptoFuturesSpread(spread) => spread.info,
527            _ => return Vec::new(),
528        };
529        let Some(info) = info else {
530            return Vec::new();
531        };
532        let Some(legs) = info
533            .get("deribit_combo_legs")
534            .and_then(serde_json::Value::as_array)
535        else {
536            return Vec::new();
537        };
538
539        let mut leg_ids = Vec::new();
540        let mut seen = AHashSet::new();
541
542        for leg in legs {
543            let Some(leg_id_str) = leg.get("instrument_id").and_then(serde_json::Value::as_str)
544            else {
545                continue;
546            };
547
548            match InstrumentId::from_as_ref(leg_id_str) {
549                Ok(leg_id) if leg_id != instrument_id && seen.insert(leg_id) => {
550                    leg_ids.push(leg_id);
551                }
552                Ok(_) => {}
553                Err(e) => {
554                    log::warn!(
555                        "Skipping invalid Deribit combo leg instrument ID {leg_id_str}: {e}"
556                    );
557                }
558            }
559        }
560
561        leg_ids
562    }
563
564    fn track_combo_leg_trade_subs(
565        subscriptions: &AtomicMap<InstrumentId, AHashMap<InstrumentId, usize>>,
566        instrument_id: InstrumentId,
567        leg_ids: &[InstrumentId],
568    ) {
569        if leg_ids.is_empty() {
570            return;
571        }
572
573        subscriptions.rcu(|subscriptions| {
574            let counts = subscriptions.entry(instrument_id).or_default();
575
576            for leg_id in leg_ids {
577                counts
578                    .entry(*leg_id)
579                    .and_modify(|count| *count += 1)
580                    .or_insert(1);
581            }
582        });
583    }
584
585    fn combo_leg_trade_unsubs(
586        subscriptions: &AtomicMap<InstrumentId, AHashMap<InstrumentId, usize>>,
587        instrument_id: InstrumentId,
588    ) -> Vec<InstrumentId> {
589        let mut leg_ids = Vec::new();
590
591        subscriptions.rcu(|subscriptions| {
592            let remove_instrument = if let Some(counts) = subscriptions.get_mut(&instrument_id) {
593                leg_ids = counts.keys().copied().collect();
594                counts.retain(|_, count| {
595                    if *count > 1 {
596                        *count -= 1;
597                        true
598                    } else {
599                        false
600                    }
601                });
602                counts.is_empty()
603            } else {
604                leg_ids = Vec::new();
605                false
606            };
607
608            if remove_instrument {
609                subscriptions.remove(&instrument_id);
610            }
611        });
612
613        leg_ids
614    }
615}
616
617#[async_trait(?Send)]
618impl DataClient for DeribitDataClient {
619    fn client_id(&self) -> ClientId {
620        self.client_id
621    }
622
623    fn venue(&self) -> Option<Venue> {
624        Some(*DERIBIT_VENUE)
625    }
626
627    fn start(&mut self) -> anyhow::Result<()> {
628        log::info!(
629            "Starting data client: client_id={}, environment={}",
630            self.client_id,
631            self.config.environment
632        );
633        Ok(())
634    }
635
636    fn stop(&mut self) -> anyhow::Result<()> {
637        log::info!("Stopping data client: {}", self.client_id);
638        self.session_tasks.begin_shutdown();
639        self.command_tasks.begin_shutdown();
640        if let Some(ws) = self.ws_client.as_ref() {
641            ws.begin_shutdown();
642        }
643        self.is_connected.store(false, Ordering::Relaxed);
644        Ok(())
645    }
646
647    fn reset(&mut self) -> anyhow::Result<()> {
648        log::info!("Resetting data client: {}", self.client_id);
649        self.session_tasks.begin_shutdown();
650        self.command_tasks.begin_shutdown();
651        if let Some(ws) = self.ws_client.as_ref() {
652            ws.begin_shutdown();
653        }
654        self.is_connected.store(false, Ordering::Relaxed);
655
656        self.instruments.store(AHashMap::new());
657        self.combo_leg_trade_subs.store(AHashMap::new());
658        Ok(())
659    }
660
661    fn dispose(&mut self) -> anyhow::Result<()> {
662        log::debug!("Disposing data client: {}", self.client_id);
663        self.stop()
664    }
665
666    fn is_connected(&self) -> bool {
667        self.is_connected.load(Ordering::SeqCst)
668    }
669
670    fn is_disconnected(&self) -> bool {
671        !self.is_connected()
672    }
673
674    async fn connect(&mut self) -> anyhow::Result<()> {
675        if self.is_connected() && self.session_tasks.is_open() && self.command_tasks.is_open() {
676            return Ok(());
677        }
678
679        self.prepare_task_groups().await?;
680        let cancellation_token = self.cancellation_token.clone();
681        let ws_client = self.ws_client.clone();
682        let setup_guard =
683            TaskGroupGuard::new(&[&self.session_tasks, &self.command_tasks], move || {
684                cancellation_token.cancel();
685
686                if let Some(ws) = ws_client {
687                    ws.begin_shutdown();
688                }
689            });
690
691        register_deribit_custom_data();
692
693        // Fetch instruments for each configured product type
694        let product_types = if self.config.product_types.is_empty() {
695            vec![DeribitProductType::Future]
696        } else {
697            self.config.product_types.clone()
698        };
699
700        let mut all_instruments = Vec::new();
701
702        for product_type in &product_types {
703            let fetched = self
704                .http_client
705                .request_instruments(DeribitCurrency::ANY, Some(*product_type))
706                .await
707                .with_context(|| format!("failed to request instruments for {product_type:?}"))?;
708
709            // Cache in http client
710            self.http_client.cache_instruments(&fetched);
711
712            // Cache locally
713            self.instruments.rcu(|m| {
714                for instrument in &fetched {
715                    m.insert(instrument.id(), instrument.clone());
716                }
717            });
718
719            all_instruments.extend(fetched);
720        }
721
722        log::debug!(
723            "Cached instruments: client_id={}, total={}",
724            self.client_id,
725            all_instruments.len()
726        );
727
728        for instrument in &all_instruments {
729            if let Err(e) = self
730                .data_sender
731                .send(DataEvent::Instrument(instrument.clone()))
732            {
733                log::warn!("Failed to send instrument: {e}");
734            }
735        }
736
737        // Cache instruments and set subscription filters in WebSocket client before connecting
738        let mark_price_subs = self.mark_price_subs.clone();
739        let index_price_subs = self.index_price_subs.clone();
740        let option_greeks_subs = self.option_greeks_subs.clone();
741        let ws = self.ws_client_mut()?;
742        ws.cache_instruments(&all_instruments);
743        ws.set_mark_price_subs(mark_price_subs);
744        ws.set_index_price_subs(index_price_subs);
745        ws.set_option_greeks_subs(option_greeks_subs);
746
747        // Connect WebSocket and wait until active
748        ws.connect().await.context("failed to connect WebSocket")?;
749        let activation_result = async {
750            ws.wait_until_active(10.0)
751                .await
752                .context("WebSocket failed to become active")?;
753
754            // Authenticate if credentials are configured (required for raw streams)
755            if ws.has_credentials() {
756                ws.authenticate_session(DERIBIT_DATA_SESSION_NAME)
757                    .await
758                    .context("failed to authenticate WebSocket")?;
759                log_debug!("WebSocket authenticated");
760            }
761            Ok::<(), anyhow::Error>(())
762        }
763        .await;
764
765        if let Err(e) = activation_result {
766            if let Err(teardown_error) = self.teardown_partial_connect().await {
767                return Err(e.context(format!(
768                    "Deribit data startup teardown failed: {teardown_error}"
769                )));
770            }
771            return Err(e);
772        }
773
774        // Get the stream and spawn processing task
775        let stream_result = self.ws_client_mut().and_then(|ws| Ok(ws.stream()?));
776        let stream = match stream_result {
777            Ok(stream) => stream,
778            Err(e) => {
779                if let Err(teardown_error) = self.teardown_partial_connect().await {
780                    return Err(e.context(format!(
781                        "Deribit data startup teardown failed: {teardown_error}"
782                    )));
783                }
784                return Err(e);
785            }
786        };
787
788        if let Err(e) = self.spawn_stream_task(stream) {
789            if let Err(teardown_error) = self.teardown_partial_connect().await {
790                return Err(e.context(format!(
791                    "Deribit data startup teardown failed: {teardown_error}"
792                )));
793            }
794            return Err(e);
795        }
796
797        self.is_connected.store(true, Ordering::Release);
798        setup_guard.disarm();
799        log_info!("Connected ({})", self.config.environment);
800        Ok(())
801    }
802
803    async fn disconnect(&mut self) -> anyhow::Result<()> {
804        self.teardown_partial_connect().await?;
805
806        log_info!("Disconnected");
807        Ok(())
808    }
809
810    fn subscribe_instruments(&mut self, cmd: SubscribeInstruments) -> anyhow::Result<()> {
811        // Extract kind and currency from params, defaulting to "any.any" (all instruments)
812        let kind = cmd
813            .params
814            .as_ref()
815            .and_then(|p| p.get_str("kind"))
816            .unwrap_or("any")
817            .to_string();
818        let currency = cmd
819            .params
820            .as_ref()
821            .and_then(|p| p.get_str("currency"))
822            .unwrap_or("any")
823            .to_string();
824
825        let ws = self
826            .ws_client
827            .as_ref()
828            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
829            .clone();
830
831        log::debug!("Subscribing to instrument state changes for {kind}.{currency}");
832
833        self.spawn_command(async move {
834            if let Err(e) = ws.subscribe_instrument_status(&kind, &currency).await {
835                log::error!("Failed to subscribe to instrument status for {kind}.{currency}: {e}");
836            }
837        });
838
839        Ok(())
840    }
841
842    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
843        let instrument_id = cmd.instrument_id;
844
845        // Check if instrument is in cache (should be from connect())
846        if !self.instruments.contains_key(&instrument_id) {
847            log::warn!(
848                "Instrument {instrument_id} not in cache - it may have been created after connect()"
849            );
850        }
851
852        // Determine kind and currency from instrument_id
853        let (kind, currency) = parse_instrument_kind_currency(&instrument_id);
854
855        let ws = self
856            .ws_client
857            .as_ref()
858            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
859            .clone();
860
861        log::debug!(
862            "Subscribing to instrument state for {instrument_id} (channel: {kind}.{currency})"
863        );
864
865        // Subscribe to broader kind/currency channel (filter in handler)
866        self.spawn_command(async move {
867            if let Err(e) = ws.subscribe_instrument_status(&kind, &currency).await {
868                log::error!("Failed to subscribe to instrument status for {instrument_id}: {e}");
869            }
870        });
871
872        Ok(())
873    }
874
875    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
876        if cmd.book_type != BookType::L2_MBP {
877            anyhow::bail!("Deribit only supports L2_MBP order book deltas");
878        }
879
880        let instrument_id = cmd.instrument_id;
881        let needs_load = self.prepare_subscribe(instrument_id)?;
882
883        let ws = self
884            .ws_client
885            .as_ref()
886            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
887            .clone();
888        let http_client = self.http_client.clone();
889        let instruments = Arc::clone(&self.instruments);
890        let interval = self.get_interval(&cmd.params);
891
892        let depth = cmd
893            .depth
894            .map(|d| d.get() as u32)
895            .or_else(|| {
896                cmd.params
897                    .as_ref()
898                    .and_then(|p| p.get_u64("depth"))
899                    .map(|n| n as u32)
900            })
901            .unwrap_or(DERIBIT_BOOK_DEFAULT_DEPTH);
902
903        if !DERIBIT_BOOK_VALID_DEPTHS.contains(&depth) {
904            anyhow::bail!("invalid depth {depth}; supported depths: {DERIBIT_BOOK_VALID_DEPTHS:?}");
905        }
906
907        let group = cmd
908            .params
909            .as_ref()
910            .and_then(|p| p.get_str("group"))
911            .unwrap_or(DERIBIT_BOOK_DEFAULT_GROUP)
912            .to_string();
913
914        log::debug!(
915            "Subscribing to book deltas for {} (group: {}, depth: {}, interval: {}, book_type: {:?})",
916            instrument_id,
917            group,
918            depth,
919            interval.map_or("100ms (default)".to_string(), |i| i.to_string()),
920            cmd.book_type
921        );
922
923        self.spawn_command(async move {
924            if needs_load
925                && let Err(e) =
926                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
927            {
928                log::error!("Lazy-load failed for {instrument_id} (book deltas): {e}");
929                return;
930            }
931
932            let result = if interval == Some(DeribitUpdateInterval::Raw) {
933                ws.subscribe_book(instrument_id, interval).await
934            } else {
935                ws.subscribe_book_grouped(instrument_id, &group, depth, interval)
936                    .await
937            };
938
939            if let Err(e) = result {
940                log::error!("Failed to subscribe to book deltas for {instrument_id}: {e}");
941            }
942        });
943
944        Ok(())
945    }
946
947    fn subscribe_book_depth10(&mut self, cmd: SubscribeBookDepth10) -> anyhow::Result<()> {
948        if cmd.book_type != BookType::L2_MBP {
949            anyhow::bail!("Deribit only supports L2_MBP order book depth");
950        }
951
952        let instrument_id = cmd.instrument_id;
953        let needs_load = self.prepare_subscribe(instrument_id)?;
954
955        let ws = self
956            .ws_client
957            .as_ref()
958            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
959            .clone();
960        let http_client = self.http_client.clone();
961        let instruments = Arc::clone(&self.instruments);
962        let interval = self.get_interval(&cmd.params);
963        let group = cmd
964            .params
965            .as_ref()
966            .and_then(|p| p.get_str("group"))
967            .unwrap_or(DERIBIT_BOOK_DEFAULT_GROUP)
968            .to_string();
969
970        log::debug!(
971            "Subscribing to book depth10 for {} (group: {}, interval: {}, book_type: {:?})",
972            instrument_id,
973            group,
974            interval.map_or("100ms (default)".to_string(), |i| i.to_string()),
975            cmd.book_type
976        );
977
978        self.spawn_command(async move {
979            if needs_load
980                && let Err(e) =
981                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
982            {
983                log::error!("Lazy-load failed for {instrument_id} (book depth10): {e}");
984                return;
985            }
986
987            if let Err(e) = ws
988                .subscribe_book_grouped(instrument_id, &group, 10, interval)
989                .await
990            {
991                log::error!("Failed to subscribe to book depth10 for {instrument_id}: {e}");
992            }
993        });
994
995        Ok(())
996    }
997
998    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
999        let instrument_id = cmd.instrument_id;
1000        let command_id = cmd.command_id;
1001        let needs_load = self.prepare_subscribe(instrument_id)?;
1002
1003        let ws = self
1004            .ws_client
1005            .as_ref()
1006            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1007            .clone();
1008        let http_client = self.http_client.clone();
1009        let instruments = Arc::clone(&self.instruments);
1010
1011        self.spawn_command(async move {
1012            if needs_load
1013                && let Err(e) =
1014                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1015            {
1016                log::error!(
1017                    "Lazy-load failed for {instrument_id} (quotes, command_id={command_id}): {e}"
1018                );
1019                return;
1020            }
1021
1022            if let Err(e) = ws.subscribe_quotes(instrument_id).await {
1023                log::error!("Failed to subscribe to quotes for {instrument_id}: {e}");
1024            }
1025        });
1026
1027        Ok(())
1028    }
1029
1030    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1031        let instrument_id = cmd.instrument_id;
1032        let command_id = cmd.command_id;
1033        let needs_load = self.prepare_subscribe(instrument_id)?;
1034        let subscribe_combo_legs = Self::subscribe_combo_legs(&cmd.params);
1035
1036        let ws = self
1037            .ws_client
1038            .as_ref()
1039            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1040            .clone();
1041        let http_client = self.http_client.clone();
1042        let instruments = Arc::clone(&self.instruments);
1043        let combo_leg_trade_subs = Arc::clone(&self.combo_leg_trade_subs);
1044        let auto_load_missing_instruments = self.config.auto_load_missing_instruments;
1045        let interval = self.get_interval(&cmd.params);
1046
1047        log::debug!(
1048            "Subscribing to trades for {} (interval: {})",
1049            instrument_id,
1050            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1051        );
1052
1053        self.spawn_command(async move {
1054            if needs_load
1055                && let Err(e) =
1056                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1057            {
1058                log::error!("Lazy-load failed for {instrument_id} (trades): {e}");
1059                return;
1060            }
1061
1062            let mut subscription_ids = vec![instrument_id];
1063
1064            if subscribe_combo_legs {
1065                let leg_ids = Self::combo_leg_trade_ids(&instruments, instrument_id);
1066                if leg_ids.is_empty() {
1067                    log::warn!(
1068                        "No Deribit combo legs found for trade subscription opt-in on {instrument_id}"
1069                    );
1070                }
1071
1072                for leg_id in leg_ids {
1073                    if !instruments.contains_key(&leg_id) {
1074                        if !auto_load_missing_instruments {
1075                            log::error!(
1076                                "Instrument {leg_id} not found and `auto_load_missing_instruments` is disabled"
1077                            );
1078                            continue;
1079                        }
1080
1081                        if let Err(e) =
1082                            Self::lazy_load_instrument(&http_client, &ws, &instruments, leg_id)
1083                                .await
1084                        {
1085                            log::error!("Lazy-load failed for {leg_id} (combo leg trades): {e}");
1086                            continue;
1087                        }
1088                    }
1089
1090                    subscription_ids.push(leg_id);
1091                }
1092            }
1093
1094            let subscription_count = subscription_ids.len();
1095            let mut opened_leg_ids = Vec::new();
1096
1097            for subscription_id in subscription_ids {
1098                if let Err(e) = ws.subscribe_trades(subscription_id, interval).await {
1099                    log::error!("Failed to subscribe to trades for {subscription_id}: {e}");
1100                    continue;
1101                }
1102
1103                if subscription_id != instrument_id {
1104                    opened_leg_ids.push(subscription_id);
1105                }
1106            }
1107
1108            Self::track_combo_leg_trade_subs(
1109                &combo_leg_trade_subs,
1110                instrument_id,
1111                &opened_leg_ids,
1112            );
1113
1114            log::debug!(
1115                "Processed trade subscription batch: command_id={command_id}, requests={subscription_count}, instrument={instrument_id}"
1116            );
1117        });
1118
1119        Ok(())
1120    }
1121
1122    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1123        let instrument_id = cmd.instrument_id;
1124        let needs_load = self.prepare_subscribe(instrument_id)?;
1125
1126        let ws = self
1127            .ws_client
1128            .as_ref()
1129            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1130            .clone();
1131        let http_client = self.http_client.clone();
1132        let instruments = Arc::clone(&self.instruments);
1133        let interval = self.get_interval(&cmd.params);
1134
1135        // Track subscription so handler gates MarkPriceUpdate emission
1136        self.mark_price_subs.insert(instrument_id);
1137
1138        log::debug!(
1139            "Subscribing to mark prices for {} (via ticker channel, interval: {})",
1140            instrument_id,
1141            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1142        );
1143
1144        self.spawn_command(async move {
1145            if needs_load
1146                && let Err(e) =
1147                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1148            {
1149                log::error!("Lazy-load failed for {instrument_id} (mark prices): {e}");
1150                return;
1151            }
1152
1153            if let Err(e) = ws.subscribe_ticker(instrument_id, interval).await {
1154                log::error!("Failed to subscribe to mark prices for {instrument_id}: {e}");
1155            }
1156        });
1157
1158        Ok(())
1159    }
1160
1161    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1162        let instrument_id = cmd.instrument_id;
1163        let needs_load = self.prepare_subscribe(instrument_id)?;
1164
1165        let ws = self
1166            .ws_client
1167            .as_ref()
1168            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1169            .clone();
1170        let http_client = self.http_client.clone();
1171        let instruments = Arc::clone(&self.instruments);
1172        let interval = self.get_interval(&cmd.params);
1173
1174        // Track subscription so handler gates IndexPriceUpdate emission
1175        self.index_price_subs.insert(instrument_id);
1176
1177        log::debug!(
1178            "Subscribing to index prices for {} (via ticker channel, interval: {})",
1179            instrument_id,
1180            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1181        );
1182
1183        self.spawn_command(async move {
1184            if needs_load
1185                && let Err(e) =
1186                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1187            {
1188                log::error!("Lazy-load failed for {instrument_id} (index prices): {e}");
1189                return;
1190            }
1191
1192            if let Err(e) = ws.subscribe_ticker(instrument_id, interval).await {
1193                log::error!("Failed to subscribe to index prices for {instrument_id}: {e}");
1194            }
1195        });
1196
1197        Ok(())
1198    }
1199
1200    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1201        let instrument_id = cmd.bar_type.instrument_id();
1202        let needs_load = self.prepare_subscribe(instrument_id)?;
1203
1204        let ws = self
1205            .ws_client
1206            .as_ref()
1207            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1208            .clone();
1209        let http_client = self.http_client.clone();
1210        let instruments = Arc::clone(&self.instruments);
1211        let resolution = bar_spec_to_resolution(&cmd.bar_type);
1212
1213        self.spawn_command(async move {
1214            if needs_load
1215                && let Err(e) =
1216                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1217            {
1218                log::error!("Lazy-load failed for {instrument_id} (bars): {e}");
1219                return;
1220            }
1221
1222            if let Err(e) = ws.subscribe_chart(instrument_id, &resolution).await {
1223                log::error!("Failed to subscribe to bars for {instrument_id}: {e}");
1224            }
1225        });
1226
1227        Ok(())
1228    }
1229
1230    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
1231        let instrument_id = cmd.instrument_id;
1232        let command_id = cmd.command_id;
1233        let needs_load = self.prepare_subscribe(instrument_id)?;
1234
1235        let ws = self
1236            .ws_client
1237            .as_ref()
1238            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1239            .clone();
1240        let http_client = self.http_client.clone();
1241        let instruments = Arc::clone(&self.instruments);
1242        let interval = self.get_interval(&cmd.params);
1243
1244        log::debug!(
1245            "Subscribing to funding rates for {} (perpetual channel, interval: {})",
1246            instrument_id,
1247            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1248        );
1249
1250        self.spawn_command(async move {
1251            if needs_load
1252                && let Err(e) =
1253                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1254            {
1255                log::error!("Lazy-load failed for {instrument_id} (funding rates): {e}");
1256                return;
1257            }
1258
1259            // Funding rates only apply to perpetual contracts; check after any lazy-load
1260            let is_perpetual = instruments
1261                .load()
1262                .get(&instrument_id)
1263                .is_some_and(|inst| matches!(inst, InstrumentAny::CryptoPerpetual(_)));
1264
1265            if !is_perpetual {
1266                log::warn!(
1267                    "Funding rates subscription rejected for {instrument_id} (command_id={command_id}): only available for perpetual instruments"
1268                );
1269                return;
1270            }
1271
1272            if let Err(e) = ws
1273                .subscribe_perpetual_interests_rates_updates(instrument_id, interval)
1274                .await
1275            {
1276                log::error!("Failed to subscribe to funding rates for {instrument_id}: {e}");
1277            }
1278        });
1279
1280        Ok(())
1281    }
1282
1283    fn subscribe_instrument_status(
1284        &mut self,
1285        cmd: SubscribeInstrumentStatus,
1286    ) -> anyhow::Result<()> {
1287        let instrument_id = cmd.instrument_id;
1288        let (kind, currency) = parse_instrument_kind_currency(&instrument_id);
1289
1290        let ws = self
1291            .ws_client
1292            .as_ref()
1293            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1294            .clone();
1295
1296        log::debug!("Subscribing to instrument status for {instrument_id} ({kind}.{currency})");
1297
1298        self.spawn_command(async move {
1299            if let Err(e) = ws.subscribe_instrument_status(&kind, &currency).await {
1300                log::error!("Failed to subscribe to instrument status for {instrument_id}: {e}");
1301            }
1302        });
1303
1304        Ok(())
1305    }
1306
1307    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
1308        let instrument_id = cmd.instrument_id;
1309        let needs_load = self.prepare_subscribe(instrument_id)?;
1310
1311        let ws = self
1312            .ws_client
1313            .as_ref()
1314            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1315            .clone();
1316        let http_client = self.http_client.clone();
1317        let instruments = Arc::clone(&self.instruments);
1318        let interval = self.get_interval(&cmd.params);
1319
1320        // Track subscription so handler gates OptionGreeks emission
1321        self.option_greeks_subs.insert(instrument_id);
1322
1323        log::debug!(
1324            "Subscribing to option greeks for {} (via ticker channel, interval: {})",
1325            instrument_id,
1326            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1327        );
1328
1329        self.spawn_command(async move {
1330            if needs_load
1331                && let Err(e) =
1332                    Self::lazy_load_instrument(&http_client, &ws, &instruments, instrument_id).await
1333            {
1334                log::error!("Lazy-load failed for {instrument_id} (option greeks): {e}");
1335                return;
1336            }
1337
1338            if let Err(e) = ws.subscribe_ticker(instrument_id, interval).await {
1339                log::error!("Failed to subscribe to option greeks for {instrument_id}: {e}");
1340            }
1341        });
1342
1343        Ok(())
1344    }
1345
1346    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
1347        let data_type = cmd.data_type.type_name();
1348        if data_type != "DeribitVolatilityIndex" {
1349            log::warn!("Unsupported custom data subscription: {data_type}");
1350            return Ok(());
1351        }
1352
1353        let Some(index_name) = cmd
1354            .data_type
1355            .metadata()
1356            .as_ref()
1357            .and_then(|m| m.get("index_name"))
1358            .and_then(|v| v.as_str())
1359            .map(str::trim)
1360            .filter(|value| !value.is_empty())
1361            .map(ToString::to_string)
1362        else {
1363            log::warn!(
1364                "Rejected Deribit volatility index subscription: missing required metadata `index_name`"
1365            );
1366            return Ok(());
1367        };
1368
1369        log::debug!("Subscribing to Deribit volatility index: {index_name}");
1370
1371        let ws = self
1372            .ws_client
1373            .as_ref()
1374            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1375            .clone();
1376
1377        self.spawn_command(async move {
1378            if let Err(e) = ws.subscribe_volatility_index(&index_name).await {
1379                log::error!("Failed to subscribe to volatility index {index_name}: {e}");
1380            }
1381        });
1382
1383        Ok(())
1384    }
1385
1386    fn unsubscribe_instrument_status(
1387        &mut self,
1388        cmd: &UnsubscribeInstrumentStatus,
1389    ) -> anyhow::Result<()> {
1390        let instrument_id = cmd.instrument_id;
1391        let (kind, currency) = parse_instrument_kind_currency(&instrument_id);
1392
1393        let ws = self
1394            .ws_client
1395            .as_ref()
1396            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1397            .clone();
1398
1399        log::debug!("Unsubscribing from instrument status for {instrument_id} ({kind}.{currency})");
1400
1401        self.spawn_command(async move {
1402            if let Err(e) = ws.unsubscribe_instrument_status(&kind, &currency).await {
1403                log::error!(
1404                    "Failed to unsubscribe from instrument status for {instrument_id}: {e}"
1405                );
1406            }
1407        });
1408
1409        Ok(())
1410    }
1411
1412    fn unsubscribe_instruments(&mut self, cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
1413        let kind = cmd
1414            .params
1415            .as_ref()
1416            .and_then(|p| p.get_str("kind"))
1417            .unwrap_or("any")
1418            .to_string();
1419        let currency = cmd
1420            .params
1421            .as_ref()
1422            .and_then(|p| p.get_str("currency"))
1423            .unwrap_or("any")
1424            .to_string();
1425
1426        let ws = self
1427            .ws_client
1428            .as_ref()
1429            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1430            .clone();
1431
1432        log::debug!("Unsubscribing from instrument state changes for {kind}.{currency}");
1433
1434        self.spawn_command(async move {
1435            if let Err(e) = ws.unsubscribe_instrument_status(&kind, &currency).await {
1436                log::error!(
1437                    "Failed to unsubscribe from instrument status for {kind}.{currency}: {e}"
1438                );
1439            }
1440        });
1441
1442        Ok(())
1443    }
1444
1445    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
1446        let instrument_id = cmd.instrument_id;
1447
1448        // Determine kind and currency from instrument_id
1449        let (kind, currency) = parse_instrument_kind_currency(&instrument_id);
1450
1451        let ws = self
1452            .ws_client
1453            .as_ref()
1454            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1455            .clone();
1456
1457        log::debug!(
1458            "Unsubscribing from instrument state for {instrument_id} (channel: {kind}.{currency})"
1459        );
1460
1461        self.spawn_command(async move {
1462            if let Err(e) = ws.unsubscribe_instrument_status(&kind, &currency).await {
1463                log::error!(
1464                    "Failed to unsubscribe from instrument status for {instrument_id}: {e}"
1465                );
1466            }
1467        });
1468
1469        Ok(())
1470    }
1471
1472    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1473        let ws = self
1474            .ws_client
1475            .as_ref()
1476            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1477            .clone();
1478        let instrument_id = cmd.instrument_id;
1479        let interval = self.get_interval(&cmd.params);
1480
1481        let depth = cmd
1482            .params
1483            .as_ref()
1484            .and_then(|p| p.get_u64("depth"))
1485            .map_or(DERIBIT_BOOK_DEFAULT_DEPTH, |n| n as u32);
1486
1487        if !DERIBIT_BOOK_VALID_DEPTHS.contains(&depth) {
1488            anyhow::bail!("invalid depth {depth}; supported depths: {DERIBIT_BOOK_VALID_DEPTHS:?}");
1489        }
1490
1491        let group = cmd
1492            .params
1493            .as_ref()
1494            .and_then(|p| p.get_str("group"))
1495            .unwrap_or(DERIBIT_BOOK_DEFAULT_GROUP)
1496            .to_string();
1497
1498        log::debug!(
1499            "Unsubscribing from book deltas for {} (group: {}, depth: {}, interval: {})",
1500            instrument_id,
1501            group,
1502            depth,
1503            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1504        );
1505
1506        self.spawn_command(async move {
1507            let result = if interval == Some(DeribitUpdateInterval::Raw) {
1508                ws.unsubscribe_book(instrument_id, interval).await
1509            } else {
1510                ws.unsubscribe_book_grouped(instrument_id, &group, depth, interval)
1511                    .await
1512            };
1513
1514            if let Err(e) = result {
1515                log::error!("Failed to unsubscribe from book deltas for {instrument_id}: {e}");
1516            }
1517        });
1518
1519        Ok(())
1520    }
1521
1522    fn unsubscribe_book_depth10(&mut self, cmd: &UnsubscribeBookDepth10) -> anyhow::Result<()> {
1523        let ws = self
1524            .ws_client
1525            .as_ref()
1526            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1527            .clone();
1528        let instrument_id = cmd.instrument_id;
1529        let interval = self.get_interval(&cmd.params);
1530        let group = cmd
1531            .params
1532            .as_ref()
1533            .and_then(|p| p.get_str("group"))
1534            .unwrap_or(DERIBIT_BOOK_DEFAULT_GROUP)
1535            .to_string();
1536
1537        log::debug!(
1538            "Unsubscribing from book depth10 for {} (group: {}, interval: {})",
1539            instrument_id,
1540            group,
1541            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1542        );
1543
1544        self.spawn_command(async move {
1545            if let Err(e) = ws
1546                .unsubscribe_book_grouped(instrument_id, &group, 10, interval)
1547                .await
1548            {
1549                log::error!("Failed to unsubscribe from book depth10 for {instrument_id}: {e}");
1550            }
1551        });
1552
1553        Ok(())
1554    }
1555
1556    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1557        let ws = self
1558            .ws_client
1559            .as_ref()
1560            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1561            .clone();
1562        let instrument_id = cmd.instrument_id;
1563
1564        self.spawn_command(async move {
1565            if let Err(e) = ws.unsubscribe_quotes(instrument_id).await {
1566                log::error!("Failed to unsubscribe from quotes for {instrument_id}: {e}");
1567            }
1568        });
1569
1570        Ok(())
1571    }
1572
1573    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1574        let ws = self
1575            .ws_client
1576            .as_ref()
1577            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1578            .clone();
1579        let instrument_id = cmd.instrument_id;
1580        let command_id = cmd.command_id;
1581        let interval = self.get_interval(&cmd.params);
1582        let mut subscription_ids = vec![instrument_id];
1583        subscription_ids.extend(Self::combo_leg_trade_unsubs(
1584            &self.combo_leg_trade_subs,
1585            instrument_id,
1586        ));
1587        let subscription_count = subscription_ids.len();
1588
1589        log::debug!(
1590            "Unsubscribing from trades for {} instruments from {} (interval: {})",
1591            subscription_ids.len(),
1592            instrument_id,
1593            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1594        );
1595
1596        self.spawn_command(async move {
1597            for subscription_id in subscription_ids {
1598                if let Err(e) = ws.unsubscribe_trades(subscription_id, interval).await {
1599                    log::error!("Failed to unsubscribe from trades for {subscription_id}: {e}");
1600                }
1601            }
1602
1603            log::debug!(
1604                "Processed trade unsubscription batch: command_id={command_id}, requests={subscription_count}, instrument={instrument_id}"
1605            );
1606        });
1607
1608        Ok(())
1609    }
1610
1611    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1612        let ws = self
1613            .ws_client
1614            .as_ref()
1615            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1616            .clone();
1617        let instrument_id = cmd.instrument_id;
1618        let interval = self.get_interval(&cmd.params);
1619
1620        // Remove subscription tracking so handler stops emitting MarkPriceUpdate
1621        self.mark_price_subs.remove(&instrument_id);
1622
1623        log::debug!(
1624            "Unsubscribing from mark prices for {} (via ticker channel, interval: {})",
1625            instrument_id,
1626            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1627        );
1628
1629        self.spawn_command(async move {
1630            if let Err(e) = ws.unsubscribe_ticker(instrument_id, interval).await {
1631                log::error!("Failed to unsubscribe from mark prices for {instrument_id}: {e}");
1632            }
1633        });
1634
1635        Ok(())
1636    }
1637
1638    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1639        let ws = self
1640            .ws_client
1641            .as_ref()
1642            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1643            .clone();
1644        let instrument_id = cmd.instrument_id;
1645        let interval = self.get_interval(&cmd.params);
1646
1647        // Remove subscription tracking so handler stops emitting IndexPriceUpdate
1648        self.index_price_subs.remove(&instrument_id);
1649
1650        log::debug!(
1651            "Unsubscribing from index prices for {} (via ticker channel, interval: {})",
1652            instrument_id,
1653            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1654        );
1655
1656        self.spawn_command(async move {
1657            if let Err(e) = ws.unsubscribe_ticker(instrument_id, interval).await {
1658                log::error!("Failed to unsubscribe from index prices for {instrument_id}: {e}");
1659            }
1660        });
1661
1662        Ok(())
1663    }
1664
1665    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1666        let ws = self
1667            .ws_client
1668            .as_ref()
1669            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1670            .clone();
1671        let instrument_id = cmd.bar_type.instrument_id();
1672        let resolution = bar_spec_to_resolution(&cmd.bar_type);
1673
1674        self.spawn_command(async move {
1675            if let Err(e) = ws.unsubscribe_chart(instrument_id, &resolution).await {
1676                log::error!("Failed to unsubscribe from bars for {instrument_id}: {e}");
1677            }
1678        });
1679
1680        Ok(())
1681    }
1682
1683    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1684        let instrument_id = cmd.instrument_id;
1685
1686        // Validate instrument is a perpetual - funding rates only apply to perpetual contracts
1687        let is_perpetual = self
1688            .instruments
1689            .load()
1690            .get(&instrument_id)
1691            .is_some_and(|inst| matches!(inst, InstrumentAny::CryptoPerpetual(_)));
1692
1693        if !is_perpetual {
1694            log::warn!(
1695                "Funding rates unsubscription rejected for {instrument_id}: only available for perpetual instruments"
1696            );
1697            return Ok(());
1698        }
1699
1700        let ws = self
1701            .ws_client
1702            .as_ref()
1703            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1704            .clone();
1705        let interval = self.get_interval(&cmd.params);
1706
1707        log::debug!(
1708            "Unsubscribing from funding rates for {} (perpetual channel, interval: {})",
1709            instrument_id,
1710            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1711        );
1712
1713        self.spawn_command(async move {
1714            if let Err(e) = ws
1715                .unsubscribe_perpetual_interest_rates_updates(instrument_id, interval)
1716                .await
1717            {
1718                log::error!("Failed to unsubscribe from funding rates for {instrument_id}: {e}");
1719            }
1720        });
1721
1722        Ok(())
1723    }
1724
1725    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1726        let ws = self
1727            .ws_client
1728            .as_ref()
1729            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1730            .clone();
1731        let instrument_id = cmd.instrument_id;
1732        let interval = self.get_interval(&cmd.params);
1733
1734        // Remove subscription tracking so handler stops emitting OptionGreeks
1735        self.option_greeks_subs.remove(&instrument_id);
1736
1737        log::debug!(
1738            "Unsubscribing from option greeks for {} (via ticker channel, interval: {})",
1739            instrument_id,
1740            interval.map_or("100ms (default)".to_string(), |i| i.to_string())
1741        );
1742
1743        self.spawn_command(async move {
1744            if let Err(e) = ws.unsubscribe_ticker(instrument_id, interval).await {
1745                log::error!("Failed to unsubscribe from option greeks for {instrument_id}: {e}");
1746            }
1747        });
1748
1749        Ok(())
1750    }
1751
1752    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
1753        let data_type = cmd.data_type.type_name();
1754        if data_type != "DeribitVolatilityIndex" {
1755            log::warn!("Unsupported custom data unsubscription: {data_type}");
1756            return Ok(());
1757        }
1758
1759        let Some(index_name) = cmd
1760            .data_type
1761            .metadata()
1762            .as_ref()
1763            .and_then(|m| m.get("index_name"))
1764            .and_then(|v| v.as_str())
1765            .map(str::trim)
1766            .filter(|value| !value.is_empty())
1767            .map(ToString::to_string)
1768        else {
1769            log::warn!(
1770                "Rejected Deribit volatility index unsubscription: missing required metadata `index_name`"
1771            );
1772            return Ok(());
1773        };
1774
1775        log::debug!("Unsubscribing from Deribit volatility index: {index_name}");
1776
1777        let ws = self
1778            .ws_client
1779            .as_ref()
1780            .ok_or_else(|| anyhow::anyhow!("WebSocket client not initialized"))?
1781            .clone();
1782
1783        self.spawn_command(async move {
1784            if let Err(e) = ws.unsubscribe_volatility_index(&index_name).await {
1785                log::error!("Failed to unsubscribe from volatility index {index_name}: {e}");
1786            }
1787        });
1788
1789        Ok(())
1790    }
1791
1792    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1793        if request.start.is_some() {
1794            log::warn!(
1795                "Requesting instruments for {:?} with specified `start` which has no effect",
1796                request.venue
1797            );
1798        }
1799
1800        if request.end.is_some() {
1801            log::warn!(
1802                "Requesting instruments for {:?} with specified `end` which has no effect",
1803                request.venue
1804            );
1805        }
1806
1807        let http_client = self.http_client.clone();
1808        let ws_client = self.ws_client.clone();
1809        let instruments_cache = Arc::clone(&self.instruments);
1810        let sender = self.data_sender.clone();
1811        let request_id = request.request_id;
1812        let client_id = request.client_id.unwrap_or(self.client_id);
1813        let start_nanos = datetime_to_unix_nanos(request.start);
1814        let end_nanos = datetime_to_unix_nanos(request.end);
1815        let params = request.params;
1816        let clock = self.clock;
1817        let venue = *DERIBIT_VENUE;
1818
1819        // Get product types from config, default to Future if empty
1820        let product_types = if self.config.product_types.is_empty() {
1821            vec![crate::http::models::DeribitProductType::Future]
1822        } else {
1823            self.config.product_types.clone()
1824        };
1825
1826        self.spawn_command(async move {
1827            let mut all_instruments = Vec::new();
1828
1829            for product_type in &product_types {
1830                log::debug!(
1831                    "Requesting instruments for currency=ANY, product_type={product_type:?}"
1832                );
1833
1834                match http_client
1835                    .request_instruments(DeribitCurrency::ANY, Some(*product_type))
1836                    .await
1837                {
1838                    Ok(instruments) => {
1839                        log::debug!(
1840                            "Fetched {} instruments for ANY/{:?}",
1841                            instruments.len(),
1842                            product_type
1843                        );
1844
1845                        instruments_cache.rcu(|m| {
1846                            for instrument in &instruments {
1847                                m.insert(instrument.id(), instrument.clone());
1848                            }
1849                        });
1850                        all_instruments.extend(instruments);
1851                    }
1852                    Err(e) => {
1853                        log::error!("Failed to fetch instruments for ANY/{product_type:?}: {e:?}");
1854                    }
1855                }
1856            }
1857
1858            // Propagate to HTTP and WebSocket caches so downstream
1859            // requests use correct precisions.
1860            if !all_instruments.is_empty() {
1861                http_client.cache_instruments(&all_instruments);
1862
1863                if let Some(ws) = &ws_client {
1864                    ws.cache_instruments(&all_instruments);
1865                }
1866            }
1867
1868            // Send response with all collected instruments
1869            let response = DataResponse::Instruments(InstrumentsResponse::new(
1870                request_id,
1871                client_id,
1872                venue,
1873                all_instruments,
1874                start_nanos,
1875                end_nanos,
1876                clock.get_time_ns(),
1877                params,
1878            ));
1879
1880            if let Err(e) = sender.send(DataEvent::Response(response)) {
1881                log::error!("Failed to send instruments response: {e}");
1882            }
1883        });
1884
1885        Ok(())
1886    }
1887
1888    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1889        if request.start.is_some() {
1890            log::warn!(
1891                "Requesting instrument {} with specified `start` which has no effect",
1892                request.instrument_id
1893            );
1894        }
1895
1896        if request.end.is_some() {
1897            log::warn!(
1898                "Requesting instrument {} with specified `end` which has no effect",
1899                request.instrument_id
1900            );
1901        }
1902
1903        log::debug!("Fetching instrument {} from API", request.instrument_id);
1904
1905        let http_client = self.http_client.clone();
1906        let ws_client = self.ws_client.clone();
1907        let instruments_cache = Arc::clone(&self.instruments);
1908        let sender = self.data_sender.clone();
1909        let instrument_id = request.instrument_id;
1910        let request_id = request.request_id;
1911        let client_id = request.client_id.unwrap_or(self.client_id);
1912        let start_nanos = datetime_to_unix_nanos(request.start);
1913        let end_nanos = datetime_to_unix_nanos(request.end);
1914        let params = request.params;
1915        let clock = self.clock;
1916
1917        self.spawn_command(async move {
1918            match http_client
1919                .request_instrument(instrument_id)
1920                .await
1921                .context("failed to request instrument from Deribit")
1922            {
1923                Ok(instrument) => {
1924                    log::debug!("Successfully fetched instrument: {instrument_id}");
1925
1926                    instruments_cache.insert(instrument.id(), instrument.clone());
1927                    http_client.cache_instruments(std::slice::from_ref(&instrument));
1928
1929                    if let Some(ws) = &ws_client {
1930                        ws.cache_instruments(std::slice::from_ref(&instrument));
1931                    }
1932
1933                    // Send response
1934                    let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1935                        request_id,
1936                        client_id,
1937                        instrument.id(),
1938                        instrument,
1939                        start_nanos,
1940                        end_nanos,
1941                        clock.get_time_ns(),
1942                        params,
1943                    )));
1944
1945                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1946                        log::error!("Failed to send instrument response: {e}");
1947                    }
1948                }
1949                Err(e) => {
1950                    log::error!("Instrument request failed for {instrument_id}: {e:?}");
1951                }
1952            }
1953        });
1954
1955        Ok(())
1956    }
1957
1958    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1959        let http_client = self.http_client.clone();
1960        let sender = self.data_sender.clone();
1961        let instrument_id = request.instrument_id;
1962        let start = request.start;
1963        let end = request.end;
1964        let limit = request.limit.map(|n| n.get() as u32);
1965        let request_id = request.request_id;
1966        let client_id = request.client_id.unwrap_or(self.client_id);
1967        let params = request.params;
1968        let clock = self.clock;
1969        let start_nanos = datetime_to_unix_nanos(start);
1970        let end_nanos = datetime_to_unix_nanos(end);
1971
1972        self.spawn_command(async move {
1973            match http_client
1974                .request_trades(instrument_id, start, end, limit)
1975                .await
1976                .context("failed to request trades from Deribit")
1977            {
1978                Ok(trades) => {
1979                    let response = DataResponse::Trades(TradesResponse::new(
1980                        request_id,
1981                        client_id,
1982                        instrument_id,
1983                        trades,
1984                        start_nanos,
1985                        end_nanos,
1986                        clock.get_time_ns(),
1987                        params,
1988                    ));
1989
1990                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1991                        log::error!("Failed to send trades response: {e}");
1992                    }
1993                }
1994                Err(e) => log::error!("Trades request failed for {instrument_id}: {e:?}"),
1995            }
1996        });
1997
1998        Ok(())
1999    }
2000
2001    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
2002        let http_client = self.http_client.clone();
2003        let sender = self.data_sender.clone();
2004        let bar_type = request.bar_type;
2005        let start = request.start;
2006        let end = request.end;
2007        let limit = request.limit.map(|n| n.get() as u32);
2008        let request_id = request.request_id;
2009        let client_id = request.client_id.unwrap_or(self.client_id);
2010        let params = request.params;
2011        let clock = self.clock;
2012        let start_nanos = datetime_to_unix_nanos(start);
2013        let end_nanos = datetime_to_unix_nanos(end);
2014
2015        self.spawn_command(async move {
2016            match http_client
2017                .request_bars(bar_type, start, end, limit)
2018                .await
2019                .context("failed to request bars from Deribit")
2020            {
2021                Ok(bars) => {
2022                    let response = DataResponse::Bars(BarsResponse::new(
2023                        request_id,
2024                        client_id,
2025                        bar_type,
2026                        bars,
2027                        start_nanos,
2028                        end_nanos,
2029                        clock.get_time_ns(),
2030                        params,
2031                    ));
2032
2033                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2034                        log::error!("Failed to send bars response: {e}");
2035                    }
2036                }
2037                Err(e) => log::error!("Bars request failed for {bar_type}: {e:?}"),
2038            }
2039        });
2040
2041        Ok(())
2042    }
2043
2044    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
2045        let http_client = self.http_client.clone();
2046        let sender = self.data_sender.clone();
2047        let instrument_id = request.instrument_id;
2048        let depth = request.depth.map(|n| n.get() as u32);
2049        let request_id = request.request_id;
2050        let client_id = request.client_id.unwrap_or(self.client_id);
2051        let params = request.params;
2052        let clock = self.clock;
2053
2054        self.spawn_command(async move {
2055            match http_client
2056                .request_book_snapshot(instrument_id, depth)
2057                .await
2058                .context("failed to request book snapshot from Deribit")
2059            {
2060                Ok(book) => {
2061                    let response = DataResponse::Book(BookResponse::new(
2062                        request_id,
2063                        client_id,
2064                        instrument_id,
2065                        book,
2066                        None,
2067                        None,
2068                        clock.get_time_ns(),
2069                        params,
2070                    ));
2071
2072                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2073                        log::error!("Failed to send book snapshot response: {e}");
2074                    }
2075                }
2076                Err(e) => {
2077                    log::error!("Book snapshot request failed for {instrument_id}: {e:?}");
2078                }
2079            }
2080        });
2081
2082        Ok(())
2083    }
2084
2085    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
2086        if request.data_type.type_name() != Self::BOOK_SUMMARY_TYPE_NAME {
2087            log::warn!(
2088                "Unsupported custom data request: {}",
2089                request.data_type.type_name()
2090            );
2091            return Ok(());
2092        }
2093
2094        let currency = Self::book_summary_metadata_currency(&request.data_type)?;
2095        let kind = Self::book_summary_metadata_kind(&request.data_type);
2096        let kind_str = kind.as_deref().unwrap_or("option").to_string();
2097        let data_type = Self::book_summary_data_type(&currency, Some(&kind_str));
2098        let http_client = self.http_client.clone();
2099        let sender = self.data_sender.clone();
2100        let request_id = request.request_id;
2101        let client_id = request.client_id;
2102        let params = request.params;
2103        let clock = self.clock;
2104        let venue = *DERIBIT_VENUE;
2105        let start = request.start;
2106        let end = request.end;
2107        let start_nanos = datetime_to_unix_nanos(start);
2108        let end_nanos = datetime_to_unix_nanos(end);
2109
2110        self.spawn_command(async move {
2111            log::debug!(
2112                "Requesting Deribit book summaries for currency={currency} kind={kind_str}"
2113            );
2114
2115            match http_client
2116                .request_book_summaries_kind(&currency, Some(&kind_str))
2117                .await
2118            {
2119                Ok(summaries) => {
2120                    let ts = clock.get_time_ns();
2121                    let data: Vec<CustomData> = summaries
2122                        .into_iter()
2123                        .map(|raw| {
2124                            CustomData::new(
2125                                Arc::new(DeribitBookSummary::from_raw(raw, ts)),
2126                                data_type.clone(),
2127                            )
2128                        })
2129                        .collect();
2130
2131                    let response = DataResponse::Data(CustomDataResponse::new(
2132                        request_id,
2133                        client_id,
2134                        Some(venue),
2135                        data_type,
2136                        data,
2137                        start_nanos,
2138                        end_nanos,
2139                        ts,
2140                        params,
2141                    ));
2142
2143                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2144                        log::error!("Failed to send book summary response: {e}");
2145                    }
2146                }
2147                Err(e) => {
2148                    // Empty response keeps request correlation closed for strategy waiters.
2149                    log::error!(
2150                        "Book summary request failed for currency={currency} kind={kind_str}: {e:?}"
2151                    );
2152                    let ts = clock.get_time_ns();
2153                    let response = DataResponse::Data(CustomDataResponse::new(
2154                        request_id,
2155                        client_id,
2156                        Some(venue),
2157                        data_type,
2158                        Vec::<CustomData>::new(),
2159                        start_nanos,
2160                        end_nanos,
2161                        ts,
2162                        params,
2163                    ));
2164
2165                    if let Err(send_err) = sender.send(DataEvent::Response(response)) {
2166                        log::error!("Failed to send empty book summary response: {send_err}");
2167                    }
2168                }
2169            }
2170        });
2171
2172        Ok(())
2173    }
2174
2175    fn request_forward_prices(&self, request: RequestForwardPrices) -> anyhow::Result<()> {
2176        let currency = request.underlying.to_string();
2177        let instrument_id = request.instrument_id;
2178        let http_client = self.http_client.clone();
2179        let sender = self.data_sender.clone();
2180        let request_id = request.request_id;
2181        let client_id = request.client_id.unwrap_or(self.client_id());
2182        let params = request.params;
2183        let clock = self.clock;
2184        let venue = *DERIBIT_VENUE;
2185
2186        self.spawn_command(async move {
2187            let result = if let Some(inst_id) = instrument_id {
2188                // Single-instrument path: 1 HTTP call to public/ticker
2189                let instrument_name = inst_id.symbol.to_string();
2190                log::debug!(
2191                    "Requesting forward price for {currency} (single instrument: {instrument_name})"
2192                );
2193
2194                match http_client.request_ticker(&instrument_name).await {
2195                    Ok(ticker) => {
2196                        let ts = clock.get_time_ns();
2197                        let forward_prices: Vec<ForwardPrice> = ticker
2198                            .underlying_price
2199                            .map(|up| {
2200                                vec![ForwardPrice::new(
2201                                    inst_id,
2202                                    up,
2203                                    ticker.underlying_index.filter(|s| !s.is_empty()),
2204                                    ts,
2205                                    ts,
2206                                )]
2207                            })
2208                            .unwrap_or_default();
2209
2210                        log::debug!(
2211                            "Fetched {} forward price for {currency} (single instrument: {instrument_name})",
2212                            forward_prices.len(),
2213                        );
2214                        Ok((forward_prices, ts))
2215                    }
2216                    Err(e) => Err(e),
2217                }
2218            } else {
2219                // Bulk path: fetch all book summaries
2220                log::debug!("Requesting option forward prices for currency={currency} (bulk)");
2221
2222                match http_client.request_book_summaries(&currency).await {
2223                    Ok(summaries) => {
2224                        let ts = clock.get_time_ns();
2225
2226                        // Deduplicate: all options at the same expiry share the same
2227                        // forward price, so keep only one entry per underlying_index.
2228                        let mut seen_indices = std::collections::HashSet::new();
2229                        let forward_prices: Vec<ForwardPrice> = summaries
2230                            .into_iter()
2231                            .filter_map(|s| {
2232                                let up = s.underlying_price?;
2233                                let idx = s.underlying_index.clone().unwrap_or_default();
2234                                if !seen_indices.insert(idx.clone()) {
2235                                    return None;
2236                                }
2237                                Some(ForwardPrice::new(
2238                                    InstrumentId::new(
2239                                        Symbol::new(&s.instrument_name),
2240                                        *DERIBIT_VENUE,
2241                                    ),
2242                                    up,
2243                                    Some(idx).filter(|s| !s.is_empty()),
2244                                    ts,
2245                                    ts,
2246                                ))
2247                            })
2248                            .collect();
2249
2250                        log::debug!(
2251                            "Fetched {} forward prices (per-expiry) for {currency}",
2252                            forward_prices.len(),
2253                        );
2254                        Ok((forward_prices, ts))
2255                    }
2256                    Err(e) => Err(e),
2257                }
2258            };
2259
2260            match result {
2261                Ok((forward_prices, ts)) => {
2262                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
2263                        request_id,
2264                        client_id,
2265                        venue,
2266                        forward_prices,
2267                        ts,
2268                        params,
2269                    ));
2270
2271                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2272                        log::error!("Failed to send forward prices response: {e}");
2273                    }
2274                }
2275                Err(e) => {
2276                    log::error!("Forward prices request failed for {currency}: {e:?}");
2277                }
2278            }
2279        });
2280
2281        Ok(())
2282    }
2283}
nautilus_deribit/data.rs

nautilus_deribit/
data.rs
