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

nautilus_deribit/websocket/
client.rs
