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, env::get_or_env_var_opt, string::secret::SecretString,
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::create_standard_nautilus_headers,
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<SecretString>,
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: proxy_url.map(SecretString::from),
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 headers = create_standard_nautilus_headers();
558
559        let config = WebSocketConfig {
560            url: self.url.clone(),
561            headers,
562            heartbeat_interval_secs: self.heartbeat_interval,
563            heartbeat_payload: None, // Deribit uses JSON-RPC heartbeat, not text ping
564            connect_timeout_ms: Some(5_000),
565            reconnect_delay_initial_ms: None,
566            reconnect_delay_max_ms: None,
567            reconnect_backoff_factor: None,
568            reconnect_jitter_ms: None,
569            reconnect_max_attempts: None,
570            heartbeat_timeout_secs: None,
571            idle_timeout_ms: None,
572            backend: self.transport_backend,
573            proxy_url: self
574                .proxy_url
575                .as_ref()
576                .map(|value| value.expose_secret().to_owned()),
577        };
578
579        // Configure rate limits
580        let keyed_quotas = vec![
581            (
582                DERIBIT_WS_SUBSCRIPTION_KEY.to_string(),
583                *DERIBIT_WS_SUBSCRIPTION_QUOTA,
584            ),
585            (DERIBIT_WS_ORDER_KEY.to_string(), *DERIBIT_WS_ORDER_QUOTA),
586        ];
587
588        // Connect the WebSocket
589        let ws_client = WebSocketClient::builder()
590            .config(config)
591            .message_handler(message_handler)
592            .keyed_quotas(keyed_quotas)
593            .default_quota(*DERIBIT_WS_SUBSCRIPTION_QUOTA)
594            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
595            .connect()
596            .await?;
597
598        // Store connection mode
599        self.connection_mode
600            .store(ws_client.connection_mode_atomic());
601        let reconnect_handle = ws_client.reconnect_handle();
602
603        // Create message channels
604        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
605        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
606
607        self.out_rx = Some(Arc::new(out_rx));
608
609        self.subscribe_errors.lock().clear();
610
611        // Create handler
612        let mut handler = DeribitWsFeedHandler::new(
613            self.signal.clone(),
614            cmd_rx,
615            raw_rx,
616            out_tx,
617            self.auth_tracker.clone(),
618            self.subscriptions_state.clone(),
619            self.option_greeks_subs.clone(),
620            self.mark_price_subs.clone(),
621            self.index_price_subs.clone(),
622            self.account_id,
623            self.bars_timestamp_on_close,
624            self.subscribe_errors.clone(),
625        );
626
627        if let Some(control) = &self.socket_control {
628            control.register(move || reconnect_handle.request_reconnect());
629        }
630
631        // Cache updates hold a read guard while mutating the central cache. Publishing the new
632        // sender under the write guard ensures later subscriptions follow this cache snapshot.
633        {
634            let mut command_sender = self.cmd_tx.write();
635            let _ = cmd_tx.send(HandlerCommand::SetClient(ws_client));
636
637            let instruments: Vec<InstrumentAny> =
638                self.instruments_cache.load().values().cloned().collect();
639
640            if !instruments.is_empty() {
641                log::debug!(
642                    "Sending {} cached instruments to handler",
643                    instruments.len()
644                );
645                let _ = cmd_tx.send(HandlerCommand::InitializeInstruments(instruments));
646            }
647
648            if let Some(interval) = self.heartbeat_interval {
649                let _ = cmd_tx.send(HandlerCommand::SetHeartbeat { interval });
650            }
651
652            *command_sender = cmd_tx.clone();
653        }
654
655        // Spawn handler task
656        let subscriptions_state = self.subscriptions_state.clone();
657        let credential = self.credential.clone();
658        let auth_tracker = self.auth_tracker.clone();
659        let auth_state = self.auth_state.clone();
660        let heartbeat_interval = self.heartbeat_interval;
661
662        let handler_task = async move {
663            const MAX_REAUTH_ATTEMPTS: u32 = 3;
664
665            let mut pending_reauth = false;
666            let mut reauth_attempts: u32 = 0;
667
668            let mut refresh_cancel = CancellationToken::new();
669            let mut retry_cancel = CancellationToken::new();
670            let mut lifecycle_futures = FuturesUnordered::new();
671
672            loop {
673                let message = {
674                    // `handler.next()` can dequeue work before awaiting transport I/O, so keep the
675                    // same future alive while handler-owned lifecycle futures complete.
676                    let next = handler.next();
677                    tokio::pin!(next);
678
679                    loop {
680                        tokio::select! {
681                            message = &mut next => break message,
682                            _ = lifecycle_futures.next(), if !lifecycle_futures.is_empty() => {}
683                        }
684                    }
685                };
686
687                match message {
688                    Some(msg) => match msg {
689                        NautilusWsMessage::Reconnected => {
690                            log::info!("Reconnected to WebSocket");
691
692                            // Cancel stale refresh and retry tasks from prior connection
693                            refresh_cancel.cancel();
694                            refresh_cancel = CancellationToken::new();
695                            retry_cancel.cancel();
696                            retry_cancel = CancellationToken::new();
697
698                            // Deribit scopes `set_heartbeat` to the connection, so the replacement
699                            // starts with heartbeats off and the venue sends no further
700                            // `test_request` until they are re-armed.
701                            if let Some(interval) = heartbeat_interval {
702                                let _ = cmd_tx.send(HandlerCommand::SetHeartbeat { interval });
703                            }
704
705                            let channels = subscriptions_state.reset_after_reconnect();
706
707                            // Check if we need to re-authenticate
708                            if let Some(cred) = &credential {
709                                log::info!("Re-authenticating after reconnection...");
710
711                                let _rx = auth_tracker.begin();
712                                pending_reauth = true;
713                                reauth_attempts = 1;
714
715                                let previous_scope =
716                                    auth_state.read().await.as_ref().map(|s| s.scope.clone());
717
718                                send_auth_request(cred, previous_scope, &cmd_tx);
719                            } else {
720                                // No credentials - resubscribe immediately
721                                if !channels.is_empty() {
722                                    let _ = cmd_tx.send(HandlerCommand::Subscribe { channels });
723                                }
724                            }
725                        }
726                        NautilusWsMessage::Authenticated(result) => {
727                            let result = *result;
728                            let timestamp = get_atomic_clock_realtime().get_time_ms();
729                            let new_auth_state = AuthState::from_auth_result(&result, timestamp);
730                            *auth_state.write().await = Some(new_auth_state);
731
732                            refresh_cancel.cancel();
733                            refresh_cancel = CancellationToken::new();
734                            retry_cancel.cancel();
735                            retry_cancel = CancellationToken::new();
736
737                            lifecycle_futures.push(
738                                refresh_token_after_delay(
739                                    result.expires_in,
740                                    result.refresh_token.clone(),
741                                    cmd_tx.clone(),
742                                    refresh_cancel.clone(),
743                                )
744                                .boxed(),
745                            );
746
747                            if pending_reauth {
748                                pending_reauth = false;
749                                reauth_attempts = 0;
750                                log::info!(
751                                    "Re-authentication successful (scope: {}), resubscribing to channels",
752                                    result.scope
753                                );
754
755                                let channels = subscriptions_state.all_topics();
756
757                                if !channels.is_empty() {
758                                    let _ = cmd_tx.send(HandlerCommand::Subscribe { channels });
759                                }
760                            } else {
761                                log::debug!(
762                                    "Auth state stored: scope={}, expires_in={}s",
763                                    result.scope,
764                                    result.expires_in
765                                );
766                            }
767                        }
768                        NautilusWsMessage::AuthenticationFailed(reason) => {
769                            if pending_reauth && reauth_attempts < MAX_REAUTH_ATTEMPTS {
770                                let delay_secs = 1u64 << reauth_attempts; // 2s, 4s
771                                log::warn!(
772                                    "Re-authentication attempt {reauth_attempts}/{MAX_REAUTH_ATTEMPTS} \
773                                    failed: {reason} - retrying in {delay_secs}s",
774                                );
775                                reauth_attempts += 1;
776
777                                // Spawn delayed retry so the handler loop keeps
778                                // processing messages during the backoff
779                                if let Some(cred) = &credential {
780                                    let cred = cred.clone();
781                                    let auth_state = auth_state.clone();
782                                    let auth_tracker = auth_tracker.clone();
783                                    let cmd_tx = cmd_tx.clone();
784                                    let cancel = retry_cancel.clone();
785
786                                    lifecycle_futures.push(
787                                        async move {
788                                            tokio::select! {
789                                                () = tokio::time::sleep(Duration::from_secs(delay_secs)) => {}
790                                                () = cancel.cancelled() => return,
791                                            }
792                                            let _rx = auth_tracker.begin();
793                                            let previous_scope = auth_state
794                                                .read()
795                                                .await
796                                                .as_ref()
797                                                .map(|s| s.scope.clone());
798                                            send_auth_request(&cred, previous_scope, &cmd_tx);
799                                        }
800                                        .boxed(),
801                                    );
802                                }
803                            } else if pending_reauth {
804                                pending_reauth = false;
805                                reauth_attempts = 0;
806                                log::error!(
807                                    "Re-authentication failed after {MAX_REAUTH_ATTEMPTS} \
808                                    attempts: {reason} \
809                                    - resubscribing to public channels only"
810                                );
811
812                                let all = subscriptions_state.all_topics();
813                                let mut public_channels = Vec::new();
814
815                                for ch in &all {
816                                    if DeribitWsChannel::requires_auth(ch) {
817                                        // Release private channels so future subscribe
818                                        // calls aren't skipped as already referenced
819                                        subscriptions_state.mark_unsubscribe(ch);
820                                        subscriptions_state.confirm_unsubscribe(ch);
821                                        subscriptions_state.remove_reference(ch);
822                                    } else {
823                                        public_channels.push(ch.clone());
824                                    }
825                                }
826
827                                if !public_channels.is_empty() {
828                                    let _ = cmd_tx.send(HandlerCommand::Subscribe {
829                                        channels: public_channels,
830                                    });
831                                }
832                            } else {
833                                log::error!("Authentication failed: {reason}");
834                            }
835                        }
836                        _ => {}
837                    },
838                    None => {
839                        log::debug!("Handler returned None, stopping task");
840                        break;
841                    }
842                }
843            }
844        };
845
846        if let Err(e) = handler_spawner.spawn(handler_task) {
847            if let Some(control) = &self.socket_control {
848                control.deregister();
849            }
850            self.out_rx = None;
851            anyhow::bail!("failed to register WebSocket handler task: {e}");
852        }
853        log::debug!("Connected to WebSocket");
854
855        Ok(())
856    }
857
858    /// Closes the WebSocket connection.
859    ///
860    /// # Errors
861    ///
862    /// Returns an error if the close operation fails.
863    pub async fn close(&self) -> DeribitWsResult<()> {
864        self.begin_shutdown();
865        let connect_lock = Arc::clone(&self.connect_lock);
866        let _connect_guard = connect_lock.lock().await;
867        self.close_locked().await
868    }
869
870    async fn close_locked(&self) -> DeribitWsResult<()> {
871        log::debug!("Closing WebSocket connection");
872        self.begin_shutdown();
873
874        let _ = self.command_sender().send(HandlerCommand::Disconnect);
875
876        self.finish_handler().await?;
877
878        self.auth_tracker.invalidate();
879
880        if let Some(control) = &self.socket_control {
881            control.deregister();
882        }
883        Ok(())
884    }
885
886    async fn finish_handler(&self) -> DeribitWsResult<()> {
887        self.handler_tasks
888            .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2))
889            .await
890            .map_err(|e| {
891                DeribitWsError::ClientError(format!("WebSocket handler shutdown failed: {e}"))
892            })
893    }
894
895    /// Returns a stream of WebSocket messages.
896    ///
897    /// # Errors
898    ///
899    /// Returns an error if called before `connect()` or if called more than once.
900    pub fn stream(&mut self) -> DeribitWsResult<impl Stream<Item = NautilusWsMessage> + 'static> {
901        let rx = self.out_rx.take().ok_or_else(|| {
902            DeribitWsError::ClientError(
903                "Stream receiver already taken or not connected".to_string(),
904            )
905        })?;
906        let mut rx = Arc::try_unwrap(rx).map_err(|_| {
907            DeribitWsError::ClientError(
908                "Cannot take stream ownership - other references exist".to_string(),
909            )
910        })?;
911
912        Ok(async_stream::stream! {
913            while let Some(msg) = rx.recv().await {
914                yield msg;
915            }
916        })
917    }
918
919    /// Returns whether the client has credentials configured.
920    #[must_use]
921    pub fn has_credentials(&self) -> bool {
922        self.credential.is_some()
923    }
924
925    /// Returns whether the client is authenticated.
926    #[must_use]
927    pub fn is_authenticated(&self) -> bool {
928        self.auth_tracker.is_authenticated()
929    }
930
931    /// Authenticates the WebSocket session with Deribit.
932    ///
933    /// Uses the `client_signature` grant type with HMAC-SHA256 signature.
934    /// This must be called before subscribing to raw data streams.
935    ///
936    /// # Arguments
937    ///
938    /// * `session_name` - Optional session name for session-scoped authentication.
939    ///   When provided, uses `session:<name>` scope which allows skipping `access_token`
940    ///   in subsequent private requests. When `None`, uses default `connection` scope.
941    ///   Recommended to use session scope for order execution compatibility.
942    ///
943    /// # Errors
944    ///
945    /// Returns an error if:
946    /// - No credentials are configured
947    /// - The authentication request fails
948    /// - The authentication times out
949    pub async fn authenticate(&self, session_name: Option<&str>) -> DeribitWsResult<()> {
950        let credential = self.credential.as_ref().ok_or_else(|| {
951            DeribitWsError::Authentication("API credentials not configured".to_string())
952        })?;
953
954        // Determine scope
955        let scope = session_name.map(|name| format!("session:{name}"));
956
957        log::debug!("Authenticating WebSocket...");
958
959        let rx = self.auth_tracker.begin();
960
961        // Send authentication request
962        let cmd_tx = self.command_sender().clone();
963        send_auth_request(credential, scope, &cmd_tx);
964
965        // Wait for authentication result with timeout
966        match self
967            .auth_tracker
968            .wait_for_result::<DeribitWsError>(Duration::from_secs(self.auth_timeout_secs), rx)
969            .await
970        {
971            Ok(()) => {
972                log::debug!("WebSocket authenticated successfully");
973                Ok(())
974            }
975            Err(e) => {
976                log::error!("WebSocket authentication failed: error={e}");
977                Err(e)
978            }
979        }
980    }
981
982    /// Authenticates with session scope using the provided session name.
983    ///
984    /// Use `DERIBIT_DATA_SESSION_NAME` for data clients and
985    /// `DERIBIT_EXECUTION_SESSION_NAME` for execution clients.
986    ///
987    /// # Errors
988    ///
989    /// Returns an error if authentication fails.
990    pub async fn authenticate_session(&self, session_name: &str) -> DeribitWsResult<()> {
991        self.authenticate(Some(session_name)).await
992    }
993
994    /// Returns the current authentication state containing tokens.
995    ///
996    /// Returns `None` if not authenticated or tokens haven't been stored yet.
997    pub async fn auth_state(&self) -> Option<AuthState> {
998        self.auth_state.read().await.clone()
999    }
1000
1001    /// Returns the current access token if available.
1002    pub async fn access_token(&self) -> Option<SecretString> {
1003        self.auth_state
1004            .read()
1005            .await
1006            .as_ref()
1007            .map(|s| s.access_token.clone())
1008    }
1009
1010    /// Sets the account ID for order/fill reports.
1011    pub fn set_account_id(&mut self, account_id: AccountId) {
1012        self.account_id = Some(account_id);
1013    }
1014
1015    /// Sets whether bar timestamps should use the close time.
1016    ///
1017    /// When `true` (default), bar `ts_event` is set to the bar's close time.
1018    pub fn set_bars_timestamp_on_close(&mut self, value: bool) {
1019        self.bars_timestamp_on_close = value;
1020    }
1021
1022    async fn send_subscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1023        let mut channels_to_subscribe = Vec::new();
1024
1025        for channel in channels {
1026            if self.subscriptions_state.add_reference(&channel) {
1027                self.subscriptions_state.mark_subscribe(&channel);
1028                channels_to_subscribe.push(channel);
1029            } else {
1030                log::debug!("Already subscribed to {channel}, skipping duplicate subscription");
1031            }
1032        }
1033
1034        if channels_to_subscribe.is_empty() {
1035            return Ok(());
1036        }
1037
1038        if let Err(e) = self.command_sender().send(HandlerCommand::Subscribe {
1039            channels: channels_to_subscribe.clone(),
1040        }) {
1041            // Roll back: remove reference and clear pending_subscribe
1042            for channel in &channels_to_subscribe {
1043                self.subscriptions_state.remove_reference(channel);
1044                self.subscriptions_state.mark_unsubscribe(channel);
1045                self.subscriptions_state.confirm_unsubscribe(channel);
1046            }
1047            return Err(DeribitWsError::Send(e.to_string()));
1048        }
1049
1050        log::debug!(
1051            "Sent subscribe for {} channels",
1052            channels_to_subscribe.len()
1053        );
1054        Ok(())
1055    }
1056
1057    async fn send_unsubscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1058        let mut channels_to_unsubscribe = Vec::new();
1059
1060        for channel in channels {
1061            if self.subscriptions_state.remove_reference(&channel) {
1062                self.subscriptions_state.mark_unsubscribe(&channel);
1063                channels_to_unsubscribe.push(channel);
1064            } else {
1065                log::debug!("Still has references to {channel}, skipping unsubscription");
1066            }
1067        }
1068
1069        if channels_to_unsubscribe.is_empty() {
1070            return Ok(());
1071        }
1072
1073        if let Err(e) = self.command_sender().send(HandlerCommand::Unsubscribe {
1074            channels: channels_to_unsubscribe.clone(),
1075        }) {
1076            // Send only fails when the handler task is dead, meaning the
1077            // connection is broken. Restore refcount and mark confirmed so
1078            // the topic is not wedged in pending_unsubscribe. This may
1079            // promote a pending_subscribe topic to confirmed, but that is
1080            // harmless: connect() calls clear() on the next connection
1081            // attempt, resetting all subscription state.
1082            for channel in &channels_to_unsubscribe {
1083                self.subscriptions_state.confirm_unsubscribe(channel);
1084                self.subscriptions_state.add_reference(channel);
1085                self.subscriptions_state.mark_subscribe(channel);
1086                self.subscriptions_state.confirm_subscribe(channel);
1087            }
1088            return Err(DeribitWsError::Send(e.to_string()));
1089        }
1090
1091        log::debug!(
1092            "Sent unsubscribe for {} channels",
1093            channels_to_unsubscribe.len()
1094        );
1095        Ok(())
1096    }
1097
1098    /// Subscribes to trade updates for an instrument.
1099    ///
1100    /// # Arguments
1101    ///
1102    /// * `instrument_id` - The instrument to subscribe to
1103    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1104    ///
1105    /// # Errors
1106    ///
1107    /// Returns an error if subscription fails or raw is requested without authentication.
1108    pub async fn subscribe_trades(
1109        &self,
1110        instrument_id: InstrumentId,
1111        interval: Option<DeribitUpdateInterval>,
1112    ) -> DeribitWsResult<()> {
1113        let interval = interval.unwrap_or_default();
1114        self.check_auth_requirement(interval)?;
1115        let channel =
1116            DeribitWsChannel::Trades.format_channel(instrument_id.symbol.as_str(), Some(interval));
1117        self.send_subscribe(vec![channel]).await
1118    }
1119
1120    /// Unsubscribes from trade updates for an instrument.
1121    ///
1122    /// # Errors
1123    ///
1124    /// Returns an error if unsubscription fails.
1125    pub async fn unsubscribe_trades(
1126        &self,
1127        instrument_id: InstrumentId,
1128        interval: Option<DeribitUpdateInterval>,
1129    ) -> DeribitWsResult<()> {
1130        let interval = interval.unwrap_or_default();
1131        let channel =
1132            DeribitWsChannel::Trades.format_channel(instrument_id.symbol.as_str(), Some(interval));
1133        self.send_unsubscribe(vec![channel]).await
1134    }
1135
1136    /// Subscribes to order book updates for an instrument.
1137    ///
1138    /// # Arguments
1139    ///
1140    /// * `instrument_id` - The instrument to subscribe to
1141    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1142    ///
1143    /// # Errors
1144    ///
1145    /// Returns an error if subscription fails or raw is requested without authentication.
1146    pub async fn subscribe_book(
1147        &self,
1148        instrument_id: InstrumentId,
1149        interval: Option<DeribitUpdateInterval>,
1150    ) -> DeribitWsResult<()> {
1151        let interval = interval.unwrap_or_default();
1152        self.check_auth_requirement(interval)?;
1153        let channel =
1154            DeribitWsChannel::Book.format_channel(instrument_id.symbol.as_str(), Some(interval));
1155        self.send_subscribe(vec![channel]).await
1156    }
1157
1158    /// Unsubscribes from order book updates for an instrument.
1159    ///
1160    /// # Errors
1161    ///
1162    /// Returns an error if unsubscription fails.
1163    pub async fn unsubscribe_book(
1164        &self,
1165        instrument_id: InstrumentId,
1166        interval: Option<DeribitUpdateInterval>,
1167    ) -> DeribitWsResult<()> {
1168        let interval = interval.unwrap_or_default();
1169        let channel =
1170            DeribitWsChannel::Book.format_channel(instrument_id.symbol.as_str(), Some(interval));
1171        self.send_unsubscribe(vec![channel]).await
1172    }
1173
1174    /// Subscribes to grouped (depth-limited) order book updates for an instrument.
1175    ///
1176    /// Uses the Deribit grouped book channel format: `book.{instrument}.{group}.{depth}.{interval}`
1177    ///
1178    /// Depth is normalized to Deribit supported values: 1, 10, or 20.
1179    ///
1180    /// # Errors
1181    ///
1182    /// Returns an error if subscription fails or raw is requested without authentication.
1183    pub async fn subscribe_book_grouped(
1184        &self,
1185        instrument_id: InstrumentId,
1186        group: &str,
1187        depth: u32,
1188        interval: Option<DeribitUpdateInterval>,
1189    ) -> DeribitWsResult<()> {
1190        // Grouped book channel only supports 100ms and agg2, not raw
1191        let interval = match interval {
1192            Some(DeribitUpdateInterval::Raw) | None => DeribitUpdateInterval::Ms100,
1193            Some(i) => i,
1194        };
1195
1196        let normalized_depth = if depth < 5 {
1197            1
1198        } else if depth < 15 {
1199            10
1200        } else {
1201            20
1202        };
1203
1204        let channel = format!(
1205            "book.{}.{}.{}.{}",
1206            instrument_id.symbol,
1207            group,
1208            normalized_depth,
1209            interval.as_str()
1210        );
1211        log::debug!("Subscribing to grouped book channel: {channel}");
1212        self.send_subscribe(vec![channel]).await
1213    }
1214
1215    /// Unsubscribes from grouped (depth-limited) order book updates for an instrument.
1216    ///
1217    /// Depth is normalized to Deribit supported values: 1, 10, or 20.
1218    ///
1219    /// # Errors
1220    ///
1221    /// Returns an error if unsubscription fails.
1222    pub async fn unsubscribe_book_grouped(
1223        &self,
1224        instrument_id: InstrumentId,
1225        group: &str,
1226        depth: u32,
1227        interval: Option<DeribitUpdateInterval>,
1228    ) -> DeribitWsResult<()> {
1229        // Grouped book channel only supports 100ms and agg2, not raw
1230        let interval = match interval {
1231            Some(DeribitUpdateInterval::Raw) | None => DeribitUpdateInterval::Ms100,
1232            Some(i) => i,
1233        };
1234
1235        let normalized_depth = if depth < 5 {
1236            1
1237        } else if depth < 15 {
1238            10
1239        } else {
1240            20
1241        };
1242
1243        let channel = format!(
1244            "book.{}.{}.{}.{}",
1245            instrument_id.symbol,
1246            group,
1247            normalized_depth,
1248            interval.as_str()
1249        );
1250        self.send_unsubscribe(vec![channel]).await
1251    }
1252
1253    /// Subscribes to ticker updates for an instrument.
1254    ///
1255    /// # Arguments
1256    ///
1257    /// * `instrument_id` - The instrument to subscribe to
1258    /// * `interval` - Update interval. Defaults to `Ms100` (100ms). `Raw` requires authentication.
1259    ///
1260    /// # Errors
1261    ///
1262    /// Returns an error if subscription fails or raw is requested without authentication.
1263    pub async fn subscribe_ticker(
1264        &self,
1265        instrument_id: InstrumentId,
1266        interval: Option<DeribitUpdateInterval>,
1267    ) -> DeribitWsResult<()> {
1268        let interval = interval.unwrap_or_default();
1269        self.check_auth_requirement(interval)?;
1270        let channel =
1271            DeribitWsChannel::Ticker.format_channel(instrument_id.symbol.as_str(), Some(interval));
1272        self.send_subscribe(vec![channel]).await
1273    }
1274
1275    /// Unsubscribes from ticker updates for an instrument.
1276    ///
1277    /// # Errors
1278    ///
1279    /// Returns an error if unsubscription fails.
1280    pub async fn unsubscribe_ticker(
1281        &self,
1282        instrument_id: InstrumentId,
1283        interval: Option<DeribitUpdateInterval>,
1284    ) -> DeribitWsResult<()> {
1285        let interval = interval.unwrap_or_default();
1286        let channel =
1287            DeribitWsChannel::Ticker.format_channel(instrument_id.symbol.as_str(), Some(interval));
1288        self.send_unsubscribe(vec![channel]).await
1289    }
1290
1291    /// Subscribes to quote (best bid/ask) updates for an instrument.
1292    ///
1293    /// Note: Quote channel does not support interval parameter.
1294    ///
1295    /// # Errors
1296    ///
1297    /// Returns an error if subscription fails.
1298    pub async fn subscribe_quotes(&self, instrument_id: InstrumentId) -> DeribitWsResult<()> {
1299        let channel = DeribitWsChannel::Quote.format_channel(instrument_id.symbol.as_str(), None);
1300        self.send_subscribe(vec![channel]).await
1301    }
1302
1303    /// Unsubscribes from quote updates for an instrument.
1304    ///
1305    /// # Errors
1306    ///
1307    /// Returns an error if unsubscription fails.
1308    pub async fn unsubscribe_quotes(&self, instrument_id: InstrumentId) -> DeribitWsResult<()> {
1309        let channel = DeribitWsChannel::Quote.format_channel(instrument_id.symbol.as_str(), None);
1310        self.send_unsubscribe(vec![channel]).await
1311    }
1312
1313    /// Subscribes to instrument status changes for lifecycle notifications.
1314    ///
1315    /// Channel format: `instrument.state.{kind}.{currency}`
1316    ///
1317    /// # Errors
1318    ///
1319    /// Returns an error if subscription fails.
1320    pub async fn subscribe_instrument_status(
1321        &self,
1322        kind: &str,
1323        currency: &str,
1324    ) -> DeribitWsResult<()> {
1325        let channel = DeribitWsChannel::format_instrument_state_channel(kind, currency);
1326        self.send_subscribe(vec![channel]).await
1327    }
1328
1329    /// Unsubscribes from instrument status changes.
1330    ///
1331    /// # Errors
1332    ///
1333    /// Returns an error if unsubscription fails.
1334    pub async fn unsubscribe_instrument_status(
1335        &self,
1336        kind: &str,
1337        currency: &str,
1338    ) -> DeribitWsResult<()> {
1339        let channel = DeribitWsChannel::format_instrument_state_channel(kind, currency);
1340        self.send_unsubscribe(vec![channel]).await
1341    }
1342
1343    /// Subscribes to volatility index updates for the given index name.
1344    ///
1345    /// Channel format: `deribit_volatility_index.{index_name}`
1346    ///
1347    /// # Errors
1348    ///
1349    /// Returns an error if subscription fails.
1350    pub async fn subscribe_volatility_index(&self, index_name: &str) -> DeribitWsResult<()> {
1351        let channel = DeribitWsChannel::VolatilityIndex.format_channel(index_name, None);
1352        self.send_subscribe(vec![channel]).await
1353    }
1354
1355    /// Unsubscribes from volatility index updates for the given index name.
1356    ///
1357    /// # Errors
1358    ///
1359    /// Returns an error if unsubscription fails.
1360    pub async fn unsubscribe_volatility_index(&self, index_name: &str) -> DeribitWsResult<()> {
1361        let channel = DeribitWsChannel::VolatilityIndex.format_channel(index_name, None);
1362        self.send_unsubscribe(vec![channel]).await
1363    }
1364
1365    /// Subscribes to perpetual interest rates updates.
1366    ///
1367    /// Channel format: `perpetual.{instrument_name}.{interval}`
1368    ///
1369    /// # Errors
1370    ///
1371    /// Returns an error if subscription fails.
1372    pub async fn subscribe_perpetual_interests_rates_updates(
1373        &self,
1374        instrument_id: InstrumentId,
1375        interval: Option<DeribitUpdateInterval>,
1376    ) -> DeribitWsResult<()> {
1377        let interval = interval.unwrap_or(DeribitUpdateInterval::Ms100);
1378        let channel = DeribitWsChannel::Perpetual
1379            .format_channel(instrument_id.symbol.as_str(), Some(interval));
1380
1381        self.send_subscribe(vec![channel]).await
1382    }
1383
1384    /// Unsubscribes from perpetual interest rates updates.
1385    ///
1386    /// # Errors
1387    ///
1388    /// Returns an error if subscription fails.
1389    pub async fn unsubscribe_perpetual_interest_rates_updates(
1390        &self,
1391        instrument_id: InstrumentId,
1392        interval: Option<DeribitUpdateInterval>,
1393    ) -> DeribitWsResult<()> {
1394        let interval = interval.unwrap_or(DeribitUpdateInterval::Ms100);
1395        let channel = DeribitWsChannel::Perpetual
1396            .format_channel(instrument_id.symbol.as_str(), Some(interval));
1397
1398        self.send_unsubscribe(vec![channel]).await
1399    }
1400
1401    /// Subscribes to chart/OHLC bar updates for an instrument.
1402    ///
1403    /// # Arguments
1404    ///
1405    /// * `instrument_id` - The instrument to subscribe to
1406    /// * `resolution` - Bar resolution: "1", "3", "5", "10", "15", "30", "60", "120", "180",
1407    ///   "360", "720", "1D" (minutes or 1D for daily)
1408    ///
1409    /// # Errors
1410    ///
1411    /// Returns an error if subscription fails.
1412    pub async fn subscribe_chart(
1413        &self,
1414        instrument_id: InstrumentId,
1415        resolution: &str,
1416    ) -> DeribitWsResult<()> {
1417        // Chart channel format: chart.trades.{instrument}.{resolution}
1418        let channel = format!("chart.trades.{}.{}", instrument_id.symbol, resolution);
1419        self.send_subscribe(vec![channel]).await
1420    }
1421
1422    /// Unsubscribes from chart/OHLC bar updates.
1423    ///
1424    /// # Errors
1425    ///
1426    /// Returns an error if unsubscription fails.
1427    pub async fn unsubscribe_chart(
1428        &self,
1429        instrument_id: InstrumentId,
1430        resolution: &str,
1431    ) -> DeribitWsResult<()> {
1432        let channel = format!("chart.trades.{}.{}", instrument_id.symbol, resolution);
1433        self.send_unsubscribe(vec![channel]).await
1434    }
1435
1436    /// Subscribes to bar updates for an instrument using a BarType specification.
1437    ///
1438    /// Converts the BarType to the nearest supported Deribit resolution and subscribes
1439    /// to the chart channel.
1440    ///
1441    /// # Errors
1442    ///
1443    /// Returns an error if the subscription request fails.
1444    pub async fn subscribe_bars(&self, bar_type: BarType) -> DeribitWsResult<()> {
1445        let resolution = bar_spec_to_resolution(&bar_type);
1446        self.subscribe_chart(bar_type.instrument_id(), &resolution)
1447            .await
1448    }
1449
1450    /// Unsubscribes from bar updates for an instrument using a BarType specification.
1451    ///
1452    /// # Errors
1453    ///
1454    /// Returns an error if the unsubscription request fails.
1455    pub async fn unsubscribe_bars(&self, bar_type: BarType) -> DeribitWsResult<()> {
1456        let resolution = bar_spec_to_resolution(&bar_type);
1457        self.unsubscribe_chart(bar_type.instrument_id(), &resolution)
1458            .await
1459    }
1460
1461    /// Checks if authentication is required for the given interval.
1462    ///
1463    /// # Errors
1464    ///
1465    /// Returns an error if raw interval is requested but client is not authenticated.
1466    fn check_auth_requirement(&self, interval: DeribitUpdateInterval) -> DeribitWsResult<()> {
1467        if interval.requires_auth() && !self.is_authenticated() {
1468            return Err(DeribitWsError::Authentication(
1469                "Raw streams require authentication. Call authenticate() first.".to_string(),
1470            ));
1471        }
1472        Ok(())
1473    }
1474
1475    /// Subscribes to user order updates for all instruments.
1476    ///
1477    /// Requires authentication. Subscribes to `user.orders.any.any.raw` channel.
1478    ///
1479    /// # Errors
1480    ///
1481    /// Returns an error if client is not authenticated or subscription fails.
1482    pub async fn subscribe_user_orders(&self) -> DeribitWsResult<()> {
1483        if !self.is_authenticated() {
1484            return Err(DeribitWsError::Authentication(
1485                "User orders subscription requires authentication".to_string(),
1486            ));
1487        }
1488        self.send_subscribe(vec!["user.orders.any.any.raw".to_string()])
1489            .await
1490    }
1491
1492    /// Unsubscribes from user order updates for all instruments.
1493    ///
1494    /// # Errors
1495    ///
1496    /// Returns an error if unsubscription fails.
1497    pub async fn unsubscribe_user_orders(&self) -> DeribitWsResult<()> {
1498        self.send_unsubscribe(vec!["user.orders.any.any.raw".to_string()])
1499            .await
1500    }
1501
1502    /// Subscribes to user trade/fill updates for all instruments.
1503    ///
1504    /// Requires authentication. Subscribes to `user.trades.any.any.raw` channel.
1505    ///
1506    /// # Errors
1507    ///
1508    /// Returns an error if client is not authenticated or subscription fails.
1509    pub async fn subscribe_user_trades(&self) -> DeribitWsResult<()> {
1510        if !self.is_authenticated() {
1511            return Err(DeribitWsError::Authentication(
1512                "User trades subscription requires authentication".to_string(),
1513            ));
1514        }
1515        self.send_subscribe(vec!["user.trades.any.any.raw".to_string()])
1516            .await
1517    }
1518
1519    /// Unsubscribes from user trade/fill updates for all instruments.
1520    ///
1521    /// # Errors
1522    ///
1523    /// Returns an error if unsubscription fails.
1524    pub async fn unsubscribe_user_trades(&self) -> DeribitWsResult<()> {
1525        self.send_unsubscribe(vec!["user.trades.any.any.raw".to_string()])
1526            .await
1527    }
1528
1529    /// Subscribes to user portfolio updates for all currencies.
1530    ///
1531    /// Requires authentication. Subscribes to `user.portfolio.any` channel which
1532    /// provides real-time account balance and margin updates for all currencies
1533    /// (BTC, ETH, USDC, USDT, etc.).
1534    ///
1535    /// # Errors
1536    ///
1537    /// Returns an error if client is not authenticated or subscription fails.
1538    pub async fn subscribe_user_portfolio(&self) -> DeribitWsResult<()> {
1539        if !self.is_authenticated() {
1540            return Err(DeribitWsError::Authentication(
1541                "User portfolio subscription requires authentication".to_string(),
1542            ));
1543        }
1544        self.send_subscribe(vec!["user.portfolio.any".to_string()])
1545            .await
1546    }
1547
1548    /// Unsubscribes from user portfolio updates for all currencies.
1549    ///
1550    /// # Errors
1551    ///
1552    /// Returns an error if unsubscription fails.
1553    pub async fn unsubscribe_user_portfolio(&self) -> DeribitWsResult<()> {
1554        self.send_unsubscribe(vec!["user.portfolio.any".to_string()])
1555            .await
1556    }
1557
1558    /// Subscribes to multiple channels at once.
1559    ///
1560    /// # Errors
1561    ///
1562    /// Returns an error if subscription fails.
1563    pub async fn subscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1564        self.send_subscribe(channels).await
1565    }
1566
1567    /// Unsubscribes from multiple channels at once.
1568    ///
1569    /// # Errors
1570    ///
1571    /// Returns an error if unsubscription fails.
1572    pub async fn unsubscribe(&self, channels: Vec<String>) -> DeribitWsResult<()> {
1573        self.send_unsubscribe(channels).await
1574    }
1575
1576    /// Submits an order to Deribit via WebSocket.
1577    ///
1578    /// Routes to `private/buy` or `private/sell` JSON-RPC method based on order side.
1579    /// Requires authentication (call `authenticate_session()` first).
1580    ///
1581    /// # Errors
1582    ///
1583    /// Returns an error if:
1584    /// - The client is not authenticated
1585    /// - The command fails to send
1586    pub async fn submit_order(
1587        &self,
1588        order_side: OrderSide,
1589        params: DeribitOrderParams,
1590        client_order_id: ClientOrderId,
1591        trader_id: TraderId,
1592        strategy_id: StrategyId,
1593        instrument_id: InstrumentId,
1594    ) -> DeribitWsResult<()> {
1595        if !self.is_authenticated() {
1596            return Err(DeribitWsError::Authentication(
1597                "Submit order requires authentication. Call authenticate_session() first."
1598                    .to_string(),
1599            ));
1600        }
1601
1602        log::debug!(
1603            "Sending {} order: instrument={}, amount={}, price={:?}, client_order_id={}",
1604            order_side,
1605            params.instrument_name,
1606            params.amount,
1607            params.price,
1608            client_order_id
1609        );
1610
1611        let cmd = match order_side {
1612            OrderSide::Buy => HandlerCommand::Buy {
1613                params,
1614                client_order_id,
1615                trader_id,
1616                strategy_id,
1617                instrument_id,
1618            },
1619            OrderSide::Sell => HandlerCommand::Sell {
1620                params,
1621                client_order_id,
1622                trader_id,
1623                strategy_id,
1624                instrument_id,
1625            },
1626        };
1627
1628        self.command_sender()
1629            .send(cmd)
1630            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1631
1632        Ok(())
1633    }
1634
1635    /// Modifies an existing order on Deribit via WebSocket.
1636    ///
1637    /// The order parameters are sent using the `private/edit` JSON-RPC method.
1638    /// Requires authentication (call `authenticate_session()` first).
1639    ///
1640    /// # Errors
1641    ///
1642    /// Returns an error if:
1643    /// - The client is not authenticated
1644    /// - The command fails to send
1645    #[expect(clippy::too_many_arguments)]
1646    pub async fn modify_order(
1647        &self,
1648        order_id: &str,
1649        quantity: Quantity,
1650        price: Price,
1651        client_order_id: ClientOrderId,
1652        trader_id: TraderId,
1653        strategy_id: StrategyId,
1654        instrument_id: InstrumentId,
1655    ) -> DeribitWsResult<()> {
1656        if !self.is_authenticated() {
1657            return Err(DeribitWsError::Authentication(
1658                "Modify order requires authentication. Call authenticate_session() first."
1659                    .to_string(),
1660            ));
1661        }
1662
1663        let params = DeribitEditParams {
1664            order_id: order_id.to_string(),
1665            amount: quantity.as_decimal(),
1666            price: Some(price.as_decimal()),
1667            post_only: None,
1668            reject_post_only: None,
1669            reduce_only: None,
1670            trigger_price: None,
1671        };
1672
1673        log::debug!(
1674            "Sending modify order: order_id={order_id}, quantity={quantity}, price={price}, client_order_id={client_order_id}"
1675        );
1676
1677        self.command_sender()
1678            .send(HandlerCommand::Edit {
1679                params,
1680                client_order_id,
1681                trader_id,
1682                strategy_id,
1683                instrument_id,
1684            })
1685            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1686
1687        Ok(())
1688    }
1689
1690    /// Cancels an existing order on Deribit via WebSocket.
1691    ///
1692    /// The order is cancelled using the `private/cancel` JSON-RPC method.
1693    /// Requires authentication (call `authenticate_session()` first).
1694    ///
1695    /// # Errors
1696    ///
1697    /// Returns an error if:
1698    /// - The client is not authenticated
1699    /// - The command fails to send
1700    pub async fn cancel_order(
1701        &self,
1702        order_id: &str,
1703        client_order_id: ClientOrderId,
1704        trader_id: TraderId,
1705        strategy_id: StrategyId,
1706        instrument_id: InstrumentId,
1707    ) -> DeribitWsResult<()> {
1708        if !self.is_authenticated() {
1709            return Err(DeribitWsError::Authentication(
1710                "Cancel order requires authentication. Call authenticate_session() first."
1711                    .to_string(),
1712            ));
1713        }
1714
1715        let params = DeribitCancelParams {
1716            order_id: order_id.to_string(),
1717        };
1718
1719        log::debug!("Sending cancel order: order_id={order_id}, client_order_id={client_order_id}");
1720
1721        self.command_sender()
1722            .send(HandlerCommand::Cancel {
1723                params,
1724                client_order_id,
1725                trader_id,
1726                strategy_id,
1727                instrument_id,
1728            })
1729            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1730
1731        Ok(())
1732    }
1733
1734    /// Cancels all orders for a specific instrument on Deribit via WebSocket.
1735    ///
1736    /// Uses the `private/cancel_all_by_instrument` JSON-RPC method.
1737    /// Requires authentication (call `authenticate_session()` first).
1738    ///
1739    /// # Errors
1740    ///
1741    /// Returns an error if:
1742    /// - The client is not authenticated
1743    /// - The command fails to send
1744    pub async fn cancel_all_orders(
1745        &self,
1746        instrument_id: InstrumentId,
1747        order_type: Option<String>,
1748    ) -> DeribitWsResult<()> {
1749        if !self.is_authenticated() {
1750            return Err(DeribitWsError::Authentication(
1751                "Cancel all orders requires authentication. Call authenticate_session() first."
1752                    .to_string(),
1753            ));
1754        }
1755
1756        let instrument_name = instrument_id.symbol.to_string();
1757        let params = DeribitCancelAllByInstrumentParams {
1758            instrument_name: instrument_name.clone(),
1759            order_type,
1760        };
1761
1762        log::debug!("Sending cancel_all_orders: instrument={instrument_name}");
1763
1764        self.command_sender()
1765            .send(HandlerCommand::CancelAllByInstrument {
1766                params,
1767                instrument_id,
1768            })
1769            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1770
1771        Ok(())
1772    }
1773
1774    /// Queries the state of an order on Deribit via WebSocket.
1775    ///
1776    /// Uses the `private/get_order_state` JSON-RPC method.
1777    /// Requires authentication (call `authenticate_session()` first).
1778    ///
1779    /// # Errors
1780    ///
1781    /// Returns an error if:
1782    /// - The client is not authenticated
1783    /// - The command fails to send
1784    pub async fn query_order(
1785        &self,
1786        order_id: &str,
1787        client_order_id: ClientOrderId,
1788        trader_id: TraderId,
1789        strategy_id: StrategyId,
1790        instrument_id: InstrumentId,
1791    ) -> DeribitWsResult<()> {
1792        if !self.is_authenticated() {
1793            return Err(DeribitWsError::Authentication(
1794                "Query order state requires authentication. Call authenticate_session() first."
1795                    .to_string(),
1796            ));
1797        }
1798
1799        log::debug!("Sending query_order: order_id={order_id}, client_order_id={client_order_id}");
1800
1801        self.command_sender()
1802            .send(HandlerCommand::GetOrderState {
1803                order_id: order_id.to_string(),
1804                client_order_id,
1805                trader_id,
1806                strategy_id,
1807                instrument_id,
1808            })
1809            .map_err(|e| DeribitWsError::Send(e.to_string()))?;
1810
1811        Ok(())
1812    }
1813
1814    fn command_sender(&self) -> RwLockReadGuard<'_, CommandSender> {
1815        self.cmd_tx.read()
1816    }
1817}
1818
1819#[cfg(test)]
1820mod tests {
1821    use rstest::rstest;
1822
1823    use super::*;
1824
1825    struct DropSignal(Arc<AtomicBool>);
1826
1827    impl Drop for DropSignal {
1828        fn drop(&mut self) {
1829            self.0.store(true, Ordering::Release);
1830        }
1831    }
1832
1833    #[tokio::test]
1834    async fn test_last_client_owner_drop_aborts_handler_task() {
1835        let client = DeribitWebSocketClient::new_unauthenticated(
1836            Some("ws://127.0.0.1:0/ws/api/v2".to_string()),
1837            30,
1838            DeribitEnvironment::Testnet,
1839        )
1840        .unwrap();
1841        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
1842        let dropped = Arc::new(AtomicBool::new(false));
1843        let drop_signal = DropSignal(Arc::clone(&dropped));
1844        client
1845            .handler_tasks
1846            .spawn(async move {
1847                let _drop_signal = drop_signal;
1848                started_tx.send(()).expect("started receiver");
1849                std::future::pending::<()>().await;
1850            })
1851            .expect("handler task should register");
1852        started_rx.await.expect("handler task started");
1853        let clone = client.clone();
1854
1855        drop(client);
1856        assert!(!dropped.load(Ordering::Acquire));
1857        drop(clone);
1858
1859        tokio::time::timeout(Duration::from_secs(1), async {
1860            while !dropped.load(Ordering::Acquire) {
1861                tokio::task::yield_now().await;
1862            }
1863        })
1864        .await
1865        .expect("handler task aborted");
1866    }
1867
1868    #[rstest]
1869    #[tokio::test]
1870    async fn test_unsubscribe_send_failure_restores_subscription() {
1871        let client = DeribitWebSocketClient::new_unauthenticated(
1872            Some("ws://127.0.0.1:0/ws/api/v2".to_string()),
1873            30,
1874            DeribitEnvironment::Testnet,
1875        )
1876        .unwrap();
1877        let channel = "trades.BTC-PERPETUAL.raw";
1878        client.subscriptions_state.add_reference(channel);
1879        client.subscriptions_state.mark_subscribe(channel);
1880        client.subscriptions_state.confirm_subscribe(channel);
1881
1882        let error = client
1883            .send_unsubscribe(vec![channel.to_string()])
1884            .await
1885            .unwrap_err();
1886
1887        assert!(matches!(error, DeribitWsError::Send(_)));
1888        assert_eq!(client.subscriptions_state.get_reference_count(channel), 1);
1889        assert_eq!(client.subscriptions_state.len(), 1);
1890        assert_eq!(client.subscriptions_state.all_topics(), [channel]);
1891        assert!(
1892            client
1893                .subscriptions_state
1894                .pending_subscribe_topics()
1895                .is_empty()
1896        );
1897        assert!(
1898            client
1899                .subscriptions_state
1900                .pending_unsubscribe_topics()
1901                .is_empty()
1902        );
1903    }
1904}
nautilus_deribit/websocket/client.rs

nautilus_deribit/websocket/
client.rs
