nautilus_deribit/websocket/
client.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//! WebSocket client for the Deribit API.
17//!
18//! The [`DeribitWebSocketClient`] provides connectivity to Deribit's WebSocket API using
19//! JSON-RPC 2.0. It supports subscribing to market data channels including trades, order books,
20//! and tickers.
21
22use std::{
23    fmt::Debug,
24    sync::{
25        Arc,
26        atomic::{AtomicBool, AtomicU8, Ordering},
27    },
28    time::Duration,
29};
30
31use arc_swap::ArcSwap;
32use futures_util::{FutureExt, Stream, StreamExt, stream::FuturesUnordered};
33use nautilus_common::{enums::LogColor, log_debug};
34use nautilus_core::{
35    AtomicMap, AtomicSet, consts::NAUTILUS_USER_AGENT, env::get_or_env_var_opt,
36    time::get_atomic_clock_realtime,
37};
38use nautilus_live::{SocketControl, task::TaskGroup};
39use nautilus_model::{
40    data::BarType,
41    enums::OrderSide,
42    identifiers::{AccountId, ClientOrderId, InstrumentId, StrategyId, TraderId},
43    instruments::{Instrument, InstrumentAny},
44    types::{Price, Quantity},
45};
46use nautilus_network::{
47    http::USER_AGENT,
48    mode::ConnectionMode,
49    websocket::{
50        AuthTracker, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
51        channel_message_handler,
52    },
53};
54use parking_lot::{Mutex, RwLock, RwLockReadGuard};
55use tokio_util::sync::CancellationToken;
56use ustr::Ustr;
57
58use super::{
59    auth::{AuthState, refresh_token_after_delay, send_auth_request},
60    enums::{DeribitUpdateInterval, DeribitWsChannel},
61    error::{DeribitWsError, DeribitWsResult},
62    handler::{DeribitWsFeedHandler, HandlerCommand},
63    messages::{
64        DeribitCancelAllByInstrumentParams, DeribitCancelParams, DeribitEditParams,
65        DeribitOrderParams, NautilusWsMessage,
66    },
67};
68use crate::common::{
69    consts::{
70        DERIBIT_TESTNET_WS_URL, DERIBIT_WS_HEARTBEAT_SECS, DERIBIT_WS_ORDER_KEY,
71        DERIBIT_WS_ORDER_QUOTA, DERIBIT_WS_SUBSCRIPTION_KEY, DERIBIT_WS_SUBSCRIPTION_QUOTA,
72        DERIBIT_WS_URL,
73    },
74    credential::{Credential, credential_env_vars},
75    enums::DeribitEnvironment,
76    parse::bar_spec_to_resolution,
77};
78
79/// Authentication timeout in seconds.
80const AUTHENTICATION_TIMEOUT_SECS: u64 = 30;
81
82type CommandSender = tokio::sync::mpsc::UnboundedSender<HandlerCommand>;
83
84/// WebSocket client for connecting to Deribit.
85#[derive(Clone)]
86pub struct DeribitWebSocketClient {
87    url: String,
88    environment: DeribitEnvironment,
89    heartbeat_interval: Option<u64>,
90    auth_timeout_secs: u64,
91    credential: Option<Credential>,
92    auth_state: Arc<tokio::sync::RwLock<Option<AuthState>>>,
93    signal: Arc<AtomicBool>,
94    connection_mode: Arc<ArcSwap<AtomicU8>>,
95    auth_tracker: AuthTracker,
96    cmd_tx: Arc<RwLock<CommandSender>>,
97    out_rx: Option<Arc<tokio::sync::mpsc::UnboundedReceiver<NautilusWsMessage>>>,
98    handler_tasks: Arc<TaskGroup>,
99    connect_lock: Arc<tokio::sync::Mutex<()>>,
100    subscriptions_state: SubscriptionState,
101    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
102    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
103    mark_price_subs: Arc<AtomicSet<InstrumentId>>,
104    index_price_subs: Arc<AtomicSet<InstrumentId>>,
105    cancellation_token: CancellationToken,
106    account_id: Option<AccountId>,
107    bars_timestamp_on_close: bool,
108    subscribe_errors: Arc<Mutex<Vec<String>>>,
109    transport_backend: TransportBackend,
110    proxy_url: Option<String>,
111    socket_control: Option<SocketControl>,
112}
113
114impl Debug for DeribitWebSocketClient {
115    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116        f.debug_struct(stringify!(DeribitWebSocketClient))
117            .field("url", &self.url)
118            .field("environment", &self.environment)
119            .field("has_credentials", &self.credential.is_some())
120            .field("is_authenticated", &self.auth_tracker.is_authenticated())
121            .field(
122                "has_auth_state",
123                &self.auth_state.try_read().is_ok_and(|s| s.is_some()),
124            )
125            .field("heartbeat_interval", &self.heartbeat_interval)
126            .finish_non_exhaustive()
127    }
128}
129
130impl DeribitWebSocketClient {
131    /// Creates a new [`DeribitWebSocketClient`] instance.
132    ///
133    /// Falls back to environment variables if credentials are not provided.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if only one of `api_key` or `api_secret` is provided.
138    #[expect(clippy::too_many_arguments)]
139    pub fn new(
140        url: Option<String>,
141        api_key: Option<String>,
142        api_secret: Option<String>,
143        heartbeat_interval: u64,
144        auth_timeout_secs: Option<u64>,
145        environment: DeribitEnvironment,
146        transport_backend: TransportBackend,
147        proxy_url: Option<String>,
148    ) -> anyhow::Result<Self> {
149        Self::new_inner(
150            url,
151            api_key,
152            api_secret,
153            heartbeat_interval,
154            auth_timeout_secs,
155            environment,
156            true,
157            transport_backend,
158            proxy_url,
159        )
160    }
161
162    /// Internal constructor with control over environment variable fallback.
163    #[expect(clippy::too_many_arguments)]
164    fn new_inner(
165        url: Option<String>,
166        api_key: Option<String>,
167        api_secret: Option<String>,
168        heartbeat_interval: u64,
169        auth_timeout_secs: Option<u64>,
170        environment: DeribitEnvironment,
171        env_fallback: bool,
172        transport_backend: TransportBackend,
173        proxy_url: Option<String>,
174    ) -> anyhow::Result<Self> {
175        let url = url.unwrap_or_else(|| match environment {
176            DeribitEnvironment::Testnet => DERIBIT_TESTNET_WS_URL.to_string(),
177            DeribitEnvironment::Mainnet => DERIBIT_WS_URL.to_string(),
178        });
179
180        // Resolve credential from config or environment variables (if env_fallback is true)
181        let credential =
182            Credential::resolve_with_env_fallback(api_key, api_secret, environment, env_fallback)?;
183
184        if credential.is_some() {
185            log::debug!("Credentials loaded ({environment})");
186        } else {
187            log::debug!("No credentials configured - unauthenticated mode");
188        }
189
190        let signal = Arc::new(AtomicBool::new(false));
191        let subscriptions_state = SubscriptionState::new('.');
192
193        Ok(Self {
194            url,
195            environment,
196            heartbeat_interval: Some(heartbeat_interval),
197            auth_timeout_secs: auth_timeout_secs.unwrap_or(AUTHENTICATION_TIMEOUT_SECS),
198            credential,
199            auth_state: Arc::new(tokio::sync::RwLock::new(None)),
200            signal,
201            connection_mode: Arc::new(ArcSwap::from_pointee(AtomicU8::new(
202                ConnectionMode::Closed.as_u8(),
203            ))),
204            auth_tracker: AuthTracker::new(),
205            cmd_tx: {
206                let (tx, _) = tokio::sync::mpsc::unbounded_channel();
207                Arc::new(RwLock::new(tx))
208            },
209            out_rx: None,
210            handler_tasks: Arc::new(TaskGroup::new()),
211            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
212            subscriptions_state,
213            instruments_cache: Arc::new(AtomicMap::new()),
214            option_greeks_subs: Arc::new(AtomicSet::new()),
215            mark_price_subs: Arc::new(AtomicSet::new()),
216            index_price_subs: Arc::new(AtomicSet::new()),
217            cancellation_token: CancellationToken::new(),
218            account_id: None,
219            bars_timestamp_on_close: true,
220            subscribe_errors: Arc::new(Mutex::new(Vec::new())),
221            transport_backend,
222            proxy_url,
223            socket_control: None,
224        })
225    }
226
227    pub(crate) fn begin_shutdown(&self) {
228        self.handler_tasks.begin_shutdown();
229        self.signal.store(true, Ordering::Relaxed);
230    }
231
232    /// Configures socket state reporting and reconnect control.
233    #[must_use]
234    pub fn with_socket_control(mut self, control: SocketControl) -> Self {
235        self.socket_control = Some(control);
236        self
237    }
238
239    /// Creates a new public (unauthenticated) client.
240    ///
241    /// Does NOT fall back to environment variables for credentials.
242    ///
243    /// # Errors
244    ///
245    /// Returns an error if initialization fails.
246    pub fn new_public(
247        environment: DeribitEnvironment,
248        proxy_url: Option<String>,
249    ) -> anyhow::Result<Self> {
250        Self::new_inner(
251            None,
252            None,
253            None,
254            DERIBIT_WS_HEARTBEAT_SECS,
255            None,
256            environment,
257            false,
258            TransportBackend::default(),
259            proxy_url,
260        )
261    }
262
263    /// Creates an unauthenticated client with a custom URL.
264    ///
265    /// Does NOT fall back to environment variables for credentials.
266    /// Useful for testing against mock servers.
267    ///
268    /// # Errors
269    ///
270    /// Returns an error if initialization fails.
271    pub fn new_unauthenticated(
272        url: Option<String>,
273        heartbeat_interval: u64,
274        environment: DeribitEnvironment,
275    ) -> anyhow::Result<Self> {
276        Self::new_inner(
277            url,
278            None,
279            None,
280            heartbeat_interval,
281            None,
282            environment,
283            false,
284            TransportBackend::default(),
285            None,
286        )
287    }
288
289    /// Creates an authenticated client with credentials.
290    ///
291    /// Resolves each credential from the provided argument first, falling back
292    /// to the environment variable for the given `environment`:
293    /// - Testnet: `DERIBIT_TESTNET_API_KEY` and `DERIBIT_TESTNET_API_SECRET`
294    /// - Mainnet: `DERIBIT_API_KEY` and `DERIBIT_API_SECRET`
295    ///
296    /// # Errors
297    ///
298    /// Returns an error if neither the argument nor the environment variable
299    /// provides a credential.
300    pub fn with_credentials(
301        environment: DeribitEnvironment,
302        api_key: Option<String>,
303        api_secret: Option<String>,
304        auth_timeout_secs: Option<u64>,
305        proxy_url: Option<String>,
306    ) -> anyhow::Result<Self> {
307        let (key_env, secret_env) = credential_env_vars(environment);
308
309        let api_key = get_or_env_var_opt(api_key, key_env)
310            .ok_or_else(|| anyhow::anyhow!("Missing environment variable: {key_env}"))?;
311        let api_secret = get_or_env_var_opt(api_secret, secret_env)
312            .ok_or_else(|| anyhow::anyhow!("Missing environment variable: {secret_env}"))?;
313
314        Self::new(
315            None,
316            Some(api_key),
317            Some(api_secret),
318            DERIBIT_WS_HEARTBEAT_SECS,
319            auth_timeout_secs,
320            environment,
321            TransportBackend::default(),
322            proxy_url,
323        )
324    }
325
326    /// Returns the current connection mode.
327    fn connection_mode(&self) -> ConnectionMode {
328        let mode_u8 = self.connection_mode.load().load(Ordering::Relaxed);
329        ConnectionMode::from_u8(mode_u8)
330    }
331
332    /// Returns whether the client is actively connected.
333    #[must_use]
334    pub fn is_active(&self) -> bool {
335        self.connection_mode() == ConnectionMode::Active
336    }
337
338    /// Returns the WebSocket URL.
339    #[must_use]
340    pub fn url(&self) -> &str {
341        &self.url
342    }
343
344    /// Returns the environment for this client.
345    #[must_use]
346    pub fn environment(&self) -> DeribitEnvironment {
347        self.environment
348    }
349
350    /// Returns whether the client is closed.
351    #[must_use]
352    pub fn is_closed(&self) -> bool {
353        let mode = self.connection_mode();
354        mode == ConnectionMode::Disconnect || mode == ConnectionMode::Closed
355    }
356
357    /// Cancel all pending WebSocket requests.
358    pub fn cancel_all_requests(&self) {
359        self.cancellation_token.cancel();
360    }
361
362    /// Returns the cancellation token for this client.
363    #[must_use]
364    pub fn cancellation_token(&self) -> &CancellationToken {
365        &self.cancellation_token
366    }
367
368    /// Waits until the client is active or timeout expires.
369    ///
370    /// # Errors
371    ///
372    /// Returns an error if the timeout expires before the client becomes active.
373    pub async fn wait_until_active(&self, timeout_secs: f64) -> DeribitWsResult<()> {
374        let timeout = tokio::time::Duration::from_secs_f64(timeout_secs);
375
376        tokio::time::timeout(timeout, async {
377            while !self.is_active() {
378                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
379            }
380        })
381        .await
382        .map_err(|_| {
383            DeribitWsError::Timeout(format!(
384                "WebSocket connection timeout after {timeout_secs} seconds"
385            ))
386        })?;
387
388        Ok(())
389    }
390
391    /// Waits until all pending subscriptions are confirmed by the server.
392    ///
393    /// # Errors
394    ///
395    /// Returns an error if the timeout expires before all subscriptions are confirmed.
396    pub async fn wait_for_subscriptions_confirmed(&self, timeout_secs: f64) -> DeribitWsResult<()> {
397        let timeout = Duration::from_secs_f64(timeout_secs);
398
399        tokio::time::timeout(timeout, async {
400            loop {
401                // Fail fast on permanent subscribe errors
402                let subscribe_error = {
403                    let mut errors = self.subscribe_errors.lock();
404                    if errors.is_empty() {
405                        None
406                    } else {
407                        let msg = errors.join("; ");
408                        errors.clear();
409                        Some(msg)
410                    }
411                };
412
413                if let Some(msg) = subscribe_error {
414                    return Err(DeribitWsError::Subscribe(msg));
415                }
416
417                let pending = self.subscriptions_state.pending_subscribe_topics();
418                if pending.is_empty() {
419                    return Ok(());
420                }
421                tokio::time::sleep(Duration::from_millis(10)).await;
422            }
423        })
424        .await
425        .map_err(|_| {
426            let pending = self.subscriptions_state.pending_subscribe_topics();
427            DeribitWsError::Timeout(format!(
428                "Subscription confirmation timeout after {timeout_secs}s, \
429                still pending: {pending:?}"
430            ))
431        })?
432    }
433
434    /// Caches instruments for use during message parsing.
435    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
436        let tx = self.command_sender();
437
438        self.instruments_cache.rcu(|m| {
439            for inst in instruments {
440                m.insert(inst.raw_symbol().inner(), inst.clone());
441            }
442        });
443        log::debug!("Cached {} instruments", self.instruments_cache.len());
444
445        // Send per-instrument updates to the live handler rather than
446        // a full snapshot, avoiding out-of-order snapshot races.
447        if self.is_active() {
448            for inst in instruments {
449                let _ = tx.send(HandlerCommand::UpdateInstrument(Box::new(inst.clone())));
450            }
451        }
452    }
453
454    /// Caches a single instrument.
455    pub fn cache_instrument(&self, instrument: InstrumentAny) {
456        let tx = self.command_sender();
457        let symbol = instrument.raw_symbol().inner();
458        self.instruments_cache.insert(symbol, instrument);
459
460        // If connected, send update to handler
461        if self.is_active() {
462            let inst = self.instruments_cache.get_cloned(&symbol);
463
464            if let Some(inst) = inst {
465                let _ = tx.send(HandlerCommand::UpdateInstrument(Box::new(inst)));
466            }
467        }
468    }
469
470    /// Sets the shared option greeks subscription set for handler-side gating.
471    pub fn set_option_greeks_subs(&mut self, subs: Arc<AtomicSet<InstrumentId>>) {
472        self.option_greeks_subs = subs;
473    }
474
475    /// Sets the shared mark price subscription set for handler-side gating.
476    pub fn set_mark_price_subs(&mut self, subs: Arc<AtomicSet<InstrumentId>>) {
477        self.mark_price_subs = subs;
478    }
479
480    /// Sets the shared index price subscription set for handler-side gating.
481    pub fn set_index_price_subs(&mut self, subs: Arc<AtomicSet<InstrumentId>>) {
482        self.index_price_subs = subs;
483    }
484
485    /// Registers an instrument for mark price emission from ticker messages.
486    pub fn add_mark_price_sub(&self, instrument_id: InstrumentId) {
487        self.mark_price_subs.insert(instrument_id);
488    }
489
490    /// Unregisters an instrument from mark price emission.
491    pub fn remove_mark_price_sub(&self, instrument_id: &InstrumentId) {
492        self.mark_price_subs.remove(instrument_id);
493    }
494
495    /// Registers an instrument for index price emission from ticker messages.
496    pub fn add_index_price_sub(&self, instrument_id: InstrumentId) {
497        self.index_price_subs.insert(instrument_id);
498    }
499
500    /// Unregisters an instrument from index price emission.
501    pub fn remove_index_price_sub(&self, instrument_id: &InstrumentId) {
502        self.index_price_subs.remove(instrument_id);
503    }
504
505    /// Registers an instrument for option greeks emission from ticker messages.
506    pub fn add_option_greeks_sub(&self, instrument_id: InstrumentId) {
507        self.option_greeks_subs.insert(instrument_id);
508    }
509
510    /// Unregisters an instrument from option greeks emission.
511    pub fn remove_option_greeks_sub(&self, instrument_id: &InstrumentId) {
512        self.option_greeks_subs.remove(instrument_id);
513    }
514
515    /// Connects to the Deribit WebSocket API.
516    ///
517    /// # Errors
518    ///
519    /// Returns an error if the connection fails.
520    pub async fn connect(&mut self) -> anyhow::Result<()> {
521        let connect_lock = Arc::clone(&self.connect_lock);
522        let _connect_guard = connect_lock.lock().await;
523
524        log_debug!(
525            "Connecting to WebSocket: {}",
526            self.url,
527            color = LogColor::Blue
528        );
529
530        if !self.handler_tasks.is_open() || !self.handler_tasks.is_empty() {
531            self.handler_tasks.begin_shutdown();
532            self.signal.store(true, Ordering::Relaxed);
533            self.finish_handler()
534                .await
535                .map_err(|e| anyhow::anyhow!("failed to stop prior WebSocket handler: {e}"))?;
536            self.handler_tasks.start_generation().map_err(|e| {
537                anyhow::anyhow!("failed to start WebSocket handler task generation: {e}")
538            })?;
539        }
540        let handler_spawner = self.handler_tasks.spawner().map_err(|e| {
541            anyhow::anyhow!("failed to acquire WebSocket handler task spawner: {e}")
542        })?;
543
544        // Reset stop signal and subscription state so callers can
545        // resubscribe cleanly after a manual disconnect/connect cycle.
546        self.signal.store(false, Ordering::Relaxed);
547        self.subscriptions_state.clear();
548
549        // Create message handler and channel
550        let (message_handler, raw_rx) = channel_message_handler();
551
552        // No-op ping handler: handler responds to pings directly
553        // Inbound Ping frames are answered by the transport, so no ping handler is needed;
554        // the reader routes them away from the message channel and the handler never sees them.
555
556        // Configure WebSocket client
557        let config = WebSocketConfig {
558            url: self.url.clone(),
559            headers: vec![(USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string())],
560            heartbeat_interval_secs: self.heartbeat_interval,
561            heartbeat_payload: None, // Deribit uses JSON-RPC heartbeat, not text ping
562            connect_timeout_ms: Some(5_000),
563            reconnect_delay_initial_ms: None,
564            reconnect_delay_max_ms: None,
565            reconnect_backoff_factor: None,
566            reconnect_jitter_ms: None,
567            reconnect_max_attempts: None,
568            heartbeat_timeout_secs: None,
569            idle_timeout_ms: None,
570            backend: self.transport_backend,
571            proxy_url: self.proxy_url.clone(),
572        };
573
574        // Configure rate limits
575        let keyed_quotas = vec![
576            (
577                DERIBIT_WS_SUBSCRIPTION_KEY.to_string(),
578                *DERIBIT_WS_SUBSCRIPTION_QUOTA,
579            ),
580            (DERIBIT_WS_ORDER_KEY.to_string(), *DERIBIT_WS_ORDER_QUOTA),
581        ];
582
583        // Connect the WebSocket
584        let ws_client = WebSocketClient::builder()
585            .config(config)
586            .message_handler(message_handler)
587            .keyed_quotas(keyed_quotas)
588            .default_quota(*DERIBIT_WS_SUBSCRIPTION_QUOTA)
589            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
590            .connect()
591            .await?;
592
593        // Store connection mode
594        self.connection_mode
595            .store(ws_client.connection_mode_atomic());
596        let reconnect_handle = ws_client.reconnect_handle();
597
598        // Create message channels
599        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
600        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
601
602        self.out_rx = Some(Arc::new(out_rx));
603
604        self.subscribe_errors.lock().clear();
605
606        // Create handler
607        let mut handler = DeribitWsFeedHandler::new(
608            self.signal.clone(),
609            cmd_rx,
610            raw_rx,
611            out_tx,
612            self.auth_tracker.clone(),
613            self.subscriptions_state.clone(),
614            self.option_greeks_subs.clone(),
615            self.mark_price_subs.clone(),
616            self.index_price_subs.clone(),
617            self.account_id,
618            self.bars_timestamp_on_close,
619            self.subscribe_errors.clone(),
620        );
621
622        if let Some(control) = &self.socket_control {
623            control.register(move || reconnect_handle.request_reconnect());
624        }
625
626        // Cache updates hold a read guard while mutating the central cache. Publishing the new
627        // sender under the write guard ensures later subscriptions follow this cache snapshot.
628        {
629            let mut command_sender = self.cmd_tx.write();
630            let _ = cmd_tx.send(HandlerCommand::SetClient(ws_client));
631
632            let instruments: Vec<InstrumentAny> =
633                self.instruments_cache.load().values().cloned().collect();
634
635            if !instruments.is_empty() {
636                log::debug!(
637                    "Sending {} cached instruments to handler",
638                    instruments.len()
639                );
640                let _ = cmd_tx.send(HandlerCommand::InitializeInstruments(instruments));
641            }
642
643            if let Some(interval) = self.heartbeat_interval {
644                let _ = cmd_tx.send(HandlerCommand::SetHeartbeat { interval });
645            }
646
647            *command_sender = cmd_tx.clone();
648        }
649
650        // Spawn handler task
651        let subscriptions_state = self.subscriptions_state.clone();
652        let credential = self.credential.clone();
653        let auth_tracker = self.auth_tracker.clone();
654        let auth_state = self.auth_state.clone();
655        let heartbeat_interval = self.heartbeat_interval;
656
657        let handler_task = async move {
658            const MAX_REAUTH_ATTEMPTS: u32 = 3;
659
660            let mut pending_reauth = false;
661            let mut reauth_attempts: u32 = 0;
662
663            let mut refresh_cancel = CancellationToken::new();
664            let mut retry_cancel = CancellationToken::new();
665            let mut lifecycle_futures = FuturesUnordered::new();
666
667            loop {
668                let message = {
669                    // `handler.next()` can dequeue work before awaiting transport I/O, so keep the
670                    // same future alive while handler-owned lifecycle futures complete.
671                    let next = handler.next();
672                    tokio::pin!(next);
673
674                    loop {
675                        tokio::select! {
676                            message = &mut next => break message,
677                            _ = lifecycle_futures.next(), if !lifecycle_futures.is_empty() => {}
678                        }
679                    }
680                };
681
682                match message {
683                    Some(msg) => match msg {
684                        NautilusWsMessage::Reconnected => {
685                            log::info!("Reconnected to WebSocket");
686
687                            // Cancel stale refresh and retry tasks from prior connection
688                            refresh_cancel.cancel();
689                            refresh_cancel = CancellationToken::new();
690                            retry_cancel.cancel();
691                            retry_cancel = CancellationToken::new();
692
693                            // Deribit scopes `set_heartbeat` to the connection, so the replacement
694                            // starts with heartbeats off and the venue sends no further
695                            // `test_request` until they are re-armed.
696                            if let Some(interval) = heartbeat_interval {
697                                let _ = cmd_tx.send(HandlerCommand::SetHeartbeat { interval });
698                            }
699
700                            let channels = subscriptions_state.reset_after_reconnect();
701
702                            // Check if we need to re-authenticate
703                            if let Some(cred) = &credential {
704                                log::info!("Re-authenticating after reconnection...");
705
706                                let _rx = auth_tracker.begin();
707                                pending_reauth = true;
708                                reauth_attempts = 1;
709
710                                let previous_scope =
711                                    auth_state.read().await.as_ref().map(|s| s.scope.clone());
712
713                                send_auth_request(cred, previous_scope, &cmd_tx);
714                            } else {
715                                // No credentials - resubscribe immediately
716                                if !channels.is_empty() {
717                                    let _ = cmd_tx.send(HandlerCommand::Subscribe { channels });
718                                }
719                            }
720                        }
721                        NautilusWsMessage::Authenticated(result) => {
722                            let timestamp = get_atomic_clock_realtime().get_time_ms();
723                            let new_auth_state = AuthState::from_auth_result(&result, timestamp);
724                            *auth_state.write().await = Some(new_auth_state);
725
726                            refresh_cancel.cancel();
727                            refresh_cancel = CancellationToken::new();
728                            retry_cancel.cancel();
729                            retry_cancel = CancellationToken::new();
730
731                            lifecycle_futures.push(
732                                refresh_token_after_delay(
733                                    result.expires_in,
734                                    result.refresh_token.clone(),
735                                    cmd_tx.clone(),
736                                    refresh_cancel.clone(),
737                                )
738                                .boxed(),
739                            );
740
741                            if pending_reauth {
742                                pending_reauth = false;
743                                reauth_attempts = 0;
744                                log::info!(
745                                    "Re-authentication successful (scope: {}), resubscribing to channels",
746                                    result.scope
747                                );
748
749                                let channels = subscriptions_state.all_topics();
750
751                                if !channels.is_empty() {
752                                    let _ = cmd_tx.send(HandlerCommand::Subscribe { channels });
753                                }
754                            } else {
755                                log::debug!(
756                                    "Auth state stored: scope={}, expires_in={}s",
757                                    result.scope,
758                                    result.expires_in
759                                );
760                            }
761                        }
762                        NautilusWsMessage::AuthenticationFailed(reason) => {
763                            if pending_reauth && reauth_attempts < MAX_REAUTH_ATTEMPTS {
764                                let delay_secs = 1u64 << reauth_attempts; // 2s, 4s
765                                log::warn!(
766                                    "Re-authentication attempt {reauth_attempts}/{MAX_REAUTH_ATTEMPTS} \
767                                    failed: {reason} - retrying in {delay_secs}s",
768                                );
769                                reauth_attempts += 1;
770
771                                // Spawn delayed retry so the handler loop keeps
772                                // processing messages during the backoff
773                                if let Some(cred) = &credential {
774                                    let cred = cred.clone();
775                                    let auth_state = auth_state.clone();
776                                    let auth_tracker = auth_tracker.clone();
777                                    let cmd_tx = cmd_tx.clone();
778                                    let cancel = retry_cancel.clone();
779
780                                    lifecycle_futures.push(
781                                        async move {
782                                            tokio::select! {
783                                                () = tokio::time::sleep(Duration::from_secs(delay_secs)) => {}
784                                                () = cancel.cancelled() => return,
785                                            }
786                                            let _rx = auth_tracker.begin();
787                                            let previous_scope = auth_state
788                                                .read()
789                                                .await
790                                                .as_ref()
791                                                .map(|s| s.scope.clone());
792                                            send_auth_request(&cred, previous_scope, &cmd_tx);
793                                        }
794                                        .boxed(),
795                                    );
796                                }
797                            } else if pending_reauth {
798                                pending_reauth = false;
799                                reauth_attempts = 0;
800                                log::error!(
801                                    "Re-authentication failed after {MAX_REAUTH_ATTEMPTS} \
802                                    attempts: {reason} \
803                                    - resubscribing to public channels only"
804                                );
805
806                                let all = subscriptions_state.all_topics();
807                                let mut public_channels = Vec::new();
808
809                                for ch in &all {
810                                    if DeribitWsChannel::requires_auth(ch) {
811                                        // Release private channels so future subscribe
812                                        // calls aren't skipped as already referenced
813                                        subscriptions_state.mark_unsubscribe(ch);
814                                        subscriptions_state.confirm_unsubscribe(ch);
815                                        subscriptions_state.remove_reference(ch);
816                                    } else {
817                                        public_channels.push(ch.clone());
818                                    }
819                                }
820
821                                if !public_channels.is_empty() {
822                                    let _ = cmd_tx.send(HandlerCommand::Subscribe {
823                                        channels: public_channels,
824                                    });
825                                }
826                            } else {
827                                log::error!("Authentication failed: {reason}");
828                            }
829                        }
830                        _ => {}
831                    },
832                    None => {
833                        log::debug!("Handler returned None, stopping task");
834                        break;
835                    }
836                }
837            }
838        };
839
840        if let Err(e) = handler_spawner.spawn(handler_task) {
841            if let Some(control) = &self.socket_control {
842                control.deregister();
843            }
844            self.out_rx = None;
845            anyhow::bail!("failed to register WebSocket handler task: {e}");
846        }
847        log::debug!("Connected to WebSocket");
848
849        Ok(())
850    }
851
852    /// Closes the WebSocket connection.
853    ///
854    /// # Errors
855    ///
856    /// Returns an error if the close operation fails.
857    pub async fn close(&self) -> DeribitWsResult<()> {
858        self.begin_shutdown();
859        let connect_lock = Arc::clone(&self.connect_lock);
860        let _connect_guard = connect_lock.lock().await;
861        self.close_locked().await
862    }
863
864    async fn close_locked(&self) -> DeribitWsResult<()> {
865        log::debug!("Closing WebSocket connection");
866        self.begin_shutdown();
867
868        let _ = self.command_sender().send(HandlerCommand::Disconnect);
869
870        self.finish_handler().await?;
871
872        self.auth_tracker.invalidate();
873
874        if let Some(control) = &self.socket_control {
875            control.deregister();
876        }
877        Ok(())
878    }
879
880    async fn finish_handler(&self) -> DeribitWsResult<()> {
881        self.handler_tasks
882            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
883            .await
884            .map_err(|e| {
885                DeribitWsError::ClientError(format!("WebSocket handler shutdown failed: {e}"))
886            })
887    }
888
889    /// Returns a stream of WebSocket messages.
890    ///
891    /// # Errors
892    ///
893    /// Returns an error if called before `connect()` or if called more than once.
894    pub fn stream(&mut self) -> DeribitWsResult<impl Stream<Item = NautilusWsMessage> + 'static> {
895        let rx = self.out_rx.take().ok_or_else(|| {
896            DeribitWsError::ClientError(
897                "Stream receiver already taken or not connected".to_string(),
898            )
899        })?;
900        let mut rx = Arc::try_unwrap(rx).map_err(|_| {
901            DeribitWsError::ClientError(
902                "Cannot take stream ownership - other references exist".to_string(),
903            )
904        })?;
905
906        Ok(async_stream::stream! {
907            while let Some(msg) = rx.recv().await {
908                yield msg;
909            }
910        })
911    }
912
913    /// Returns whether the client has credentials configured.
914    #[must_use]
915    pub fn has_credentials(&self) -> bool {
916        self.credential.is_some()
917    }
918
919    /// Returns whether the client is authenticated.
920    #[must_use]
921    pub fn is_authenticated(&self) -> bool {
922        self.auth_tracker.is_authenticated()
923    }
924
925    /// Authenticates the WebSocket session with Deribit.
926    ///
927    /// Uses the `client_signature` grant type with HMAC-SHA256 signature.
928    /// This must be called before subscribing to raw data streams.
929    ///
930    /// # Arguments
931    ///
932    /// * `session_name` - Optional session name for session-scoped authentication.
933    ///   When provided, uses `session:<name>` scope which allows skipping `access_token`
934    ///   in subsequent private requests. When `None`, uses default `connection` scope.
935    ///   Recommended to use session scope for order execution compatibility.
936    ///
937    /// # Errors
938    ///
939    /// Returns an error if:
940    /// - No credentials are configured
941    /// - The authentication request fails
942    /// - The authentication times out
943    pub async fn authenticate(&self, session_name: Option<&str>) -> DeribitWsResult<()> {
944        let credential = self.credential.as_ref().ok_or_else(|| {
945            DeribitWsError::Authentication("API credentials not configured".to_string())
946        })?;
947
948        // Determine scope
949        let scope = session_name.map(|name| format!("session:{name}"));
950
951        log::debug!("Authenticating WebSocket...");
952
953        let rx = self.auth_tracker.begin();
954
955        // Send authentication request
956        let cmd_tx = self.command_sender().clone();
957        send_auth_request(credential, scope, &cmd_tx);
958
959        // Wait for authentication result with timeout
960        match self
961            .auth_tracker
962            .wait_for_result::<DeribitWsError>(Duration::from_secs(self.auth_timeout_secs), rx)
963            .await
964        {
965            Ok(()) => {
966                log::debug!("WebSocket authenticated successfully");
967                Ok(())
968            }
969            Err(e) => {
970                log::error!("WebSocket authentication failed: error={e}");
971                Err(e)
972            }
973        }
974    }
975
976    /// Authenticates with session scope using the provided session name.
977    ///
978    /// Use `DERIBIT_DATA_SESSION_NAME` for data clients and
979    /// `DERIBIT_EXECUTION_SESSION_NAME` for execution clients.
980    ///
981    /// # Errors
982    ///
983    /// Returns an error if authentication fails.
984    pub async fn authenticate_session(&self, session_name: &str) -> DeribitWsResult<()> {
985        self.authenticate(Some(session_name)).await
986    }
987
988    /// Returns the current authentication state containing tokens.
989    ///
990    /// Returns `None` if not authenticated or tokens haven't been stored yet.
991    pub async fn auth_state(&self) -> Option<AuthState> {
992        self.auth_state.read().await.clone()
993    }
994
995    /// Returns the current access token if available.
996    pub async fn access_token(&self) -> Option<String> {
997        self.auth_state
998            .read()
999            .await
1000            .as_ref()
1001            .map(|s| s.access_token.clone())
1002    }
1003
1004    /// Sets the account ID for order/fill reports.
1005    pub fn set_account_id(&mut self, account_id: AccountId) {
1006        self.account_id = Some(account_id);
1007    }
1008
1009    /// Sets whether bar timestamps should use the close time.
1010    ///
1011    /// When `true` (default), bar `ts_event` is set to the bar's close time.
1012    pub fn set_bars_timestamp_on_close(&mut self, value: bool) {
1013        self.bars_timestamp_on_close = value;
1014    }
1015
1016    async fn send_subscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1017        let mut channels_to_subscribe = Vec::new();
1018
1019        for channel in channels {
1020            if self.subscriptions_state.add_reference(&channel) {
1021                self.subscriptions_state.mark_subscribe(&channel);
1022                channels_to_subscribe.push(channel);
1023            } else {
1024                log::debug!("Already subscribed to {channel}, skipping duplicate subscription");
1025            }
1026        }
1027
1028        if channels_to_subscribe.is_empty() {
1029            return Ok(());
1030        }
1031
1032        if let Err(e) = self.command_sender().send(HandlerCommand::Subscribe {
1033            channels: channels_to_subscribe.clone(),
1034        }) {
1035            // Roll back: remove reference and clear pending_subscribe
1036            for channel in &channels_to_subscribe {
1037                self.subscriptions_state.remove_reference(channel);
1038                self.subscriptions_state.mark_unsubscribe(channel);
1039                self.subscriptions_state.confirm_unsubscribe(channel);
1040            }
1041            return Err(DeribitWsError::Send(e.to_string()));
1042        }
1043
1044        log::debug!(
1045            "Sent subscribe for {} channels",
1046            channels_to_subscribe.len()
1047        );
1048        Ok(())
1049    }
1050
1051    async fn send_unsubscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1052        let mut channels_to_unsubscribe = Vec::new();
1053
1054        for channel in channels {
1055            if self.subscriptions_state.remove_reference(&channel) {
1056                self.subscriptions_state.mark_unsubscribe(&channel);
1057                channels_to_unsubscribe.push(channel);
1058            } else {
1059                log::debug!("Still has references to {channel}, skipping unsubscription");
1060            }
1061        }
1062
1063        if channels_to_unsubscribe.is_empty() {
1064            return Ok(());
1065        }
1066
1067        if let Err(e) = self.command_sender().send(HandlerCommand::Unsubscribe {
1068            channels: channels_to_unsubscribe.clone(),
1069        }) {
1070            // Send only fails when the handler task is dead, meaning the
1071            // connection is broken. Restore refcount and mark confirmed so
1072            // the topic is not wedged in pending_unsubscribe. This may
1073            // promote a pending_subscribe topic to confirmed, but that is
1074            // harmless: connect() calls clear() on the next connection
1075            // attempt, resetting all subscription state.
1076            for channel in &channels_to_unsubscribe {
1077                self.subscriptions_state.confirm_unsubscribe(channel);
1078                self.subscriptions_state.add_reference(channel);
1079                self.subscriptions_state.mark_subscribe(channel);
1080                self.subscriptions_state.confirm_subscribe(channel);
1081            }
1082            return Err(DeribitWsError::Send(e.to_string()));
1083        }
1084
1085        log::debug!(
1086            "Sent unsubscribe for {} channels",
1087            channels_to_unsubscribe.len()
1088        );
1089        Ok(())
1090    }
1091
1092    /// Subscribes to trade updates for an instrument.
1093    ///
1094    /// # Arguments
1095    ///
1096    /// * `instrument_id` - The instrument to subscribe to
1097    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1098    ///
1099    /// # Errors
1100    ///
1101    /// Returns an error if subscription fails or raw is requested without authentication.
1102    pub async fn subscribe_trades(
1103        &self,
1104        instrument_id: InstrumentId,
1105        interval: Option<DeribitUpdateInterval>,
1106    ) -> DeribitWsResult<()> {
1107        let interval = interval.unwrap_or_default();
1108        self.check_auth_requirement(interval)?;
1109        let channel =
1110            DeribitWsChannel::Trades.format_channel(instrument_id.symbol.as_str(), Some(interval));
1111        self.send_subscribe(vec![channel]).await
1112    }
1113
1114    /// Unsubscribes from trade updates for an instrument.
1115    ///
1116    /// # Errors
1117    ///
1118    /// Returns an error if unsubscription fails.
1119    pub async fn unsubscribe_trades(
1120        &self,
1121        instrument_id: InstrumentId,
1122        interval: Option<DeribitUpdateInterval>,
1123    ) -> DeribitWsResult<()> {
1124        let interval = interval.unwrap_or_default();
1125        let channel =
1126            DeribitWsChannel::Trades.format_channel(instrument_id.symbol.as_str(), Some(interval));
1127        self.send_unsubscribe(vec![channel]).await
1128    }
1129
1130    /// Subscribes to order book updates for an instrument.
1131    ///
1132    /// # Arguments
1133    ///
1134    /// * `instrument_id` - The instrument to subscribe to
1135    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1136    ///
1137    /// # Errors
1138    ///
1139    /// Returns an error if subscription fails or raw is requested without authentication.
1140    pub async fn subscribe_book(
1141        &self,
1142        instrument_id: InstrumentId,
1143        interval: Option<DeribitUpdateInterval>,
1144    ) -> DeribitWsResult<()> {
1145        let interval = interval.unwrap_or_default();
1146        self.check_auth_requirement(interval)?;
1147        let channel =
1148            DeribitWsChannel::Book.format_channel(instrument_id.symbol.as_str(), Some(interval));
1149        self.send_subscribe(vec![channel]).await
1150    }
1151
1152    /// Unsubscribes from order book updates for an instrument.
1153    ///
1154    /// # Errors
1155    ///
1156    /// Returns an error if unsubscription fails.
1157    pub async fn unsubscribe_book(
1158        &self,
1159        instrument_id: InstrumentId,
1160        interval: Option<DeribitUpdateInterval>,
1161    ) -> DeribitWsResult<()> {
1162        let interval = interval.unwrap_or_default();
1163        let channel =
1164            DeribitWsChannel::Book.format_channel(instrument_id.symbol.as_str(), Some(interval));
1165        self.send_unsubscribe(vec![channel]).await
1166    }
1167
1168    /// Subscribes to grouped (depth-limited) order book updates for an instrument.
1169    ///
1170    /// Uses the Deribit grouped book channel format: `book.{instrument}.{group}.{depth}.{interval}`
1171    ///
1172    /// Depth is normalized to Deribit supported values: 1, 10, or 20.
1173    ///
1174    /// # Errors
1175    ///
1176    /// Returns an error if subscription fails or raw is requested without authentication.
1177    pub async fn subscribe_book_grouped(
1178        &self,
1179        instrument_id: InstrumentId,
1180        group: &str,
1181        depth: u32,
1182        interval: Option<DeribitUpdateInterval>,
1183    ) -> DeribitWsResult<()> {
1184        // Grouped book channel only supports 100ms and agg2, not raw
1185        let interval = match interval {
1186            Some(DeribitUpdateInterval::Raw) | None => DeribitUpdateInterval::Ms100,
1187            Some(i) => i,
1188        };
1189
1190        let normalized_depth = if depth < 5 {
1191            1
1192        } else if depth < 15 {
1193            10
1194        } else {
1195            20
1196        };
1197
1198        let channel = format!(
1199            "book.{}.{}.{}.{}",
1200            instrument_id.symbol,
1201            group,
1202            normalized_depth,
1203            interval.as_str()
1204        );
1205        log::debug!("Subscribing to grouped book channel: {channel}");
1206        self.send_subscribe(vec![channel]).await
1207    }
1208
1209    /// Unsubscribes from grouped (depth-limited) order book updates for an instrument.
1210    ///
1211    /// Depth is normalized to Deribit supported values: 1, 10, or 20.
1212    ///
1213    /// # Errors
1214    ///
1215    /// Returns an error if unsubscription fails.
1216    pub async fn unsubscribe_book_grouped(
1217        &self,
1218        instrument_id: InstrumentId,
1219        group: &str,
1220        depth: u32,
1221        interval: Option<DeribitUpdateInterval>,
1222    ) -> DeribitWsResult<()> {
1223        // Grouped book channel only supports 100ms and agg2, not raw
1224        let interval = match interval {
1225            Some(DeribitUpdateInterval::Raw) | None => DeribitUpdateInterval::Ms100,
1226            Some(i) => i,
1227        };
1228
1229        let normalized_depth = if depth < 5 {
1230            1
1231        } else if depth < 15 {
1232            10
1233        } else {
1234            20
1235        };
1236
1237        let channel = format!(
1238            "book.{}.{}.{}.{}",
1239            instrument_id.symbol,
1240            group,
1241            normalized_depth,
1242            interval.as_str()
1243        );
1244        self.send_unsubscribe(vec![channel]).await
1245    }
1246
1247    /// Subscribes to ticker updates for an instrument.
1248    ///
1249    /// # Arguments
1250    ///
1251    /// * `instrument_id` - The instrument to subscribe to
1252    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1253    ///
1254    /// # Errors
1255    ///
1256    /// Returns an error if subscription fails or raw is requested without authentication.
1257    pub async fn subscribe_ticker(
1258        &self,
1259        instrument_id: InstrumentId,
1260        interval: Option<DeribitUpdateInterval>,
1261    ) -> DeribitWsResult<()> {
1262        let interval = interval.unwrap_or_default();
1263        self.check_auth_requirement(interval)?;
1264        let channel =
1265            DeribitWsChannel::Ticker.format_channel(instrument_id.symbol.as_str(), Some(interval));
1266        self.send_subscribe(vec![channel]).await
1267    }
1268
1269    /// Unsubscribes from ticker updates for an instrument.
1270    ///
1271    /// # Errors
1272    ///
1273    /// Returns an error if unsubscription fails.
1274    pub async fn unsubscribe_ticker(
1275        &self,
1276        instrument_id: InstrumentId,
1277        interval: Option<DeribitUpdateInterval>,
1278    ) -> DeribitWsResult<()> {
1279        let interval = interval.unwrap_or_default();
1280        let channel =
1281            DeribitWsChannel::Ticker.format_channel(instrument_id.symbol.as_str(), Some(interval));
1282        self.send_unsubscribe(vec![channel]).await
1283    }
1284
1285    /// Subscribes to quote (best bid/ask) updates for an instrument.
1286    ///
1287    /// Note: Quote channel does not support interval parameter.
1288    ///
1289    /// # Errors
1290    ///
1291    /// Returns an error if subscription fails.
1292    pub async fn subscribe_quotes(&self, instrument_id: InstrumentId) -> DeribitWsResult<()> {
1293        let channel = DeribitWsChannel::Quote.format_channel(instrument_id.symbol.as_str(), None);
1294        self.send_subscribe(vec![channel]).await
1295    }
1296
1297    /// Unsubscribes from quote updates for an instrument.
1298    ///
1299    /// # Errors
1300    ///
1301    /// Returns an error if unsubscription fails.
1302    pub async fn unsubscribe_quotes(&self, instrument_id: InstrumentId) -> DeribitWsResult<()> {
1303        let channel = DeribitWsChannel::Quote.format_channel(instrument_id.symbol.as_str(), None);
1304        self.send_unsubscribe(vec![channel]).await
1305    }
1306
1307    /// Subscribes to instrument status changes for lifecycle notifications.
1308    ///
1309    /// Channel format: `instrument.state.{kind}.{currency}`
1310    ///
1311    /// # Errors
1312    ///
1313    /// Returns an error if subscription fails.
1314    pub async fn subscribe_instrument_status(
1315        &self,
1316        kind: &str,
1317        currency: &str,
1318    ) -> DeribitWsResult<()> {
1319        let channel = DeribitWsChannel::format_instrument_state_channel(kind, currency);
1320        self.send_subscribe(vec![channel]).await
1321    }
1322
1323    /// Unsubscribes from instrument status changes.
1324    ///
1325    /// # Errors
1326    ///
1327    /// Returns an error if unsubscription fails.
1328    pub async fn unsubscribe_instrument_status(
1329        &self,
1330        kind: &str,
1331        currency: &str,
1332    ) -> DeribitWsResult<()> {
1333        let channel = DeribitWsChannel::format_instrument_state_channel(kind, currency);
1334        self.send_unsubscribe(vec![channel]).await
1335    }
1336
1337    /// Subscribes to volatility index updates for the given index name.
1338    ///
1339    /// Channel format: `deribit_volatility_index.{index_name}`
1340    ///
1341    /// # Errors
1342    ///
1343    /// Returns an error if subscription fails.
1344    pub async fn subscribe_volatility_index(&self, index_name: &str) -> DeribitWsResult<()> {
1345        let channel = DeribitWsChannel::VolatilityIndex.format_channel(index_name, None);
1346        self.send_subscribe(vec![channel]).await
1347    }
1348
1349    /// Unsubscribes from volatility index updates for the given index name.
1350    ///
1351    /// # Errors
1352    ///
1353    /// Returns an error if unsubscription fails.
1354    pub async fn unsubscribe_volatility_index(&self, index_name: &str) -> DeribitWsResult<()> {
1355        let channel = DeribitWsChannel::VolatilityIndex.format_channel(index_name, None);
1356        self.send_unsubscribe(vec![channel]).await
1357    }
1358
1359    /// Subscribes to perpetual interest rates updates.
1360    ///
1361    /// Channel format: `perpetual.{instrument_name}.{interval}`
1362    ///
1363    /// # Errors
1364    ///
1365    /// Returns an error if subscription fails.
1366    pub async fn subscribe_perpetual_interests_rates_updates(
1367        &self,
1368        instrument_id: InstrumentId,
1369        interval: Option<DeribitUpdateInterval>,
1370    ) -> DeribitWsResult<()> {
1371        let interval = interval.unwrap_or(DeribitUpdateInterval::Ms100);
1372        let channel = DeribitWsChannel::Perpetual
1373            .format_channel(instrument_id.symbol.as_str(), Some(interval));
1374
1375        self.send_subscribe(vec![channel]).await
1376    }
1377
1378    /// Unsubscribes from perpetual interest rates updates.
1379    ///
1380    /// # Errors
1381    ///
1382    /// Returns an error if subscription fails.
1383    pub async fn unsubscribe_perpetual_interest_rates_updates(
1384        &self,
1385        instrument_id: InstrumentId,
1386        interval: Option<DeribitUpdateInterval>,
1387    ) -> DeribitWsResult<()> {
1388        let interval = interval.unwrap_or(DeribitUpdateInterval::Ms100);
1389        let channel = DeribitWsChannel::Perpetual
1390            .format_channel(instrument_id.symbol.as_str(), Some(interval));
1391
1392        self.send_unsubscribe(vec![channel]).await
1393    }
1394
1395    /// Subscribes to chart/OHLC bar updates for an instrument.
1396    ///
1397    /// # Arguments
1398    ///
1399    /// * `instrument_id` - The instrument to subscribe to
1400    /// * `resolution` - Bar resolution: "1", "3", "5", "10", "15", "30", "60", "120", "180",
1401    ///   "360", "720", "1D" (minutes or 1D for daily)
1402    ///
1403    /// # Errors
1404    ///
1405    /// Returns an error if subscription fails.
1406    pub async fn subscribe_chart(
1407        &self,
1408        instrument_id: InstrumentId,
1409        resolution: &str,
1410    ) -> DeribitWsResult<()> {
1411        // Chart channel format: chart.trades.{instrument}.{resolution}
1412        let channel = format!("chart.trades.{}.{}", instrument_id.symbol, resolution);
1413        self.send_subscribe(vec![channel]).await
1414    }
1415
1416    /// Unsubscribes from chart/OHLC bar updates.
1417    ///
1418    /// # Errors
1419    ///
1420    /// Returns an error if unsubscription fails.
1421    pub async fn unsubscribe_chart(
1422        &self,
1423        instrument_id: InstrumentId,
1424        resolution: &str,
1425    ) -> DeribitWsResult<()> {
1426        let channel = format!("chart.trades.{}.{}", instrument_id.symbol, resolution);
1427        self.send_unsubscribe(vec![channel]).await
1428    }
1429
1430    /// Subscribes to bar updates for an instrument using a BarType specification.
1431    ///
1432    /// Converts the BarType to the nearest supported Deribit resolution and subscribes
1433    /// to the chart channel.
1434    ///
1435    /// # Errors
1436    ///
1437    /// Returns an error if the subscription request fails.
1438    pub async fn subscribe_bars(&self, bar_type: BarType) -> DeribitWsResult<()> {
1439        let resolution = bar_spec_to_resolution(&bar_type);
1440        self.subscribe_chart(bar_type.instrument_id(), &resolution)
1441            .await
1442    }
1443
1444    /// Unsubscribes from bar updates for an instrument using a BarType specification.
1445    ///
1446    /// # Errors
1447    ///
1448    /// Returns an error if the unsubscription request fails.
1449    pub async fn unsubscribe_bars(&self, bar_type: BarType) -> DeribitWsResult<()> {
1450        let resolution = bar_spec_to_resolution(&bar_type);
1451        self.unsubscribe_chart(bar_type.instrument_id(), &resolution)
1452            .await
1453    }
1454
1455    /// Checks if authentication is required for the given interval.
1456    ///
1457    /// # Errors
1458    ///
1459    /// Returns an error if raw interval is requested but client is not authenticated.
1460    fn check_auth_requirement(&self, interval: DeribitUpdateInterval) -> DeribitWsResult<()> {
1461        if interval.requires_auth() && !self.is_authenticated() {
1462            return Err(DeribitWsError::Authentication(
1463                "Raw streams require authentication. Call authenticate() first.".to_string(),
1464            ));
1465        }
1466        Ok(())
1467    }
1468
1469    /// Subscribes to user order updates for all instruments.
1470    ///
1471    /// Requires authentication. Subscribes to `user.orders.any.any.raw` channel.
1472    ///
1473    /// # Errors
1474    ///
1475    /// Returns an error if client is not authenticated or subscription fails.
1476    pub async fn subscribe_user_orders(&self) -> DeribitWsResult<()> {
1477        if !self.is_authenticated() {
1478            return Err(DeribitWsError::Authentication(
1479                "User orders subscription requires authentication".to_string(),
1480            ));
1481        }
1482        self.send_subscribe(vec!["user.orders.any.any.raw".to_string()])
1483            .await
1484    }
1485
1486    /// Unsubscribes from user order updates for all instruments.
1487    ///
1488    /// # Errors
1489    ///
1490    /// Returns an error if unsubscription fails.
1491    pub async fn unsubscribe_user_orders(&self) -> DeribitWsResult<()> {
1492        self.send_unsubscribe(vec!["user.orders.any.any.raw".to_string()])
1493            .await
1494    }
1495
1496    /// Subscribes to user trade/fill updates for all instruments.
1497    ///
1498    /// Requires authentication. Subscribes to `user.trades.any.any.raw` channel.
1499    ///
1500    /// # Errors
1501    ///
1502    /// Returns an error if client is not authenticated or subscription fails.
1503    pub async fn subscribe_user_trades(&self) -> DeribitWsResult<()> {
1504        if !self.is_authenticated() {
1505            return Err(DeribitWsError::Authentication(
1506                "User trades subscription requires authentication".to_string(),
1507            ));
1508        }
1509        self.send_subscribe(vec!["user.trades.any.any.raw".to_string()])
1510            .await
1511    }
1512
1513    /// Unsubscribes from user trade/fill updates for all instruments.
1514    ///
1515    /// # Errors
1516    ///
1517    /// Returns an error if unsubscription fails.
1518    pub async fn unsubscribe_user_trades(&self) -> DeribitWsResult<()> {
1519        self.send_unsubscribe(vec!["user.trades.any.any.raw".to_string()])
1520            .await
1521    }
1522
1523    /// Subscribes to user portfolio updates for all currencies.
1524    ///
1525    /// Requires authentication. Subscribes to `user.portfolio.any` channel which
1526    /// provides real-time account balance and margin updates for all currencies
1527    /// (BTC, ETH, USDC, USDT, etc.).
1528    ///
1529    /// # Errors
1530    ///
1531    /// Returns an error if client is not authenticated or subscription fails.
1532    pub async fn subscribe_user_portfolio(&self) -> DeribitWsResult<()> {
1533        if !self.is_authenticated() {
1534            return Err(DeribitWsError::Authentication(
1535                "User portfolio subscription requires authentication".to_string(),
1536            ));
1537        }
1538        self.send_subscribe(vec!["user.portfolio.any".to_string()])
1539            .await
1540    }
1541
1542    /// Unsubscribes from user portfolio updates for all currencies.
1543    ///
1544    /// # Errors
1545    ///
1546    /// Returns an error if unsubscription fails.
1547    pub async fn unsubscribe_user_portfolio(&self) -> DeribitWsResult<()> {
1548        self.send_unsubscribe(vec!["user.portfolio.any".to_string()])
1549            .await
1550    }
1551
1552    /// Subscribes to multiple channels at once.
1553    ///
1554    /// # Errors
1555    ///
1556    /// Returns an error if subscription fails.
1557    pub async fn subscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1558        self.send_subscribe(channels).await
1559    }
1560
1561    /// Unsubscribes from multiple channels at once.
1562    ///
1563    /// # Errors
1564    ///
1565    /// Returns an error if unsubscription fails.
1566    pub async fn unsubscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1567        self.send_unsubscribe(channels).await
1568    }
1569
1570    /// Submits an order to Deribit via WebSocket.
1571    ///
1572    /// Routes to `private/buy` or `private/sell` JSON-RPC method based on order side.
1573    /// Requires authentication (call `authenticate_session()` first).
1574    ///
1575    /// # Errors
1576    ///
1577    /// Returns an error if:
1578    /// - The client is not authenticated
1579    /// - The command fails to send
1580    pub async fn submit_order(
1581        &self,
1582        order_side: OrderSide,
1583        params: DeribitOrderParams,
1584        client_order_id: ClientOrderId,
1585        trader_id: TraderId,
1586        strategy_id: StrategyId,
1587        instrument_id: InstrumentId,
1588    ) -> DeribitWsResult<()> {
1589        if !self.is_authenticated() {
1590            return Err(DeribitWsError::Authentication(
1591                "Submit order requires authentication. Call authenticate_session() first."
1592                    .to_string(),
1593            ));
1594        }
1595
1596        log::debug!(
1597            "Sending {} order: instrument={}, amount={}, price={:?}, client_order_id={}",
1598            order_side,
1599            params.instrument_name,
1600            params.amount,
1601            params.price,
1602            client_order_id
1603        );
1604
1605        let cmd = match order_side {
1606            OrderSide::Buy => HandlerCommand::Buy {
1607                params,
1608                client_order_id,
1609                trader_id,
1610                strategy_id,
1611                instrument_id,
1612            },
1613            OrderSide::Sell => HandlerCommand::Sell {
1614                params,
1615                client_order_id,
1616                trader_id,
1617                strategy_id,
1618                instrument_id,
1619            },
1620        };
1621
1622        self.command_sender()
1623            .send(cmd)
1624            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1625
1626        Ok(())
1627    }
1628
1629    /// Modifies an existing order on Deribit via WebSocket.
1630    ///
1631    /// The order parameters are sent using the `private/edit` JSON-RPC method.
1632    /// Requires authentication (call `authenticate_session()` first).
1633    ///
1634    /// # Errors
1635    ///
1636    /// Returns an error if:
1637    /// - The client is not authenticated
1638    /// - The command fails to send
1639    #[expect(clippy::too_many_arguments)]
1640    pub async fn modify_order(
1641        &self,
1642        order_id: &str,
1643        quantity: Quantity,
1644        price: Price,
1645        client_order_id: ClientOrderId,
1646        trader_id: TraderId,
1647        strategy_id: StrategyId,
1648        instrument_id: InstrumentId,
1649    ) -> DeribitWsResult<()> {
1650        if !self.is_authenticated() {
1651            return Err(DeribitWsError::Authentication(
1652                "Modify order requires authentication. Call authenticate_session() first."
1653                    .to_string(),
1654            ));
1655        }
1656
1657        let params = DeribitEditParams {
1658            order_id: order_id.to_string(),
1659            amount: quantity.as_decimal(),
1660            price: Some(price.as_decimal()),
1661            post_only: None,
1662            reject_post_only: None,
1663            reduce_only: None,
1664            trigger_price: None,
1665        };
1666
1667        log::debug!(
1668            "Sending modify order: order_id={order_id}, quantity={quantity}, price={price}, client_order_id={client_order_id}"
1669        );
1670
1671        self.command_sender()
1672            .send(HandlerCommand::Edit {
1673                params,
1674                client_order_id,
1675                trader_id,
1676                strategy_id,
1677                instrument_id,
1678            })
1679            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1680
1681        Ok(())
1682    }
1683
1684    /// Cancels an existing order on Deribit via WebSocket.
1685    ///
1686    /// The order is cancelled using the `private/cancel` JSON-RPC method.
1687    /// Requires authentication (call `authenticate_session()` first).
1688    ///
1689    /// # Errors
1690    ///
1691    /// Returns an error if:
1692    /// - The client is not authenticated
1693    /// - The command fails to send
1694    pub async fn cancel_order(
1695        &self,
1696        order_id: &str,
1697        client_order_id: ClientOrderId,
1698        trader_id: TraderId,
1699        strategy_id: StrategyId,
1700        instrument_id: InstrumentId,
1701    ) -> DeribitWsResult<()> {
1702        if !self.is_authenticated() {
1703            return Err(DeribitWsError::Authentication(
1704                "Cancel order requires authentication. Call authenticate_session() first."
1705                    .to_string(),
1706            ));
1707        }
1708
1709        let params = DeribitCancelParams {
1710            order_id: order_id.to_string(),
1711        };
1712
1713        log::debug!("Sending cancel order: order_id={order_id}, client_order_id={client_order_id}");
1714
1715        self.command_sender()
1716            .send(HandlerCommand::Cancel {
1717                params,
1718                client_order_id,
1719                trader_id,
1720                strategy_id,
1721                instrument_id,
1722            })
1723            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1724
1725        Ok(())
1726    }
1727
1728    /// Cancels all orders for a specific instrument on Deribit via WebSocket.
1729    ///
1730    /// Uses the `private/cancel_all_by_instrument` JSON-RPC method.
1731    /// Requires authentication (call `authenticate_session()` first).
1732    ///
1733    /// # Errors
1734    ///
1735    /// Returns an error if:
1736    /// - The client is not authenticated
1737    /// - The command fails to send
1738    pub async fn cancel_all_orders(
1739        &self,
1740        instrument_id: InstrumentId,
1741        order_type: Option<String>,
1742    ) -> DeribitWsResult<()> {
1743        if !self.is_authenticated() {
1744            return Err(DeribitWsError::Authentication(
1745                "Cancel all orders requires authentication. Call authenticate_session() first."
1746                    .to_string(),
1747            ));
1748        }
1749
1750        let instrument_name = instrument_id.symbol.to_string();
1751        let params = DeribitCancelAllByInstrumentParams {
1752            instrument_name: instrument_name.clone(),
1753            order_type,
1754        };
1755
1756        log::debug!("Sending cancel_all_orders: instrument={instrument_name}");
1757
1758        self.command_sender()
1759            .send(HandlerCommand::CancelAllByInstrument {
1760                params,
1761                instrument_id,
1762            })
1763            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1764
1765        Ok(())
1766    }
1767
1768    /// Queries the state of an order on Deribit via WebSocket.
1769    ///
1770    /// Uses the `private/get_order_state` JSON-RPC method.
1771    /// Requires authentication (call `authenticate_session()` first).
1772    ///
1773    /// # Errors
1774    ///
1775    /// Returns an error if:
1776    /// - The client is not authenticated
1777    /// - The command fails to send
1778    pub async fn query_order(
1779        &self,
1780        order_id: &str,
1781        client_order_id: ClientOrderId,
1782        trader_id: TraderId,
1783        strategy_id: StrategyId,
1784        instrument_id: InstrumentId,
1785    ) -> DeribitWsResult<()> {
1786        if !self.is_authenticated() {
1787            return Err(DeribitWsError::Authentication(
1788                "Query order state requires authentication. Call authenticate_session() first."
1789                    .to_string(),
1790            ));
1791        }
1792
1793        log::debug!("Sending query_order: order_id={order_id}, client_order_id={client_order_id}");
1794
1795        self.command_sender()
1796            .send(HandlerCommand::GetOrderState {
1797                order_id: order_id.to_string(),
1798                client_order_id,
1799                trader_id,
1800                strategy_id,
1801                instrument_id,
1802            })
1803            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1804
1805        Ok(())
1806    }
1807
1808    fn command_sender(&self) -> RwLockReadGuard<'_, CommandSender> {
1809        self.cmd_tx.read()
1810    }
1811}
1812
1813#[cfg(test)]
1814mod tests {
1815    use rstest::rstest;
1816
1817    use super::*;
1818
1819    struct DropSignal(Arc<AtomicBool>);
1820
1821    impl Drop for DropSignal {
1822        fn drop(&mut self) {
1823            self.0.store(true, Ordering::Release);
1824        }
1825    }
1826
1827    #[tokio::test]
1828    async fn test_last_client_owner_drop_aborts_handler_task() {
1829        let client = DeribitWebSocketClient::new_unauthenticated(
1830            Some("ws://127.0.0.1:0/ws/api/v2".to_string()),
1831            30,
1832            DeribitEnvironment::Testnet,
1833        )
1834        .unwrap();
1835        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
1836        let dropped = Arc::new(AtomicBool::new(false));
1837        let drop_signal = DropSignal(Arc::clone(&dropped));
1838        client
1839            .handler_tasks
1840            .spawn(async move {
1841                let _drop_signal = drop_signal;
1842                started_tx.send(()).expect("started receiver");
1843                std::future::pending::<()>().await;
1844            })
1845            .expect("handler task should register");
1846        started_rx.await.expect("handler task started");
1847        let clone = client.clone();
1848
1849        drop(client);
1850        assert!(!dropped.load(Ordering::Acquire));
1851        drop(clone);
1852
1853        tokio::time::timeout(Duration::from_secs(1), async {
1854            while !dropped.load(Ordering::Acquire) {
1855                tokio::task::yield_now().await;
1856            }
1857        })
1858        .await
1859        .expect("handler task aborted");
1860    }
1861
1862    #[rstest]
1863    #[tokio::test]
1864    async fn test_unsubscribe_send_failure_restores_subscription() {
1865        let client = DeribitWebSocketClient::new_unauthenticated(
1866            Some("ws://127.0.0.1:0/ws/api/v2".to_string()),
1867            30,
1868            DeribitEnvironment::Testnet,
1869        )
1870        .unwrap();
1871        let channel = "trades.BTC-PERPETUAL.raw";
1872        client.subscriptions_state.add_reference(channel);
1873        client.subscriptions_state.mark_subscribe(channel);
1874        client.subscriptions_state.confirm_subscribe(channel);
1875
1876        let error = client
1877            .send_unsubscribe(vec![channel.to_string()])
1878            .await
1879            .unwrap_err();
1880
1881        assert!(matches!(error, DeribitWsError::Send(_)));
1882        assert_eq!(client.subscriptions_state.get_reference_count(channel), 1);
1883        assert_eq!(client.subscriptions_state.len(), 1);
1884        assert_eq!(client.subscriptions_state.all_topics(), [channel]);
1885        assert!(
1886            client
1887                .subscriptions_state
1888                .pending_subscribe_topics()
1889                .is_empty()
1890        );
1891        assert!(
1892            client
1893                .subscriptions_state
1894                .pending_unsubscribe_topics()
1895                .is_empty()
1896        );
1897    }
1898}
nautilus_deribit/websocket/client.rs

nautilus_deribit/websocket/
client.rs
