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

nautilus_deribit/websocket/
client.rs
