Skip to main content

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