nautilus_bybit/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//! Bybit WebSocket client providing public market data streaming.
17//!
18//! Bybit API reference <https://bybit-exchange.github.io/docs/>.
19
20use std::{
21    collections::HashSet,
22    fmt::Debug,
23    num::NonZeroU32,
24    sync::{
25        Arc,
26        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
27    },
28    time::Duration,
29};
30
31use arc_swap::ArcSwap;
32#[cfg(test)]
33use nautilus_common::live::get_runtime;
34use nautilus_core::{AtomicMap, AtomicSet, UUID4, consts::NAUTILUS_USER_AGENT};
35use nautilus_live::{
36    SocketControl,
37    task::{SharedTaskSlot, TaskJoinOutcome},
38};
39use nautilus_model::{
40    data::BarType,
41    enums::{AggregationSource, OrderSide, OrderType, PriceType, TimeInForce, TriggerType},
42    identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
43    instruments::{Instrument, InstrumentAny},
44    types::{Price, Quantity},
45};
46use nautilus_network::{
47    http::USER_AGENT,
48    mode::ConnectionMode,
49    ratelimiter::{RateLimiter, clock::MonotonicClock},
50    websocket::{
51        AuthTracker, InitialConnectRetryPolicy, SubscriptionState, TransportBackend,
52        WebSocketClient, WebSocketConfig, channel_message_handler,
53    },
54};
55use serde_json::Value;
56use tokio_util::sync::CancellationToken;
57use ustr::Ustr;
58
59use crate::{
60    common::{
61        consts::{BYBIT_NAUTILUS_BROKER_ID, BYBIT_WS_TOPIC_DELIMITER},
62        credential::Credential,
63        enums::{
64            BybitBboSideType, BybitEnvironment, BybitOrderSide, BybitOrderType, BybitPositionIdx,
65            BybitProductType, BybitTimeInForce, BybitTpSlMode, BybitWsOrderRequestOp,
66            resolve_trigger_type,
67        },
68        parse::{
69            bar_spec_to_bybit_interval, extract_base_coin, extract_raw_symbol, map_time_in_force,
70            spot_leverage, spot_market_unit, trigger_direction,
71        },
72        rate_limit::{
73            BYBIT_OPTION_SUBSCRIPTION_LIMIT, BybitRateLimiter, batch_send_limit, batch_weight,
74            websocket_connection_key, websocket_connection_limiter,
75        },
76        symbol::BybitSymbol,
77        urls::{bybit_ws_private_url, bybit_ws_public_url, bybit_ws_trade_url},
78    },
79    websocket::{
80        enums::{BybitWsOperation, BybitWsPrivateChannel, BybitWsPublicChannel},
81        error::{BybitWsError, BybitWsResult},
82        handler::{BybitWsFeedHandler, BybitWsOrderCommand, HandlerCommand},
83        messages::{
84            BybitAuthRequest, BybitSubscription, BybitWsAmendOrderParams, BybitWsBatchAmendItem,
85            BybitWsBatchAmendOrderArgs, BybitWsBatchCancelItem, BybitWsBatchCancelOrderArgs,
86            BybitWsBatchPlaceItem, BybitWsBatchPlaceOrderArgs, BybitWsCancelOrderParams,
87            BybitWsMessage, BybitWsPlaceOrderParams,
88        },
89    },
90};
91
92const WEBSOCKET_AUTH_WINDOW_MS: i64 = 5_000;
93const AUTH_WAIT_TIMEOUT: Duration = Duration::from_secs(5);
94/// Legacy non-Spot batch endpoint maximum.
95pub const BATCH_PROCESSING_LIMIT: usize = 20;
96/// Public/market data WebSocket client for Bybit.
97pub struct BybitWebSocketClient {
98    url: String,
99    environment: BybitEnvironment,
100    product_type: Option<BybitProductType>,
101    credential: Option<Credential>,
102    requires_auth: bool,
103    auth_tracker: AuthTracker,
104    heartbeat: Option<u64>,
105    auth_wait_timeout: Duration,
106    connection_mode: Arc<ArcSwap<AtomicU8>>,
107    cmd_tx: Arc<tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<HandlerCommand>>>,
108    out_rx: Option<Arc<tokio::sync::mpsc::UnboundedReceiver<BybitWsMessage>>>,
109    signal: Arc<AtomicBool>,
110    task_handle: Arc<SharedTaskSlot<()>>,
111    connect_lock: Arc<tokio::sync::Mutex<()>>,
112    subscriptions: SubscriptionState,
113    subscription_guard: Arc<tokio::sync::Mutex<()>>,
114    rate_limiter: BybitRateLimiter,
115    recv_window_ms: Arc<AtomicU64>,
116    account_id: Option<AccountId>,
117    mm_level: Arc<AtomicU8>,
118    bar_types_cache: Arc<AtomicMap<String, BarType>>,
119    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
120    trade_subs: Arc<AtomicSet<InstrumentId>>,
121    option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
122    bars_timestamp_on_close: Arc<AtomicBool>,
123    transport_backend: TransportBackend,
124    cancellation_token: Arc<ArcSwap<CancellationToken>>,
125    proxy_url: Option<String>,
126    socket_control: Option<SocketControl>,
127}
128
129struct ConnectRollback {
130    signal: Arc<AtomicBool>,
131    cancellation_token: Arc<ArcSwap<CancellationToken>>,
132    task_handle: Arc<SharedTaskSlot<()>>,
133    armed: bool,
134}
135
136impl ConnectRollback {
137    fn new(client: &BybitWebSocketClient) -> Self {
138        Self {
139            signal: Arc::clone(&client.signal),
140            cancellation_token: Arc::clone(&client.cancellation_token),
141            task_handle: Arc::clone(&client.task_handle),
142            armed: true,
143        }
144    }
145
146    fn disarm(&mut self) {
147        self.armed = false;
148    }
149}
150
151impl Drop for ConnectRollback {
152    fn drop(&mut self) {
153        if self.armed {
154            self.signal.store(true, Ordering::Release);
155            self.cancellation_token.load().cancel();
156            self.task_handle.abort();
157        }
158    }
159}
160
161impl Debug for BybitWebSocketClient {
162    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163        f.debug_struct(stringify!(BybitWebSocketClient))
164            .field("url", &self.url)
165            .field("environment", &self.environment)
166            .field("product_type", &self.product_type)
167            .field("requires_auth", &self.requires_auth)
168            .field("heartbeat", &self.heartbeat)
169            .field("confirmed_subscriptions", &self.subscriptions.len())
170            .finish()
171    }
172}
173
174impl Clone for BybitWebSocketClient {
175    fn clone(&self) -> Self {
176        Self {
177            url: self.url.clone(),
178            environment: self.environment,
179            product_type: self.product_type,
180            credential: self.credential.clone(),
181            requires_auth: self.requires_auth,
182            auth_tracker: self.auth_tracker.clone(),
183            heartbeat: self.heartbeat,
184            auth_wait_timeout: self.auth_wait_timeout,
185            connection_mode: Arc::clone(&self.connection_mode),
186            cmd_tx: Arc::clone(&self.cmd_tx),
187            out_rx: None, // Each clone gets its own receiver
188            signal: Arc::clone(&self.signal),
189            task_handle: Arc::clone(&self.task_handle),
190            connect_lock: Arc::clone(&self.connect_lock),
191            subscriptions: self.subscriptions.clone(),
192            subscription_guard: Arc::clone(&self.subscription_guard),
193            rate_limiter: self.rate_limiter.clone(),
194            recv_window_ms: Arc::clone(&self.recv_window_ms),
195            account_id: self.account_id,
196            mm_level: Arc::clone(&self.mm_level),
197            bar_types_cache: Arc::clone(&self.bar_types_cache),
198            instruments_cache: Arc::clone(&self.instruments_cache),
199            trade_subs: Arc::clone(&self.trade_subs),
200            option_greeks_subs: Arc::clone(&self.option_greeks_subs),
201            bars_timestamp_on_close: Arc::clone(&self.bars_timestamp_on_close),
202            transport_backend: self.transport_backend,
203            cancellation_token: Arc::clone(&self.cancellation_token),
204            proxy_url: self.proxy_url.clone(),
205            socket_control: self.socket_control.clone(),
206        }
207    }
208}
209
210impl BybitWebSocketClient {
211    fn initial_connect_retry_policy() -> InitialConnectRetryPolicy {
212        InitialConnectRetryPolicy {
213            max_attempts: NonZeroU32::new(5).expect("initial connect attempts must be non-zero"),
214            delay_initial: Duration::from_millis(500),
215            delay_max: Duration::from_secs(5),
216            backoff_factor: 2.0,
217            jitter_ms: 250,
218        }
219    }
220
221    /// Creates a new Bybit public WebSocket client.
222    #[must_use]
223    pub fn new_public(url: Option<String>, heartbeat: u64) -> Self {
224        Self::new_public_with(
225            BybitProductType::Linear,
226            BybitEnvironment::Mainnet,
227            url,
228            heartbeat,
229            TransportBackend::default(),
230            None,
231        )
232    }
233
234    /// Sets the timeout for waiting on (re)authentication before failing an
235    /// authenticated operation. Defaults to `AUTH_WAIT_TIMEOUT` (5s).
236    pub fn set_auth_wait_timeout(&mut self, timeout: Duration) {
237        self.auth_wait_timeout = timeout;
238    }
239
240    /// Sets the receive window sent with WebSocket trade commands.
241    pub fn set_recv_window_ms(&self, recv_window_ms: u64) {
242        self.recv_window_ms.store(recv_window_ms, Ordering::Release);
243    }
244
245    /// Creates a new Bybit public WebSocket client targeting the specified product/environment.
246    #[must_use]
247    pub fn new_public_with(
248        product_type: BybitProductType,
249        environment: BybitEnvironment,
250        url: Option<String>,
251        heartbeat: u64,
252        transport_backend: TransportBackend,
253        proxy_url: Option<String>,
254    ) -> Self {
255        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel::<HandlerCommand>();
256
257        let initial_mode = AtomicU8::new(ConnectionMode::Closed.as_u8());
258        let connection_mode = Arc::new(ArcSwap::from_pointee(initial_mode));
259        let resolved_url = url.unwrap_or_else(|| bybit_ws_public_url(product_type, environment));
260        let rate_limiter =
261            BybitRateLimiter::for_websocket(&resolved_url, None, proxy_url.as_deref());
262
263        Self {
264            url: resolved_url,
265            environment,
266            product_type: Some(product_type),
267            credential: None,
268            requires_auth: false,
269            auth_tracker: AuthTracker::new(),
270            heartbeat: Some(heartbeat),
271            auth_wait_timeout: AUTH_WAIT_TIMEOUT,
272            connection_mode,
273            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
274            out_rx: None,
275            signal: Arc::new(AtomicBool::new(false)),
276            task_handle: Arc::new(SharedTaskSlot::new()),
277            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
278            subscriptions: SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER),
279            subscription_guard: Arc::new(tokio::sync::Mutex::new(())),
280            rate_limiter,
281            recv_window_ms: Arc::new(AtomicU64::new(5_000)),
282            bar_types_cache: Arc::new(AtomicMap::new()),
283            instruments_cache: Arc::new(AtomicMap::new()),
284            trade_subs: Arc::new(AtomicSet::new()),
285            option_greeks_subs: Arc::new(AtomicSet::new()),
286            bars_timestamp_on_close: Arc::new(AtomicBool::new(true)),
287            account_id: None,
288            mm_level: Arc::new(AtomicU8::new(0)),
289            transport_backend,
290            cancellation_token: Arc::new(ArcSwap::from_pointee(CancellationToken::new())),
291            proxy_url,
292            socket_control: None,
293        }
294    }
295
296    /// Configures socket state reporting and reconnect control.
297    #[must_use]
298    pub fn with_socket_control(mut self, control: SocketControl) -> Self {
299        self.socket_control = Some(control);
300        self
301    }
302
303    /// Creates a new Bybit private WebSocket client.
304    ///
305    /// If `api_key` or `api_secret` are not provided, they will be loaded from
306    /// environment variables based on the environment:
307    /// - Demo: `BYBIT_DEMO_API_KEY`, `BYBIT_DEMO_API_SECRET`
308    /// - Testnet: `BYBIT_TESTNET_API_KEY`, `BYBIT_TESTNET_API_SECRET`
309    /// - Mainnet: `BYBIT_API_KEY`, `BYBIT_API_SECRET`
310    #[must_use]
311    pub fn new_private(
312        environment: BybitEnvironment,
313        api_key: Option<String>,
314        api_secret: Option<String>,
315        url: Option<String>,
316        heartbeat: u64,
317        transport_backend: TransportBackend,
318        proxy_url: Option<String>,
319    ) -> Self {
320        let credential = Credential::resolve(api_key, api_secret, environment);
321
322        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel::<HandlerCommand>();
323
324        let initial_mode = AtomicU8::new(ConnectionMode::Closed.as_u8());
325        let connection_mode = Arc::new(ArcSwap::from_pointee(initial_mode));
326        let resolved_url = url.unwrap_or_else(|| bybit_ws_private_url(environment).to_string());
327        let rate_limiter = BybitRateLimiter::for_websocket(
328            &resolved_url,
329            credential.as_ref().map(Credential::api_key),
330            proxy_url.as_deref(),
331        );
332
333        Self {
334            url: resolved_url,
335            environment,
336            product_type: None,
337            credential,
338            requires_auth: true,
339            auth_tracker: AuthTracker::new(),
340            heartbeat: Some(heartbeat),
341            auth_wait_timeout: AUTH_WAIT_TIMEOUT,
342            connection_mode,
343            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
344            out_rx: None,
345            signal: Arc::new(AtomicBool::new(false)),
346            task_handle: Arc::new(SharedTaskSlot::new()),
347            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
348            subscriptions: SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER),
349            subscription_guard: Arc::new(tokio::sync::Mutex::new(())),
350            rate_limiter,
351            recv_window_ms: Arc::new(AtomicU64::new(5_000)),
352            bar_types_cache: Arc::new(AtomicMap::new()),
353            instruments_cache: Arc::new(AtomicMap::new()),
354            trade_subs: Arc::new(AtomicSet::new()),
355            option_greeks_subs: Arc::new(AtomicSet::new()),
356            bars_timestamp_on_close: Arc::new(AtomicBool::new(true)),
357            account_id: None,
358            mm_level: Arc::new(AtomicU8::new(0)),
359            transport_backend,
360            cancellation_token: Arc::new(ArcSwap::from_pointee(CancellationToken::new())),
361            proxy_url,
362            socket_control: None,
363        }
364    }
365
366    /// Creates a new Bybit trade WebSocket client for order operations.
367    ///
368    /// If `api_key` or `api_secret` are not provided, they will be loaded from
369    /// environment variables based on the environment:
370    /// - Demo: `BYBIT_DEMO_API_KEY`, `BYBIT_DEMO_API_SECRET`
371    /// - Testnet: `BYBIT_TESTNET_API_KEY`, `BYBIT_TESTNET_API_SECRET`
372    /// - Mainnet: `BYBIT_API_KEY`, `BYBIT_API_SECRET`
373    #[must_use]
374    pub fn new_trade(
375        environment: BybitEnvironment,
376        api_key: Option<String>,
377        api_secret: Option<String>,
378        url: Option<String>,
379        heartbeat: u64,
380        transport_backend: TransportBackend,
381        proxy_url: Option<String>,
382    ) -> Self {
383        let credential = Credential::resolve(api_key, api_secret, environment);
384
385        let (cmd_tx, _) = tokio::sync::mpsc::unbounded_channel::<HandlerCommand>();
386
387        let initial_mode = AtomicU8::new(ConnectionMode::Closed.as_u8());
388        let connection_mode = Arc::new(ArcSwap::from_pointee(initial_mode));
389        let resolved_url = url.unwrap_or_else(|| bybit_ws_trade_url(environment).to_string());
390        let rate_limiter = BybitRateLimiter::for_websocket(
391            &resolved_url,
392            credential.as_ref().map(Credential::api_key),
393            proxy_url.as_deref(),
394        );
395
396        Self {
397            url: resolved_url,
398            environment,
399            product_type: None,
400            credential,
401            requires_auth: true,
402            auth_tracker: AuthTracker::new(),
403            heartbeat: Some(heartbeat),
404            auth_wait_timeout: AUTH_WAIT_TIMEOUT,
405            connection_mode,
406            cmd_tx: Arc::new(tokio::sync::RwLock::new(cmd_tx)),
407            out_rx: None,
408            signal: Arc::new(AtomicBool::new(false)),
409            task_handle: Arc::new(SharedTaskSlot::new()),
410            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
411            subscriptions: SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER),
412            subscription_guard: Arc::new(tokio::sync::Mutex::new(())),
413            rate_limiter,
414            recv_window_ms: Arc::new(AtomicU64::new(5_000)),
415            bar_types_cache: Arc::new(AtomicMap::new()),
416            instruments_cache: Arc::new(AtomicMap::new()),
417            trade_subs: Arc::new(AtomicSet::new()),
418            option_greeks_subs: Arc::new(AtomicSet::new()),
419            bars_timestamp_on_close: Arc::new(AtomicBool::new(true)),
420            account_id: None,
421            mm_level: Arc::new(AtomicU8::new(0)),
422            transport_backend,
423            cancellation_token: Arc::new(ArcSwap::from_pointee(CancellationToken::new())),
424            proxy_url,
425            socket_control: None,
426        }
427    }
428
429    pub(crate) fn begin_shutdown(&self) {
430        self.cancellation_token.load().cancel();
431        self.signal.store(true, Ordering::Release);
432    }
433
434    /// Establishes the WebSocket connection.
435    ///
436    /// # Errors
437    ///
438    /// Returns an error if the underlying WebSocket connection cannot be established,
439    /// after retrying multiple times with exponential backoff.
440    pub async fn connect(&mut self) -> BybitWsResult<()> {
441        let connect_lock = Arc::clone(&self.connect_lock);
442        let _guard = connect_lock.lock().await;
443        self.connect_locked().await
444    }
445
446    async fn connect_locked(&mut self) -> BybitWsResult<()> {
447        if !self.task_handle.is_empty() {
448            self.close_locked().await?;
449        }
450        self.signal.store(false, Ordering::Relaxed);
451        let cancellation_token = CancellationToken::new();
452        self.cancellation_token
453            .store(Arc::new(cancellation_token.clone()));
454
455        let (raw_handler, raw_rx) = channel_message_handler();
456
457        // Inbound Ping frames are answered by the transport, so no ping handler is needed;
458        // the reader routes them away from the message channel and the handler never sees them.
459
460        let ping_msg = serde_json::to_string(&BybitSubscription {
461            op: BybitWsOperation::Ping,
462            args: vec![],
463            req_id: None,
464        })?;
465
466        let config = WebSocketConfig {
467            url: self.url.clone(),
468            headers: Self::default_headers(),
469            heartbeat_interval_secs: self.heartbeat,
470            heartbeat_payload: Some(ping_msg),
471            connect_timeout_ms: Some(5_000),
472            reconnect_delay_initial_ms: Some(500),
473            reconnect_delay_max_ms: Some(5_000),
474            reconnect_backoff_factor: Some(1.5),
475            reconnect_jitter_ms: Some(250),
476            reconnect_max_attempts: None,
477            heartbeat_timeout_secs: None,
478            idle_timeout_ms: None,
479            backend: self.transport_backend,
480            proxy_url: self.proxy_url.clone(),
481        };
482
483        let message_rate_limiter = Arc::new(RateLimiter::<Ustr, MonotonicClock>::new_with_quota(
484            None,
485            vec![],
486        ));
487        let connection_rate_limiter =
488            websocket_connection_limiter(&self.url, self.proxy_url.as_deref());
489        let connection_rate_keys: Arc<[Ustr]> = Arc::from([websocket_connection_key()]);
490        let client = WebSocketClient::builder()
491            .config(config.clone())
492            .message_handler(raw_handler.clone())
493            .rate_limiter(Arc::clone(&message_rate_limiter))
494            .connection_rate_limiter(Arc::clone(&connection_rate_limiter))
495            .connection_rate_keys(Arc::clone(&connection_rate_keys))
496            .initial_connect_retry_policy(Self::initial_connect_retry_policy())
497            .cancellation_token(cancellation_token)
498            .maybe_state_sink(self.socket_control.as_ref().map(SocketControl::sink))
499            .connect()
500            .await
501            .map_err(|e| {
502                BybitWsError::Transport(format!(
503                    "Failed to connect to {}: {e}. \
504                    If this is a DNS error, check your network configuration and DNS settings.",
505                    self.url,
506                ))
507            })?;
508
509        self.connection_mode.store(client.connection_mode_atomic());
510        let reconnect_handle = client.reconnect_handle();
511        client.set_auth_tracker(self.auth_tracker.clone(), self.requires_auth);
512
513        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel::<BybitWsMessage>();
514        self.out_rx = Some(Arc::new(out_rx));
515
516        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel::<HandlerCommand>();
517        *self.cmd_tx.write().await = cmd_tx.clone();
518
519        let cmd = HandlerCommand::SetClient(client);
520
521        self.send_cmd(cmd).await?;
522
523        let signal = Arc::clone(&self.signal);
524        let subscriptions = self.subscriptions.clone();
525        let credential = self.credential.clone();
526        let requires_auth = self.requires_auth;
527        let cmd_tx_for_reconnect = cmd_tx.clone();
528        let auth_tracker = self.auth_tracker.clone();
529        let auth_tracker_for_handler = auth_tracker.clone();
530        let rate_limiter = self.rate_limiter.clone();
531        let recv_window_ms = Arc::clone(&self.recv_window_ms);
532        let mut rollback = ConnectRollback::new(self);
533
534        if let Err(e) = self.task_handle.spawn(async move {
535            let mut handler = BybitWsFeedHandler::new(
536                signal.clone(),
537                cmd_rx,
538                raw_rx,
539                auth_tracker_for_handler,
540                subscriptions.clone(),
541                rate_limiter,
542                recv_window_ms,
543            );
544
545            // Helper closure to resubscribe all tracked subscriptions after reconnection
546            let resubscribe_all = || async {
547                let topics = subscriptions.all_topics();
548
549                if topics.is_empty() {
550                    return;
551                }
552
553                log::debug!(
554                    "Resubscribing to confirmed subscriptions: count={}",
555                    topics.len()
556                );
557
558                for topic in &topics {
559                    subscriptions.mark_subscribe(topic.as_str());
560                }
561
562                let mut payloads = Vec::with_capacity(topics.len());
563                for topic in &topics {
564                    let message = BybitSubscription {
565                        op: BybitWsOperation::Subscribe,
566                        args: vec![topic.clone()],
567                        req_id: Some(topic.clone()),
568                    };
569
570                    if let Ok(payload) = serde_json::to_string(&message) {
571                        payloads.push(payload);
572                    }
573                }
574
575                let cmd = HandlerCommand::Subscribe { topics: payloads };
576
577                if let Err(e) = cmd_tx_for_reconnect.send(cmd) {
578                    log::error!("Failed to send resubscribe command: {e}");
579                }
580            };
581
582            // Run message processing with reconnection handling
583            loop {
584                match handler.next().await {
585                    Some(BybitWsMessage::Reconnected) => {
586                        if signal.load(Ordering::Relaxed) {
587                            continue;
588                        }
589
590                        log::info!("WebSocket reconnected");
591
592                        subscriptions.reset_after_reconnect();
593
594                        if requires_auth {
595                            log::debug!("Re-authenticating after reconnection");
596
597                            if let Some(cred) = &credential {
598                                // Begin auth attempt so succeed() will update state
599                                let _rx = auth_tracker.begin();
600
601                                let expires = jiff::Timestamp::now().as_millisecond()
602                                    + WEBSOCKET_AUTH_WINDOW_MS;
603                                let signature = cred.sign_websocket_auth(expires);
604
605                                let auth_message = BybitAuthRequest {
606                                    op: BybitWsOperation::Auth,
607                                    args: vec![
608                                        Value::String(cred.api_key().to_string()),
609                                        Value::Number(expires.into()),
610                                        Value::String(signature),
611                                    ],
612                                };
613
614                                if let Ok(payload) = serde_json::to_string(&auth_message) {
615                                    let cmd = HandlerCommand::Authenticate { payload };
616                                    if let Err(e) = cmd_tx_for_reconnect.send(cmd) {
617                                        log::error!(
618                                            "Failed to send reconnection auth command: error={e}"
619                                        );
620                                    }
621                                } else {
622                                    log::error!("Failed to serialize reconnection auth message");
623                                }
624                            }
625                        }
626
627                        // Unauthenticated sessions resubscribe immediately after reconnection,
628                        // authenticated sessions wait for Auth message
629                        if !requires_auth {
630                            log::debug!("No authentication required, resubscribing immediately");
631                            resubscribe_all().await;
632                        }
633
634                        // Forward to out_tx so caller sees the Reconnected message
635                        if out_tx.send(BybitWsMessage::Reconnected).is_err() {
636                            if handler.is_stopped() {
637                                log::debug!("Receiver dropped, stopping");
638                            } else {
639                                log::error!("Receiver dropped, stopping");
640                            }
641                            break;
642                        }
643                    }
644                    Some(BybitWsMessage::Auth(ref auth)) => {
645                        let is_success = auth.success.unwrap_or(false) || auth.ret_code == Some(0);
646                        if is_success {
647                            log::debug!("Authenticated, resubscribing");
648                            resubscribe_all().await;
649                        }
650
651                        if out_tx.send(BybitWsMessage::Auth(auth.clone())).is_err() {
652                            if handler.is_stopped() {
653                                log::debug!("Failed to send message (receiver dropped)");
654                            } else {
655                                log::error!("Failed to send message (receiver dropped)");
656                            }
657                            break;
658                        }
659                    }
660                    Some(msg) => {
661                        if out_tx.send(msg).is_err() {
662                            if handler.is_stopped() {
663                                log::debug!("Failed to send message (receiver dropped)");
664                            } else {
665                                log::error!("Failed to send message (receiver dropped)");
666                            }
667                            break;
668                        }
669                    }
670                    None => {
671                        // Stream ended - check if it's a stop signal
672                        if handler.is_stopped() {
673                            log::debug!("Stop signal received, ending message processing");
674                            break;
675                        }
676                        // Otherwise it's an unexpected stream end
677                        log::warn!("WebSocket stream ended unexpectedly");
678                        break;
679                    }
680                }
681            }
682
683            log::debug!("Handler task exiting");
684        }) {
685            let shutdown_result = self.close_locked().await;
686            return Err(BybitWsError::ClientError(match shutdown_result {
687                Ok(()) => format!("Failed to start WebSocket handler task: {e}"),
688                Err(shutdown_error) => format!(
689                    "Failed to start WebSocket handler task: {e}; startup rollback failed: \
690                     {shutdown_error}"
691                ),
692            }));
693        }
694
695        if let Some(control) = &self.socket_control {
696            control.register(move || reconnect_handle.request_reconnect());
697        }
698
699        if requires_auth && let Err(e) = self.authenticate_if_required().await {
700            let result = match self.close_locked().await {
701                Ok(()) => Err(e),
702                Err(shutdown_error) => Err(BybitWsError::ClientError(format!(
703                    "{e}; startup rollback failed: {shutdown_error}"
704                ))),
705            };
706            rollback.disarm();
707            return result;
708        }
709
710        rollback.disarm();
711        Ok(())
712    }
713
714    /// Disconnects the WebSocket client and stops the background task.
715    pub async fn close(&mut self) -> BybitWsResult<()> {
716        let connect_lock = Arc::clone(&self.connect_lock);
717        let _guard = connect_lock.lock().await;
718        self.close_locked().await
719    }
720
721    async fn close_locked(&self) -> BybitWsResult<()> {
722        log::debug!("Starting close process");
723
724        self.signal.store(true, Ordering::Relaxed);
725        self.cancellation_token.load().cancel();
726
727        let cmd = HandlerCommand::Disconnect;
728        if let Err(e) = self.cmd_tx.read().await.send(cmd) {
729            log::debug!(
730                "Failed to send disconnect command (handler may already be shut down): {e}"
731            );
732        }
733
734        let task_result = if self.task_handle.is_empty() {
735            log::debug!("No task handle to await");
736            Ok(())
737        } else {
738            log::debug!("Waiting for task handle to complete");
739
740            if let Some(outcome) = self
741                .task_handle
742                .finish(Duration::from_secs(2), Duration::from_secs(2))
743                .await
744            {
745                match outcome {
746                    TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted => Ok(()),
747                    TaskJoinOutcome::Failed(error) => Err(BybitWsError::ClientError(format!(
748                        "WebSocket handler task failed: {error}"
749                    ))),
750                    TaskJoinOutcome::Incomplete => Err(BybitWsError::ClientError(
751                        "WebSocket handler task did not stop after abort".to_string(),
752                    )),
753                }
754            } else {
755                Ok(())
756            }
757        };
758
759        self.auth_tracker.invalidate();
760
761        if let Some(control) = &self.socket_control {
762            control.deregister();
763        }
764
765        log::debug!("Closed");
766
767        task_result
768    }
769
770    /// Returns a value indicating whether the client is active.
771    #[must_use]
772    pub fn is_active(&self) -> bool {
773        let connection_mode_arc = self.connection_mode.load();
774        ConnectionMode::from_atomic(&connection_mode_arc).is_active()
775            && !self.signal.load(Ordering::Relaxed)
776    }
777
778    /// Returns a value indicating whether the client is closed.
779    pub fn is_closed(&self) -> bool {
780        let connection_mode_arc = self.connection_mode.load();
781        ConnectionMode::from_atomic(&connection_mode_arc).is_closed()
782            || self.signal.load(Ordering::Relaxed)
783    }
784
785    /// Waits until the WebSocket client becomes active or times out.
786    ///
787    /// # Errors
788    ///
789    /// Returns an error if the timeout is exceeded before the client becomes active.
790    pub async fn wait_until_active(&self, timeout_secs: f64) -> BybitWsResult<()> {
791        let timeout = tokio::time::Duration::from_secs_f64(timeout_secs);
792
793        tokio::time::timeout(timeout, async {
794            while !self.is_active() {
795                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
796            }
797        })
798        .await
799        .map_err(|_| {
800            BybitWsError::ClientError(format!(
801                "WebSocket connection timeout after {timeout_secs} seconds"
802            ))
803        })?;
804
805        Ok(())
806    }
807
808    /// Subscribe to the provided topic strings.
809    pub async fn subscribe(&self, topics: Vec<String>) -> BybitWsResult<()> {
810        if topics.is_empty() {
811            return Ok(());
812        }
813        let _guard = self.subscription_guard.lock().await;
814
815        if self.product_type == Some(BybitProductType::Option) {
816            let occupied_topics = self
817                .subscriptions
818                .all_topics()
819                .into_iter()
820                .chain(self.subscriptions.pending_unsubscribe_topics())
821                .collect::<HashSet<_>>();
822            let new_topics = topics
823                .iter()
824                .filter(|topic| !occupied_topics.contains(topic.as_str()))
825                .collect::<HashSet<_>>()
826                .len();
827            let requested = occupied_topics.len() + new_topics;
828            if requested > BYBIT_OPTION_SUBSCRIPTION_LIMIT {
829                return Err(BybitWsError::ClientError(format!(
830                    "Option WebSocket subscription limit is {BYBIT_OPTION_SUBSCRIPTION_LIMIT} arguments per connection, requested {requested}"
831                )));
832            }
833        }
834
835        log::debug!("Subscribing to topics: {topics:?}");
836
837        // Use reference counting to deduplicate subscriptions
838        let mut topics_to_send = Vec::new();
839
840        for topic in topics {
841            // Returns true if this is the first subscription (ref count 0 -> 1)
842            if self.subscriptions.add_reference(&topic) {
843                self.subscriptions.mark_subscribe(&topic);
844                topics_to_send.push(topic.clone());
845            } else {
846                log::debug!("Already subscribed to {topic}, skipping duplicate subscription");
847            }
848        }
849
850        if topics_to_send.is_empty() {
851            return Ok(());
852        }
853
854        // Serialize subscription messages
855        let mut payloads = Vec::with_capacity(topics_to_send.len());
856        for topic in &topics_to_send {
857            let message = BybitSubscription {
858                op: BybitWsOperation::Subscribe,
859                args: vec![topic.clone()],
860                req_id: Some(topic.clone()),
861            };
862            let payload = serde_json::to_string(&message).map_err(|e| {
863                BybitWsError::Json(format!("Failed to serialize subscription: {e}"))
864            })?;
865            payloads.push(payload);
866        }
867
868        let cmd = HandlerCommand::Subscribe { topics: payloads };
869        self.cmd_tx
870            .read()
871            .await
872            .send(cmd)
873            .map_err(|e| BybitWsError::Send(format!("Failed to send subscribe command: {e}")))?;
874
875        Ok(())
876    }
877
878    /// Unsubscribe from the provided topics.
879    pub async fn unsubscribe(&self, topics: Vec<String>) -> BybitWsResult<()> {
880        if topics.is_empty() {
881            return Ok(());
882        }
883
884        log::debug!("Attempting to unsubscribe from topics: {topics:?}");
885
886        if self.signal.load(Ordering::Relaxed) {
887            log::debug!("Shutdown signal detected, skipping unsubscribe");
888            return Ok(());
889        }
890        let _guard = self.subscription_guard.lock().await;
891
892        // Use reference counting to avoid unsubscribing while other consumers still need the topic
893        let mut topics_to_send = Vec::new();
894
895        for topic in topics {
896            // Returns true if this was the last subscription (ref count 1 -> 0)
897            if self.subscriptions.remove_reference(&topic) {
898                self.subscriptions.mark_unsubscribe(&topic);
899                topics_to_send.push(topic.clone());
900            } else {
901                log::debug!("Topic {topic} still has active subscriptions, not unsubscribing");
902            }
903        }
904
905        if topics_to_send.is_empty() {
906            return Ok(());
907        }
908
909        // Serialize unsubscription messages
910        let mut payloads = Vec::with_capacity(topics_to_send.len());
911        for topic in &topics_to_send {
912            let message = BybitSubscription {
913                op: BybitWsOperation::Unsubscribe,
914                args: vec![topic.clone()],
915                req_id: Some(topic.clone()),
916            };
917
918            if let Ok(payload) = serde_json::to_string(&message) {
919                payloads.push(payload);
920            }
921        }
922
923        let cmd = HandlerCommand::Unsubscribe { topics: payloads };
924        if let Err(e) = self.cmd_tx.read().await.send(cmd) {
925            log::debug!("Failed to send unsubscribe command: error={e}");
926        }
927
928        Ok(())
929    }
930
931    /// Returns a stream of venue-typed [`BybitWsMessage`] items.
932    ///
933    /// # Panics
934    ///
935    /// Panics if called before [`Self::connect`] or if the stream has already been taken.
936    pub fn stream(&mut self) -> impl futures_util::Stream<Item = BybitWsMessage> + use<> {
937        let rx = self
938            .out_rx
939            .take()
940            .expect("Stream receiver already taken or client not connected");
941        let mut rx = Arc::try_unwrap(rx).expect("Cannot take ownership - other references exist");
942        async_stream::stream! {
943            while let Some(msg) = rx.recv().await {
944                yield msg;
945            }
946        }
947    }
948
949    /// Returns the number of currently registered subscriptions.
950    #[must_use]
951    pub fn subscription_count(&self) -> usize {
952        self.subscriptions.len()
953    }
954
955    /// Returns the credential associated with this client, if any.
956    #[must_use]
957    pub fn credential(&self) -> Option<&Credential> {
958        self.credential.as_ref()
959    }
960
961    /// Sets the account ID for account message parsing.
962    pub fn set_account_id(&mut self, account_id: AccountId) {
963        self.account_id = Some(account_id);
964    }
965
966    /// Sets the account market maker level.
967    pub fn set_mm_level(&self, mm_level: u8) {
968        self.mm_level.store(mm_level, Ordering::Relaxed);
969    }
970
971    /// Returns the account ID if set.
972    #[must_use]
973    pub fn account_id(&self) -> Option<AccountId> {
974        self.account_id
975    }
976
977    /// Returns the product type for public connections.
978    #[must_use]
979    pub fn product_type(&self) -> Option<BybitProductType> {
980        self.product_type
981    }
982
983    /// Returns a reference to the bar types cache.
984    #[must_use]
985    pub fn bar_types_cache(&self) -> &Arc<AtomicMap<String, BarType>> {
986        &self.bar_types_cache
987    }
988
989    /// Adds an instrument to the shared instruments cache.
990    pub fn cache_instrument(&self, instrument: InstrumentAny) {
991        self.instruments_cache
992            .insert(instrument.id().symbol.inner(), instrument);
993    }
994
995    /// Returns a snapshot of the instruments cache keyed by symbol.
996    #[must_use]
997    pub fn instruments_snapshot(&self) -> ahash::AHashMap<Ustr, InstrumentAny> {
998        (**self.instruments_cache.load()).clone()
999    }
1000
1001    /// Sets whether bar timestamps use the close time.
1002    pub fn set_bars_timestamp_on_close(&self, value: bool) {
1003        self.bars_timestamp_on_close.store(value, Ordering::Relaxed);
1004    }
1005
1006    /// Returns whether bar timestamps use the close time.
1007    #[must_use]
1008    pub fn bars_timestamp_on_close(&self) -> bool {
1009        self.bars_timestamp_on_close.load(Ordering::Relaxed)
1010    }
1011
1012    /// Adds an instrument ID to the option greeks subscription set.
1013    pub fn add_option_greeks_sub(&self, instrument_id: InstrumentId) {
1014        self.option_greeks_subs.insert(instrument_id);
1015    }
1016
1017    /// Removes an instrument ID from the option greeks subscription set.
1018    pub fn remove_option_greeks_sub(&self, instrument_id: &InstrumentId) {
1019        self.option_greeks_subs.remove(instrument_id);
1020    }
1021
1022    /// Returns a reference to the option greeks subscription set.
1023    #[must_use]
1024    pub fn option_greeks_subs(&self) -> &Arc<AtomicSet<InstrumentId>> {
1025        &self.option_greeks_subs
1026    }
1027
1028    /// Returns a reference to the trade subscriptions set.
1029    #[must_use]
1030    pub fn trade_subs(&self) -> &Arc<AtomicSet<InstrumentId>> {
1031        &self.trade_subs
1032    }
1033
1034    /// Returns a reference to the live instruments cache Arc.
1035    #[must_use]
1036    pub fn instruments_cache_ref(&self) -> &Arc<AtomicMap<Ustr, InstrumentAny>> {
1037        &self.instruments_cache
1038    }
1039
1040    /// Subscribes to orderbook updates for a specific instrument.
1041    ///
1042    /// # Errors
1043    ///
1044    /// Returns an error if the subscription request fails.
1045    ///
1046    /// # References
1047    ///
1048    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook>
1049    pub async fn subscribe_orderbook(
1050        &self,
1051        instrument_id: InstrumentId,
1052        depth: u32,
1053    ) -> BybitWsResult<()> {
1054        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1055        let topic = format!(
1056            "{}.{depth}.{raw_symbol}",
1057            BybitWsPublicChannel::OrderBook.as_ref()
1058        );
1059        self.subscribe(vec![topic]).await
1060    }
1061
1062    /// Unsubscribes from orderbook updates for a specific instrument.
1063    pub async fn unsubscribe_orderbook(
1064        &self,
1065        instrument_id: InstrumentId,
1066        depth: u32,
1067    ) -> BybitWsResult<()> {
1068        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1069        let topic = format!(
1070            "{}.{depth}.{raw_symbol}",
1071            BybitWsPublicChannel::OrderBook.as_ref()
1072        );
1073        self.unsubscribe(vec![topic]).await
1074    }
1075
1076    /// Subscribes to public trade updates for a specific instrument.
1077    ///
1078    /// # Errors
1079    ///
1080    /// Returns an error if the subscription request fails.
1081    ///
1082    /// # References
1083    ///
1084    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/trade>
1085    pub async fn subscribe_trades(&self, instrument_id: InstrumentId) -> BybitWsResult<()> {
1086        self.trade_subs.insert(instrument_id);
1087        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1088        // Bybit option trades use baseCoin topic (e.g. publicTrade.BTC)
1089        let topic_symbol = match self.product_type {
1090            Some(BybitProductType::Option) => extract_base_coin(raw_symbol),
1091            _ => raw_symbol,
1092        };
1093        let topic = format!(
1094            "{}.{topic_symbol}",
1095            BybitWsPublicChannel::PublicTrade.as_ref()
1096        );
1097        self.subscribe(vec![topic]).await
1098    }
1099
1100    /// Unsubscribes from public trade updates for a specific instrument.
1101    pub async fn unsubscribe_trades(&self, instrument_id: InstrumentId) -> BybitWsResult<()> {
1102        self.trade_subs.remove(&instrument_id);
1103        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1104        let topic_symbol = match self.product_type {
1105            Some(BybitProductType::Option) => extract_base_coin(raw_symbol),
1106            _ => raw_symbol,
1107        };
1108        let topic = format!(
1109            "{}.{topic_symbol}",
1110            BybitWsPublicChannel::PublicTrade.as_ref()
1111        );
1112        self.unsubscribe(vec![topic]).await
1113    }
1114
1115    /// Subscribes to ticker updates for a specific instrument.
1116    ///
1117    /// # Errors
1118    ///
1119    /// Returns an error if the subscription request fails.
1120    ///
1121    /// # References
1122    ///
1123    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/ticker>
1124    pub async fn subscribe_ticker(&self, instrument_id: InstrumentId) -> BybitWsResult<()> {
1125        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1126        let topic = format!("{}.{raw_symbol}", BybitWsPublicChannel::Tickers.as_ref());
1127        self.subscribe(vec![topic]).await
1128    }
1129
1130    /// Unsubscribes from ticker updates for a specific instrument.
1131    pub async fn unsubscribe_ticker(&self, instrument_id: InstrumentId) -> BybitWsResult<()> {
1132        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1133        let topic = format!("{}.{raw_symbol}", BybitWsPublicChannel::Tickers.as_ref());
1134        self.unsubscribe(vec![topic]).await
1135    }
1136
1137    /// Subscribes to kline/candlestick updates for a specific instrument.
1138    ///
1139    /// # Errors
1140    ///
1141    /// Returns an error if the subscription request fails.
1142    ///
1143    /// # References
1144    ///
1145    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/kline>
1146    pub async fn subscribe_bars(&self, bar_type: BarType) -> BybitWsResult<()> {
1147        if self.product_type == Some(BybitProductType::Option) {
1148            return Err(BybitWsError::ClientError(
1149                "Bybit does not support kline/bar data for options".to_string(),
1150            ));
1151        }
1152
1153        let spec = bar_type.spec();
1154
1155        if spec.price_type != PriceType::Last {
1156            return Err(BybitWsError::ClientError(format!(
1157                "Invalid bar type: Bybit bars only support LAST price type, received {}",
1158                spec.price_type
1159            )));
1160        }
1161
1162        if bar_type.aggregation_source() != AggregationSource::External {
1163            return Err(BybitWsError::ClientError(format!(
1164                "Invalid bar type: Bybit bars only support EXTERNAL aggregation source, received {}",
1165                bar_type.aggregation_source()
1166            )));
1167        }
1168
1169        let interval = bar_spec_to_bybit_interval(spec.aggregation, spec.step.get() as u64)
1170            .map_err(|e| BybitWsError::ClientError(e.to_string()))?;
1171
1172        let instrument_id = bar_type.instrument_id();
1173        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1174        let topic = format!(
1175            "{}.{}.{raw_symbol}",
1176            BybitWsPublicChannel::Kline.as_ref(),
1177            interval
1178        );
1179
1180        // Coordinate with reference counting to avoid duplicate cache entries
1181        if self.subscriptions.get_reference_count(&topic) == 0 {
1182            self.bar_types_cache.insert(topic.clone(), bar_type);
1183        }
1184
1185        self.subscribe(vec![topic]).await
1186    }
1187
1188    /// Unsubscribes from kline/candlestick updates for a specific instrument.
1189    pub async fn unsubscribe_bars(&self, bar_type: BarType) -> BybitWsResult<()> {
1190        let spec = bar_type.spec();
1191        let interval = bar_spec_to_bybit_interval(spec.aggregation, spec.step.get() as u64)
1192            .map_err(|e| BybitWsError::ClientError(e.to_string()))?;
1193
1194        let instrument_id = bar_type.instrument_id();
1195        let raw_symbol = extract_raw_symbol(instrument_id.symbol.as_str());
1196        let topic = format!(
1197            "{}.{}.{raw_symbol}",
1198            BybitWsPublicChannel::Kline.as_ref(),
1199            interval
1200        );
1201
1202        // Coordinate with reference counting to preserve cache for other subscribers
1203        if self.subscriptions.get_reference_count(&topic) == 1 {
1204            self.bar_types_cache.remove(&topic);
1205        }
1206
1207        self.unsubscribe(vec![topic]).await
1208    }
1209
1210    /// Subscribes to order updates.
1211    ///
1212    /// # Errors
1213    ///
1214    /// Returns an error if the subscription request fails or if not authenticated.
1215    ///
1216    /// # References
1217    ///
1218    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/order>
1219    pub async fn subscribe_orders(&self) -> BybitWsResult<()> {
1220        if !self.requires_auth {
1221            return Err(BybitWsError::Authentication(
1222                "Order subscription requires authentication".to_string(),
1223            ));
1224        }
1225        self.subscribe(vec![BybitWsPrivateChannel::Order.as_ref().to_string()])
1226            .await
1227    }
1228
1229    /// Unsubscribes from order updates.
1230    pub async fn unsubscribe_orders(&self) -> BybitWsResult<()> {
1231        self.unsubscribe(vec![BybitWsPrivateChannel::Order.as_ref().to_string()])
1232            .await
1233    }
1234
1235    /// Subscribes to execution/fill updates.
1236    ///
1237    /// # Errors
1238    ///
1239    /// Returns an error if the subscription request fails or if not authenticated.
1240    ///
1241    /// # References
1242    ///
1243    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/execution>
1244    pub async fn subscribe_executions(&self) -> BybitWsResult<()> {
1245        if !self.requires_auth {
1246            return Err(BybitWsError::Authentication(
1247                "Execution subscription requires authentication".to_string(),
1248            ));
1249        }
1250        self.subscribe(vec![BybitWsPrivateChannel::Execution.as_ref().to_string()])
1251            .await
1252    }
1253
1254    /// Unsubscribes from execution/fill updates.
1255    pub async fn unsubscribe_executions(&self) -> BybitWsResult<()> {
1256        self.unsubscribe(vec![BybitWsPrivateChannel::Execution.as_ref().to_string()])
1257            .await
1258    }
1259
1260    /// Subscribes to fast execution updates (slim payload, lower latency).
1261    ///
1262    /// # Errors
1263    ///
1264    /// Returns an error if the subscription request fails or if not authenticated.
1265    ///
1266    /// # References
1267    ///
1268    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution>
1269    pub async fn subscribe_executions_fast(&self) -> BybitWsResult<()> {
1270        if !self.requires_auth {
1271            return Err(BybitWsError::Authentication(
1272                "Fast execution subscription requires authentication".to_string(),
1273            ));
1274        }
1275        self.subscribe(vec![
1276            BybitWsPrivateChannel::ExecutionFast.as_ref().to_string(),
1277        ])
1278        .await
1279    }
1280
1281    /// Unsubscribes from fast execution updates.
1282    pub async fn unsubscribe_executions_fast(&self) -> BybitWsResult<()> {
1283        self.unsubscribe(vec![
1284            BybitWsPrivateChannel::ExecutionFast.as_ref().to_string(),
1285        ])
1286        .await
1287    }
1288
1289    /// Subscribes to position updates.
1290    ///
1291    /// # Errors
1292    ///
1293    /// Returns an error if the subscription request fails or if not authenticated.
1294    ///
1295    /// # References
1296    ///
1297    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/position>
1298    pub async fn subscribe_positions(&self) -> BybitWsResult<()> {
1299        if !self.requires_auth {
1300            return Err(BybitWsError::Authentication(
1301                "Position subscription requires authentication".to_string(),
1302            ));
1303        }
1304        self.subscribe(vec![BybitWsPrivateChannel::Position.as_ref().to_string()])
1305            .await
1306    }
1307
1308    /// Unsubscribes from position updates.
1309    pub async fn unsubscribe_positions(&self) -> BybitWsResult<()> {
1310        self.unsubscribe(vec![BybitWsPrivateChannel::Position.as_ref().to_string()])
1311            .await
1312    }
1313
1314    /// Subscribes to wallet/balance updates.
1315    ///
1316    /// # Errors
1317    ///
1318    /// Returns an error if the subscription request fails or if not authenticated.
1319    ///
1320    /// # References
1321    ///
1322    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/wallet>
1323    pub async fn subscribe_wallet(&self) -> BybitWsResult<()> {
1324        if !self.requires_auth {
1325            return Err(BybitWsError::Authentication(
1326                "Wallet subscription requires authentication".to_string(),
1327            ));
1328        }
1329        self.subscribe(vec![BybitWsPrivateChannel::Wallet.as_ref().to_string()])
1330            .await
1331    }
1332
1333    /// Unsubscribes from wallet/balance updates.
1334    pub async fn unsubscribe_wallet(&self) -> BybitWsResult<()> {
1335        self.unsubscribe(vec![BybitWsPrivateChannel::Wallet.as_ref().to_string()])
1336            .await
1337    }
1338
1339    /// Waits for the session to be authenticated, aborting early if the client
1340    /// enters a terminal state (closed or disconnecting) during the wait.
1341    async fn require_authenticated(&self) -> BybitWsResult<()> {
1342        if self.is_closed() {
1343            return Err(BybitWsError::ClientError(
1344                "WebSocket client is closed".to_string(),
1345            ));
1346        }
1347
1348        if self.auth_tracker.is_authenticated() {
1349            return Ok(());
1350        }
1351
1352        tokio::select! {
1353            authenticated = self.auth_tracker.wait_for_authenticated(self.auth_wait_timeout) => {
1354                if authenticated {
1355                    Ok(())
1356                } else {
1357                    Err(BybitWsError::Authentication(
1358                        "Must be authenticated".to_string(),
1359                    ))
1360                }
1361            }
1362            () = async {
1363                loop {
1364                    tokio::time::sleep(Duration::from_millis(100)).await;
1365
1366                    if self.is_closed() {
1367                        return;
1368                    }
1369                }
1370            } => {
1371                Err(BybitWsError::ClientError(
1372                    "WebSocket client closed during authentication wait".to_string(),
1373                ))
1374            }
1375        }
1376    }
1377
1378    /// Allocates correlation IDs for product-specific batch chunks.
1379    #[must_use]
1380    pub(crate) fn batch_request_ids(category: BybitProductType, order_count: usize) -> Vec<String> {
1381        let request_count = order_count.div_ceil(batch_send_limit(category));
1382        (0..request_count)
1383            .map(|_| UUID4::new().to_string())
1384            .collect()
1385    }
1386
1387    fn batch_category(
1388        mut categories: impl Iterator<Item = BybitProductType>,
1389    ) -> BybitWsResult<BybitProductType> {
1390        let category = categories.next().ok_or_else(|| {
1391            BybitWsError::ClientError("Batch order request cannot be empty".to_string())
1392        })?;
1393
1394        if categories.any(|candidate| candidate != category) {
1395            return Err(BybitWsError::ClientError(
1396                "Batch order request cannot mix product categories".to_string(),
1397            ));
1398        }
1399        Ok(category)
1400    }
1401
1402    /// Places an order via WebSocket, returning the request ID for correlation.
1403    ///
1404    /// # Errors
1405    ///
1406    /// Returns an error if the order request fails or if not authenticated.
1407    pub async fn place_order(&self, params: BybitWsPlaceOrderParams) -> BybitWsResult<String> {
1408        let req_id = UUID4::new().to_string();
1409        self.place_order_with_id(params, req_id.clone()).await?;
1410        Ok(req_id)
1411    }
1412
1413    pub(crate) async fn place_order_with_id(
1414        &self,
1415        params: BybitWsPlaceOrderParams,
1416        req_id: String,
1417    ) -> BybitWsResult<()> {
1418        self.require_authenticated().await?;
1419        let category = params.category;
1420
1421        let referer = if self.include_referer_header(params.time_in_force) {
1422            Some(BYBIT_NAUTILUS_BROKER_ID.to_string())
1423        } else {
1424            None
1425        };
1426
1427        let command = BybitWsOrderCommand {
1428            req_id,
1429            op: BybitWsOrderRequestOp::Create,
1430            category,
1431            weight: 1,
1432            referer,
1433            args: vec![serde_json::to_value(params)?],
1434        };
1435        self.send_cmd(HandlerCommand::SendOrder { command }).await
1436    }
1437
1438    /// Amends an existing order via WebSocket, returning the request ID for correlation.
1439    ///
1440    /// # Errors
1441    ///
1442    /// Returns an error if the amend request fails or if not authenticated.
1443    pub async fn amend_order(&self, params: BybitWsAmendOrderParams) -> BybitWsResult<String> {
1444        let req_id = UUID4::new().to_string();
1445        self.amend_order_with_id(params, req_id.clone()).await?;
1446        Ok(req_id)
1447    }
1448
1449    pub(crate) async fn amend_order_with_id(
1450        &self,
1451        params: BybitWsAmendOrderParams,
1452        req_id: String,
1453    ) -> BybitWsResult<()> {
1454        self.require_authenticated().await?;
1455        let command = BybitWsOrderCommand {
1456            category: params.category,
1457            req_id,
1458            op: BybitWsOrderRequestOp::Amend,
1459            weight: 1,
1460            referer: None,
1461            args: vec![serde_json::to_value(params)?],
1462        };
1463        self.send_cmd(HandlerCommand::SendOrder { command }).await
1464    }
1465
1466    /// Cancels an order via WebSocket, returning the request ID for correlation.
1467    ///
1468    /// # Errors
1469    ///
1470    /// Returns an error if the cancel request fails or if not authenticated.
1471    pub async fn cancel_order(&self, params: BybitWsCancelOrderParams) -> BybitWsResult<String> {
1472        let req_id = UUID4::new().to_string();
1473        self.cancel_order_with_id(params, req_id.clone()).await?;
1474        Ok(req_id)
1475    }
1476
1477    pub(crate) async fn cancel_order_with_id(
1478        &self,
1479        params: BybitWsCancelOrderParams,
1480        req_id: String,
1481    ) -> BybitWsResult<()> {
1482        self.require_authenticated().await?;
1483        let command = BybitWsOrderCommand {
1484            category: params.category,
1485            req_id,
1486            op: BybitWsOrderRequestOp::Cancel,
1487            weight: 1,
1488            referer: None,
1489            args: vec![serde_json::to_value(params)?],
1490        };
1491        self.send_cmd(HandlerCommand::SendOrder { command }).await
1492    }
1493
1494    /// Batch creates multiple orders via WebSocket, returning the request ID for correlation.
1495    ///
1496    /// # Errors
1497    ///
1498    /// Returns an error if the batch request fails or if not authenticated.
1499    pub async fn batch_place_orders(
1500        &self,
1501        orders: Vec<BybitWsPlaceOrderParams>,
1502    ) -> BybitWsResult<Vec<String>> {
1503        self.require_authenticated().await?;
1504
1505        if orders.is_empty() {
1506            log::warn!("Batch place orders called with empty orders list");
1507            return Ok(vec![]);
1508        }
1509
1510        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1511        let req_ids = Self::batch_request_ids(category, orders.len());
1512        self.batch_place_orders_with_ids(orders, req_ids.clone())
1513            .await?;
1514        Ok(req_ids)
1515    }
1516
1517    pub(crate) async fn batch_place_orders_with_ids(
1518        &self,
1519        orders: Vec<BybitWsPlaceOrderParams>,
1520        req_ids: Vec<String>,
1521    ) -> BybitWsResult<()> {
1522        self.require_authenticated().await?;
1523        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1524        let chunk_limit = batch_send_limit(category);
1525        if req_ids.len() != orders.len().div_ceil(chunk_limit) {
1526            return Err(BybitWsError::ClientError(
1527                "Batch request ID count does not match order chunks".to_string(),
1528            ));
1529        }
1530
1531        let mut commands = Vec::with_capacity(req_ids.len());
1532        for (orders, req_id) in orders.chunks(chunk_limit).zip(req_ids) {
1533            commands.push(self.build_batch_place_command(orders.to_vec(), req_id)?);
1534        }
1535        self.send_cmd(HandlerCommand::SendOrders { commands }).await
1536    }
1537
1538    fn build_batch_place_command(
1539        &self,
1540        orders: Vec<BybitWsPlaceOrderParams>,
1541        req_id: String,
1542    ) -> BybitWsResult<BybitWsOrderCommand> {
1543        let category = orders[0].category;
1544        let order_count = orders.len();
1545
1546        let mm_level = self.mm_level.load(Ordering::Relaxed);
1547        let has_non_post_only = orders
1548            .iter()
1549            .any(|o| !matches!(o.time_in_force, Some(BybitTimeInForce::PostOnly)));
1550        let referer = if has_non_post_only || mm_level == 0 {
1551            Some(BYBIT_NAUTILUS_BROKER_ID.to_string())
1552        } else {
1553            None
1554        };
1555
1556        let request_items: Vec<BybitWsBatchPlaceItem> = orders
1557            .into_iter()
1558            .map(|order| BybitWsBatchPlaceItem {
1559                symbol: order.symbol,
1560                side: order.side,
1561                order_type: order.order_type,
1562                qty: order.qty,
1563                is_leverage: order.is_leverage,
1564                market_unit: order.market_unit,
1565                price: order.price,
1566                time_in_force: order.time_in_force,
1567                order_link_id: order.order_link_id,
1568                reduce_only: order.reduce_only,
1569                close_on_trigger: order.close_on_trigger,
1570                trigger_price: order.trigger_price,
1571                trigger_by: order.trigger_by,
1572                trigger_direction: order.trigger_direction,
1573                tpsl_mode: order.tpsl_mode,
1574                take_profit: order.take_profit,
1575                stop_loss: order.stop_loss,
1576                tp_trigger_by: order.tp_trigger_by,
1577                sl_trigger_by: order.sl_trigger_by,
1578                sl_trigger_price: order.sl_trigger_price,
1579                tp_trigger_price: order.tp_trigger_price,
1580                sl_order_type: order.sl_order_type,
1581                tp_order_type: order.tp_order_type,
1582                sl_limit_price: order.sl_limit_price,
1583                tp_limit_price: order.tp_limit_price,
1584                order_iv: order.order_iv,
1585                mmp: order.mmp,
1586                position_idx: order.position_idx,
1587                bbo_side_type: order.bbo_side_type,
1588                bbo_level: order.bbo_level,
1589            })
1590            .collect();
1591
1592        let args = BybitWsBatchPlaceOrderArgs {
1593            category,
1594            request: request_items,
1595        };
1596
1597        Ok(BybitWsOrderCommand {
1598            req_id,
1599            op: BybitWsOrderRequestOp::CreateBatch,
1600            category,
1601            weight: batch_weight(category, order_count),
1602            referer,
1603            args: vec![serde_json::to_value(args)?],
1604        })
1605    }
1606
1607    /// Batch amends multiple orders via WebSocket.
1608    ///
1609    /// # Errors
1610    ///
1611    /// Returns an error if the batch request fails or if not authenticated.
1612    pub async fn batch_amend_orders(
1613        &self,
1614        orders: Vec<BybitWsAmendOrderParams>,
1615    ) -> BybitWsResult<Vec<String>> {
1616        self.require_authenticated().await?;
1617
1618        if orders.is_empty() {
1619            log::warn!("Batch amend orders called with empty orders list");
1620            return Ok(vec![]);
1621        }
1622
1623        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1624        let req_ids = Self::batch_request_ids(category, orders.len());
1625        self.batch_amend_orders_with_ids(orders, req_ids.clone())
1626            .await?;
1627        Ok(req_ids)
1628    }
1629
1630    pub(crate) async fn batch_amend_orders_with_ids(
1631        &self,
1632        orders: Vec<BybitWsAmendOrderParams>,
1633        req_ids: Vec<String>,
1634    ) -> BybitWsResult<()> {
1635        self.require_authenticated().await?;
1636        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1637        let chunk_limit = batch_send_limit(category);
1638        if req_ids.len() != orders.len().div_ceil(chunk_limit) {
1639            return Err(BybitWsError::ClientError(
1640                "Batch request ID count does not match order chunks".to_string(),
1641            ));
1642        }
1643
1644        let mut commands = Vec::with_capacity(req_ids.len());
1645        for (orders, req_id) in orders.chunks(chunk_limit).zip(req_ids) {
1646            commands.push(Self::build_batch_amend_command(orders.to_vec(), req_id)?);
1647        }
1648        self.send_cmd(HandlerCommand::SendOrders { commands }).await
1649    }
1650
1651    fn build_batch_amend_command(
1652        orders: Vec<BybitWsAmendOrderParams>,
1653        req_id: String,
1654    ) -> BybitWsResult<BybitWsOrderCommand> {
1655        let category = orders[0].category;
1656        let order_count = orders.len();
1657
1658        let request_items = orders
1659            .into_iter()
1660            .map(|order| BybitWsBatchAmendItem {
1661                symbol: order.symbol,
1662                order_id: order.order_id,
1663                order_link_id: order.order_link_id,
1664                qty: order.qty,
1665                price: order.price,
1666                trigger_price: order.trigger_price,
1667                take_profit: order.take_profit,
1668                stop_loss: order.stop_loss,
1669                tp_trigger_by: order.tp_trigger_by,
1670                sl_trigger_by: order.sl_trigger_by,
1671                order_iv: order.order_iv,
1672            })
1673            .collect();
1674
1675        let args = BybitWsBatchAmendOrderArgs {
1676            category,
1677            request: request_items,
1678        };
1679
1680        Ok(BybitWsOrderCommand {
1681            req_id,
1682            op: BybitWsOrderRequestOp::AmendBatch,
1683            category,
1684            weight: batch_weight(category, order_count),
1685            referer: None,
1686            args: vec![serde_json::to_value(args)?],
1687        })
1688    }
1689
1690    /// Batch cancels multiple orders via WebSocket, returning the request ID for correlation.
1691    ///
1692    /// # Errors
1693    ///
1694    /// Returns an error if the batch request fails or if not authenticated.
1695    pub async fn batch_cancel_orders(
1696        &self,
1697        orders: Vec<BybitWsCancelOrderParams>,
1698    ) -> BybitWsResult<Vec<String>> {
1699        self.require_authenticated().await?;
1700
1701        if orders.is_empty() {
1702            log::warn!("Batch cancel orders called with empty orders list");
1703            return Ok(vec![]);
1704        }
1705
1706        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1707        let req_ids = Self::batch_request_ids(category, orders.len());
1708        self.batch_cancel_orders_with_ids(orders, req_ids.clone())
1709            .await?;
1710        Ok(req_ids)
1711    }
1712
1713    pub(crate) async fn batch_cancel_orders_with_ids(
1714        &self,
1715        orders: Vec<BybitWsCancelOrderParams>,
1716        req_ids: Vec<String>,
1717    ) -> BybitWsResult<()> {
1718        self.require_authenticated().await?;
1719
1720        if orders.is_empty() {
1721            return Ok(());
1722        }
1723
1724        let category = Self::batch_category(orders.iter().map(|order| order.category))?;
1725        let chunk_limit = batch_send_limit(category);
1726        if req_ids.len() != orders.len().div_ceil(chunk_limit) {
1727            return Err(BybitWsError::ClientError(
1728                "Batch request ID count does not match order chunks".to_string(),
1729            ));
1730        }
1731
1732        let mut commands = Vec::with_capacity(req_ids.len());
1733        for (orders, req_id) in orders.chunks(chunk_limit).zip(req_ids) {
1734            commands.push(Self::build_batch_cancel_command(orders.to_vec(), req_id)?);
1735        }
1736        self.send_cmd(HandlerCommand::SendOrders { commands }).await
1737    }
1738
1739    fn build_batch_cancel_command(
1740        orders: Vec<BybitWsCancelOrderParams>,
1741        req_id: String,
1742    ) -> BybitWsResult<BybitWsOrderCommand> {
1743        let category = orders[0].category;
1744        let order_count = orders.len();
1745
1746        let request_items: Vec<BybitWsBatchCancelItem> = orders
1747            .into_iter()
1748            .map(|order| BybitWsBatchCancelItem {
1749                symbol: order.symbol,
1750                order_id: order.order_id,
1751                order_link_id: order.order_link_id,
1752            })
1753            .collect();
1754
1755        let args = BybitWsBatchCancelOrderArgs {
1756            category,
1757            request: request_items,
1758        };
1759
1760        Ok(BybitWsOrderCommand {
1761            req_id,
1762            op: BybitWsOrderRequestOp::CancelBatch,
1763            category,
1764            weight: batch_weight(category, order_count),
1765            referer: None,
1766            args: vec![serde_json::to_value(args)?],
1767        })
1768    }
1769
1770    /// Submits an order using Nautilus domain objects.
1771    ///
1772    /// # Errors
1773    ///
1774    /// Returns an error if order submission fails or if not authenticated.
1775    #[expect(clippy::too_many_arguments)]
1776    pub async fn submit_order(
1777        &self,
1778        product_type: BybitProductType,
1779        instrument_id: InstrumentId,
1780        client_order_id: ClientOrderId,
1781        order_side: OrderSide,
1782        order_type: OrderType,
1783        quantity: Quantity,
1784        is_quote_quantity: bool,
1785        time_in_force: Option<TimeInForce>,
1786        price: Option<Price>,
1787        trigger_price: Option<Price>,
1788        trigger_type: Option<TriggerType>,
1789        post_only: Option<bool>,
1790        reduce_only: Option<bool>,
1791        is_leverage: bool,
1792        position_idx: Option<BybitPositionIdx>,
1793        bbo_side_type: Option<BybitBboSideType>,
1794        bbo_level: Option<String>,
1795    ) -> BybitWsResult<String> {
1796        let params = self.build_place_order_params(
1797            product_type,
1798            instrument_id,
1799            client_order_id,
1800            order_side,
1801            order_type,
1802            quantity,
1803            is_quote_quantity,
1804            time_in_force,
1805            price,
1806            trigger_price,
1807            trigger_type,
1808            post_only,
1809            reduce_only,
1810            is_leverage,
1811            None,
1812            None,
1813            position_idx,
1814            bbo_side_type,
1815            bbo_level,
1816        )?;
1817
1818        self.place_order(params).await
1819    }
1820
1821    /// Modifies an existing order using Nautilus domain objects.
1822    ///
1823    /// # Errors
1824    ///
1825    /// Returns an error if modification fails or if not authenticated.
1826    pub async fn modify_order(
1827        &self,
1828        product_type: BybitProductType,
1829        instrument_id: InstrumentId,
1830        client_order_id: ClientOrderId,
1831        venue_order_id: Option<VenueOrderId>,
1832        quantity: Option<Quantity>,
1833        price: Option<Price>,
1834    ) -> BybitWsResult<String> {
1835        let params = self.build_amend_order_params(
1836            product_type,
1837            instrument_id,
1838            venue_order_id,
1839            Some(client_order_id),
1840            quantity,
1841            price,
1842        )?;
1843
1844        self.amend_order(params).await
1845    }
1846
1847    /// Cancels an order using Nautilus domain objects.
1848    ///
1849    /// # Errors
1850    ///
1851    /// Returns an error if cancellation fails or if not authenticated.
1852    pub async fn cancel_order_by_id(
1853        &self,
1854        product_type: BybitProductType,
1855        instrument_id: InstrumentId,
1856        client_order_id: ClientOrderId,
1857        venue_order_id: Option<VenueOrderId>,
1858    ) -> BybitWsResult<String> {
1859        let params = self.build_cancel_order_params(
1860            product_type,
1861            instrument_id,
1862            venue_order_id,
1863            Some(client_order_id),
1864        )?;
1865
1866        self.cancel_order(params).await
1867    }
1868
1869    /// Builds order params for placing an order.
1870    #[expect(clippy::too_many_arguments)]
1871    pub fn build_place_order_params(
1872        &self,
1873        product_type: BybitProductType,
1874        instrument_id: InstrumentId,
1875        client_order_id: ClientOrderId,
1876        order_side: OrderSide,
1877        order_type: OrderType,
1878        quantity: Quantity,
1879        is_quote_quantity: bool,
1880        time_in_force: Option<TimeInForce>,
1881        price: Option<Price>,
1882        trigger_price: Option<Price>,
1883        trigger_type: Option<TriggerType>,
1884        post_only: Option<bool>,
1885        reduce_only: Option<bool>,
1886        is_leverage: bool,
1887        take_profit: Option<Price>,
1888        stop_loss: Option<Price>,
1889        position_idx: Option<BybitPositionIdx>,
1890        bbo_side_type: Option<BybitBboSideType>,
1891        bbo_level: Option<String>,
1892    ) -> BybitWsResult<BybitWsPlaceOrderParams> {
1893        let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())
1894            .map_err(|e| BybitWsError::ClientError(e.to_string()))?;
1895        let raw_symbol = Ustr::from(bybit_symbol.raw_symbol());
1896
1897        let bybit_side = match order_side {
1898            OrderSide::Buy => BybitOrderSide::Buy,
1899            OrderSide::Sell => BybitOrderSide::Sell,
1900        };
1901
1902        let (bybit_order_type, is_stop_order) = match order_type {
1903            OrderType::Market => (BybitOrderType::Market, false),
1904            OrderType::Limit => (BybitOrderType::Limit, false),
1905            OrderType::StopMarket | OrderType::MarketIfTouched => (BybitOrderType::Market, true),
1906            OrderType::StopLimit | OrderType::LimitIfTouched => (BybitOrderType::Limit, true),
1907            _ => {
1908                return Err(BybitWsError::ClientError(format!(
1909                    "Unsupported order type: {order_type:?}"
1910                )));
1911            }
1912        };
1913
1914        let bybit_tif =
1915            map_time_in_force(bybit_order_type, time_in_force, post_only).map_err(|tif| {
1916                BybitWsError::ClientError(format!("Unsupported time in force: {tif:?}"))
1917            })?;
1918        let market_unit = spot_market_unit(product_type, bybit_order_type, is_quote_quantity);
1919        let is_leverage_value = spot_leverage(product_type, is_leverage);
1920        let trigger_dir =
1921            trigger_direction(order_type, order_side, is_stop_order).map(|d| d as i32);
1922
1923        let params = if is_stop_order {
1924            BybitWsPlaceOrderParams {
1925                category: product_type,
1926                symbol: raw_symbol,
1927                side: bybit_side,
1928                order_type: bybit_order_type,
1929                qty: quantity.to_string(),
1930                is_leverage: is_leverage_value,
1931                market_unit,
1932                price: if bbo_side_type.is_some() {
1933                    None
1934                } else {
1935                    price.map(|p| p.to_string())
1936                },
1937                time_in_force: bybit_tif,
1938                order_link_id: Some(client_order_id.to_string()),
1939                reduce_only: reduce_only.filter(|&r| r),
1940                close_on_trigger: None,
1941                trigger_price: trigger_price.map(|p| p.to_string()),
1942                trigger_by: Some(resolve_trigger_type(trigger_type)),
1943                trigger_direction: trigger_dir,
1944                tpsl_mode: if take_profit.is_some() || stop_loss.is_some() {
1945                    Some(BybitTpSlMode::Full)
1946                } else {
1947                    None
1948                },
1949                take_profit: take_profit.map(|p| p.to_string()),
1950                stop_loss: stop_loss.map(|p| p.to_string()),
1951                tp_trigger_by: take_profit.map(|_| resolve_trigger_type(trigger_type)),
1952                sl_trigger_by: stop_loss.map(|_| resolve_trigger_type(trigger_type)),
1953                sl_trigger_price: None,
1954                tp_trigger_price: None,
1955                sl_order_type: None,
1956                tp_order_type: None,
1957                sl_limit_price: None,
1958                tp_limit_price: None,
1959                order_iv: None,
1960                mmp: None,
1961                position_idx,
1962                bbo_side_type,
1963                bbo_level,
1964            }
1965        } else {
1966            BybitWsPlaceOrderParams {
1967                category: product_type,
1968                symbol: raw_symbol,
1969                side: bybit_side,
1970                order_type: bybit_order_type,
1971                qty: quantity.to_string(),
1972                is_leverage: is_leverage_value,
1973                market_unit,
1974                price: if bbo_side_type.is_some() {
1975                    None
1976                } else {
1977                    price.map(|p| p.to_string())
1978                },
1979                time_in_force: bybit_tif,
1980                order_link_id: Some(client_order_id.to_string()),
1981                reduce_only: reduce_only.filter(|&r| r),
1982                close_on_trigger: None,
1983                trigger_price: None,
1984                trigger_by: None,
1985                trigger_direction: None,
1986                tpsl_mode: if take_profit.is_some() || stop_loss.is_some() {
1987                    Some(BybitTpSlMode::Full)
1988                } else {
1989                    None
1990                },
1991                take_profit: take_profit.map(|p| p.to_string()),
1992                stop_loss: stop_loss.map(|p| p.to_string()),
1993                tp_trigger_by: take_profit.map(|_| resolve_trigger_type(trigger_type)),
1994                sl_trigger_by: stop_loss.map(|_| resolve_trigger_type(trigger_type)),
1995                sl_trigger_price: None,
1996                tp_trigger_price: None,
1997                sl_order_type: None,
1998                tp_order_type: None,
1999                sl_limit_price: None,
2000                tp_limit_price: None,
2001                order_iv: None,
2002                mmp: None,
2003                position_idx,
2004                bbo_side_type,
2005                bbo_level,
2006            }
2007        };
2008
2009        Ok(params)
2010    }
2011
2012    /// Builds order params for amending an order.
2013    pub fn build_amend_order_params(
2014        &self,
2015        product_type: BybitProductType,
2016        instrument_id: InstrumentId,
2017        venue_order_id: Option<VenueOrderId>,
2018        client_order_id: Option<ClientOrderId>,
2019        quantity: Option<Quantity>,
2020        price: Option<Price>,
2021    ) -> BybitWsResult<BybitWsAmendOrderParams> {
2022        let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())
2023            .map_err(|e| BybitWsError::ClientError(e.to_string()))?;
2024        let raw_symbol = Ustr::from(bybit_symbol.raw_symbol());
2025
2026        Ok(BybitWsAmendOrderParams {
2027            category: product_type,
2028            symbol: raw_symbol,
2029            order_id: venue_order_id.map(|v| v.to_string()),
2030            order_link_id: client_order_id.map(|c| c.to_string()),
2031            qty: quantity.map(|q| q.to_string()),
2032            price: price.map(|p| p.to_string()),
2033            trigger_price: None,
2034            take_profit: None,
2035            stop_loss: None,
2036            tp_trigger_by: None,
2037            sl_trigger_by: None,
2038            order_iv: None,
2039        })
2040    }
2041
2042    /// Builds order params for canceling an order via WebSocket.
2043    ///
2044    /// # Errors
2045    ///
2046    /// Returns an error if symbol parsing fails or if neither venue_order_id
2047    /// nor client_order_id is provided.
2048    pub fn build_cancel_order_params(
2049        &self,
2050        product_type: BybitProductType,
2051        instrument_id: InstrumentId,
2052        venue_order_id: Option<VenueOrderId>,
2053        client_order_id: Option<ClientOrderId>,
2054    ) -> BybitWsResult<BybitWsCancelOrderParams> {
2055        if venue_order_id.is_none() && client_order_id.is_none() {
2056            return Err(BybitWsError::ClientError(
2057                "Either venue_order_id or client_order_id must be provided".to_string(),
2058            ));
2059        }
2060
2061        let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())
2062            .map_err(|e| BybitWsError::ClientError(e.to_string()))?;
2063        let raw_symbol = Ustr::from(bybit_symbol.raw_symbol());
2064
2065        Ok(BybitWsCancelOrderParams {
2066            category: product_type,
2067            symbol: raw_symbol,
2068            order_id: venue_order_id.map(|v| v.to_string()),
2069            order_link_id: client_order_id.map(|c| c.to_string()),
2070        })
2071    }
2072
2073    fn include_referer_header(&self, time_in_force: Option<BybitTimeInForce>) -> bool {
2074        let is_post_only = matches!(time_in_force, Some(BybitTimeInForce::PostOnly));
2075        let mm_level = self.mm_level.load(Ordering::Relaxed);
2076        !(is_post_only && mm_level > 0)
2077    }
2078
2079    fn default_headers() -> Vec<(String, String)> {
2080        vec![
2081            ("Content-Type".to_string(), "application/json".to_string()),
2082            (USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string()),
2083        ]
2084    }
2085
2086    async fn authenticate_if_required(&self) -> BybitWsResult<()> {
2087        if !self.requires_auth {
2088            return Ok(());
2089        }
2090
2091        let credential = self.credential.as_ref().ok_or_else(|| {
2092            BybitWsError::Authentication("Credentials required for authentication".to_string())
2093        })?;
2094
2095        let expires = jiff::Timestamp::now().as_millisecond() + WEBSOCKET_AUTH_WINDOW_MS;
2096        let signature = credential.sign_websocket_auth(expires);
2097
2098        let auth_message = BybitAuthRequest {
2099            op: BybitWsOperation::Auth,
2100            args: vec![
2101                Value::String(credential.api_key().to_string()),
2102                Value::Number(expires.into()),
2103                Value::String(signature),
2104            ],
2105        };
2106
2107        let payload = serde_json::to_string(&auth_message)?;
2108
2109        // Begin auth attempt so succeed() will update state
2110        let _rx = self.auth_tracker.begin();
2111
2112        self.cmd_tx
2113            .read()
2114            .await
2115            .send(HandlerCommand::Authenticate { payload })
2116            .map_err(|e| BybitWsError::Send(format!("Failed to send auth command: {e}")))?;
2117
2118        Ok(())
2119    }
2120
2121    async fn send_cmd(&self, cmd: HandlerCommand) -> BybitWsResult<()> {
2122        self.cmd_tx
2123            .read()
2124            .await
2125            .send(cmd)
2126            .map_err(|e| BybitWsError::Send(e.to_string()))
2127    }
2128}
2129
2130impl Drop for BybitWebSocketClient {
2131    fn drop(&mut self) {
2132        if Arc::strong_count(&self.task_handle) == 1 && !self.task_handle.is_empty() {
2133            self.cancellation_token.load().cancel();
2134            self.signal.store(true, Ordering::Relaxed);
2135            self.task_handle.abort();
2136        }
2137    }
2138}
2139
2140#[cfg(test)]
2141mod tests {
2142    use rstest::rstest;
2143
2144    use super::*;
2145    use crate::{
2146        common::{enums::BybitMarketUnit, testing::load_test_json},
2147        websocket::{messages::BybitWsFrame, parse_bybit_ws_frame},
2148    };
2149
2150    #[tokio::test]
2151    async fn test_drop_clone_does_not_cancel_handler() {
2152        let client = BybitWebSocketClient::new_public(Some("wss://test".to_string()), 30);
2153        let cancellation_token = CancellationToken::new();
2154        client
2155            .cancellation_token
2156            .store(Arc::new(cancellation_token.clone()));
2157        client
2158            .task_handle
2159            .insert(get_runtime().spawn(std::future::pending()));
2160        let clone = client.clone();
2161
2162        drop(clone);
2163
2164        assert!(!cancellation_token.is_cancelled());
2165        assert!(!client.task_handle.is_empty());
2166    }
2167
2168    #[rstest]
2169    fn classify_orderbook_snapshot() {
2170        let json: Value = serde_json::from_str(&load_test_json("ws_orderbook_snapshot.json"))
2171            .expect("invalid fixture");
2172        let frame = parse_bybit_ws_frame(json);
2173        assert!(matches!(frame, BybitWsFrame::Orderbook(_)));
2174    }
2175
2176    #[rstest]
2177    fn classify_trade_snapshot() {
2178        let json: Value =
2179            serde_json::from_str(&load_test_json("ws_public_trade.json")).expect("invalid fixture");
2180        let frame = parse_bybit_ws_frame(json);
2181        assert!(matches!(frame, BybitWsFrame::Trade(_)));
2182    }
2183
2184    #[rstest]
2185    fn classify_ticker_linear_snapshot() {
2186        let json: Value = serde_json::from_str(&load_test_json("ws_ticker_linear.json"))
2187            .expect("invalid fixture");
2188        let frame = parse_bybit_ws_frame(json);
2189        assert!(matches!(frame, BybitWsFrame::TickerLinear(_)));
2190    }
2191
2192    #[rstest]
2193    fn classify_ticker_option_snapshot() {
2194        let json: Value = serde_json::from_str(&load_test_json("ws_ticker_option.json"))
2195            .expect("invalid fixture");
2196        let frame = parse_bybit_ws_frame(json);
2197        assert!(matches!(frame, BybitWsFrame::TickerOption(_)));
2198    }
2199
2200    #[rstest]
2201    fn test_race_unsubscribe_failure_recovery() {
2202        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);
2203        let topic = "publicTrade.BTCUSDT";
2204
2205        subscriptions.mark_subscribe(topic);
2206        subscriptions.confirm_subscribe(topic);
2207        assert_eq!(subscriptions.len(), 1);
2208
2209        subscriptions.mark_unsubscribe(topic);
2210        assert_eq!(subscriptions.len(), 0);
2211        assert_eq!(subscriptions.pending_unsubscribe_topics(), vec![topic]);
2212
2213        subscriptions.confirm_unsubscribe(topic);
2214        subscriptions.mark_subscribe(topic);
2215        subscriptions.confirm_subscribe(topic);
2216
2217        assert_eq!(subscriptions.len(), 1);
2218        assert!(subscriptions.pending_unsubscribe_topics().is_empty());
2219        assert!(subscriptions.pending_subscribe_topics().is_empty());
2220
2221        let all = subscriptions.all_topics();
2222        assert_eq!(all.len(), 1);
2223        assert!(all.contains(&topic.to_string()));
2224    }
2225
2226    #[rstest]
2227    fn test_race_resubscribe_before_unsubscribe_ack() {
2228        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);
2229        let topic = "orderbook.50.BTCUSDT";
2230
2231        subscriptions.mark_subscribe(topic);
2232        subscriptions.confirm_subscribe(topic);
2233        assert_eq!(subscriptions.len(), 1);
2234
2235        subscriptions.mark_unsubscribe(topic);
2236        assert_eq!(subscriptions.len(), 0);
2237        assert_eq!(subscriptions.pending_unsubscribe_topics(), vec![topic]);
2238
2239        subscriptions.mark_subscribe(topic);
2240        assert_eq!(subscriptions.pending_subscribe_topics(), vec![topic]);
2241
2242        subscriptions.confirm_unsubscribe(topic);
2243        assert!(subscriptions.pending_unsubscribe_topics().is_empty());
2244        assert_eq!(subscriptions.pending_subscribe_topics(), vec![topic]);
2245
2246        subscriptions.confirm_subscribe(topic);
2247        assert_eq!(subscriptions.len(), 1);
2248        assert!(subscriptions.pending_subscribe_topics().is_empty());
2249
2250        let all = subscriptions.all_topics();
2251        assert_eq!(all.len(), 1);
2252        assert!(all.contains(&topic.to_string()));
2253    }
2254
2255    #[rstest]
2256    fn test_race_late_subscribe_confirmation_after_unsubscribe() {
2257        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);
2258        let topic = "tickers.ETHUSDT";
2259
2260        subscriptions.mark_subscribe(topic);
2261        assert_eq!(subscriptions.pending_subscribe_topics(), vec![topic]);
2262
2263        subscriptions.mark_unsubscribe(topic);
2264        assert!(subscriptions.pending_subscribe_topics().is_empty());
2265        assert_eq!(subscriptions.pending_unsubscribe_topics(), vec![topic]);
2266
2267        subscriptions.confirm_subscribe(topic);
2268        assert_eq!(subscriptions.len(), 0);
2269        assert_eq!(subscriptions.pending_unsubscribe_topics(), vec![topic]);
2270
2271        subscriptions.confirm_unsubscribe(topic);
2272
2273        assert!(subscriptions.is_empty());
2274        assert!(subscriptions.all_topics().is_empty());
2275    }
2276
2277    #[rstest]
2278    fn test_race_reconnection_with_pending_states() {
2279        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);
2280
2281        let trade_btc = "publicTrade.BTCUSDT";
2282        subscriptions.mark_subscribe(trade_btc);
2283        subscriptions.confirm_subscribe(trade_btc);
2284
2285        let trade_eth = "publicTrade.ETHUSDT";
2286        subscriptions.mark_subscribe(trade_eth);
2287
2288        let book_btc = "orderbook.50.BTCUSDT";
2289        subscriptions.mark_subscribe(book_btc);
2290        subscriptions.confirm_subscribe(book_btc);
2291        subscriptions.mark_unsubscribe(book_btc);
2292
2293        let topics_to_restore = subscriptions.all_topics();
2294
2295        assert_eq!(topics_to_restore.len(), 2);
2296        assert!(topics_to_restore.contains(&trade_btc.to_string()));
2297        assert!(topics_to_restore.contains(&trade_eth.to_string()));
2298        assert!(!topics_to_restore.contains(&book_btc.to_string()));
2299    }
2300
2301    #[tokio::test]
2302    async fn option_limit_counts_pending_unsubscriptions() {
2303        let client = BybitWebSocketClient::new_public_with(
2304            BybitProductType::Option,
2305            BybitEnvironment::Mainnet,
2306            Some("ws://option-pending-limit.invalid/v5/public/option".to_string()),
2307            20,
2308            TransportBackend::default(),
2309            None,
2310        );
2311
2312        for index in 0..BYBIT_OPTION_SUBSCRIPTION_LIMIT {
2313            let topic = format!("tickers.OPTION-{index}");
2314            assert!(client.subscriptions.add_reference(&topic));
2315            client.subscriptions.mark_subscribe(&topic);
2316            client.subscriptions.confirm_subscribe(&topic);
2317        }
2318        let pending = "tickers.OPTION-0";
2319        assert!(client.subscriptions.remove_reference(pending));
2320        client.subscriptions.mark_unsubscribe(pending);
2321
2322        let new_topic = "tickers.OPTION-new";
2323        let error = client
2324            .subscribe(vec![new_topic.to_string()])
2325            .await
2326            .unwrap_err();
2327
2328        assert!(error.to_string().contains("2000 arguments"));
2329        assert_eq!(client.subscriptions.get_reference_count(new_topic), 0);
2330        assert_eq!(
2331            client.subscriptions.pending_unsubscribe_topics(),
2332            vec![pending]
2333        );
2334    }
2335
2336    #[tokio::test]
2337    async fn batch_chunks_enter_handler_atomically() {
2338        let client = BybitWebSocketClient::new_trade(
2339            BybitEnvironment::Testnet,
2340            Some("test-key".to_string()),
2341            Some("test-secret".to_string()),
2342            None,
2343            20,
2344            TransportBackend::default(),
2345            None,
2346        );
2347        client
2348            .connection_mode
2349            .load()
2350            .store(ConnectionMode::Active.as_u8(), Ordering::Release);
2351        client.auth_tracker.succeed();
2352        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel();
2353        *client.cmd_tx.write().await = cmd_tx;
2354        let orders = (0..21)
2355            .map(|index| BybitWsCancelOrderParams {
2356                category: BybitProductType::Linear,
2357                symbol: Ustr::from("BTCUSDT"),
2358                order_id: Some(format!("order-{index}")),
2359                order_link_id: Some(format!("client-order-{index}")),
2360            })
2361            .collect::<Vec<_>>();
2362        let req_ids =
2363            BybitWebSocketClient::batch_request_ids(BybitProductType::Linear, orders.len());
2364
2365        client
2366            .batch_cancel_orders_with_ids(orders, req_ids.clone())
2367            .await
2368            .unwrap();
2369
2370        let command = cmd_rx.recv().await.expect("expected batch command");
2371        let HandlerCommand::SendOrders { commands } = command else {
2372            panic!("expected atomic batch command, was {command:?}");
2373        };
2374        assert_eq!(commands.len(), 3);
2375        assert_eq!(
2376            commands
2377                .iter()
2378                .map(|command| command.req_id.as_str())
2379                .collect::<Vec<_>>(),
2380            req_ids.iter().map(String::as_str).collect::<Vec<_>>()
2381        );
2382        assert_eq!(
2383            commands
2384                .iter()
2385                .map(|command| command.weight)
2386                .collect::<Vec<_>>(),
2387            vec![10, 10, 1]
2388        );
2389        assert!(cmd_rx.try_recv().is_err());
2390    }
2391
2392    #[tokio::test]
2393    async fn option_batch_chunks_preserve_request_correlation() {
2394        let client = BybitWebSocketClient::new_trade(
2395            BybitEnvironment::Testnet,
2396            Some("test-key".to_string()),
2397            Some("test-secret".to_string()),
2398            None,
2399            20,
2400            TransportBackend::default(),
2401            None,
2402        );
2403        client
2404            .connection_mode
2405            .load()
2406            .store(ConnectionMode::Active.as_u8(), Ordering::Release);
2407        client.auth_tracker.succeed();
2408        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel();
2409        *client.cmd_tx.write().await = cmd_tx;
2410
2411        let place_template = BybitWsPlaceOrderParams {
2412            category: BybitProductType::Option,
2413            symbol: Ustr::from("BTC-30JUN25-100000-C"),
2414            side: BybitOrderSide::Buy,
2415            order_type: BybitOrderType::Limit,
2416            qty: "0.1".to_string(),
2417            is_leverage: None,
2418            market_unit: None,
2419            price: Some("500".to_string()),
2420            time_in_force: Some(BybitTimeInForce::Gtc),
2421            order_link_id: None,
2422            reduce_only: None,
2423            close_on_trigger: None,
2424            trigger_price: None,
2425            trigger_by: None,
2426            trigger_direction: None,
2427            tpsl_mode: None,
2428            take_profit: None,
2429            stop_loss: None,
2430            tp_trigger_by: None,
2431            sl_trigger_by: None,
2432            sl_trigger_price: None,
2433            tp_trigger_price: None,
2434            sl_order_type: None,
2435            tp_order_type: None,
2436            sl_limit_price: None,
2437            tp_limit_price: None,
2438            order_iv: Some("0.80".to_string()),
2439            mmp: Some(true),
2440            position_idx: None,
2441            bbo_side_type: None,
2442            bbo_level: None,
2443        };
2444        let place_order_link_ids = (0..6)
2445            .map(|index| format!("option-place-{index}"))
2446            .collect::<Vec<_>>();
2447        let place_orders = place_order_link_ids
2448            .iter()
2449            .map(|order_link_id| BybitWsPlaceOrderParams {
2450                order_link_id: Some(order_link_id.clone()),
2451                ..place_template.clone()
2452            })
2453            .collect::<Vec<_>>();
2454        let place_req_ids =
2455            BybitWebSocketClient::batch_request_ids(BybitProductType::Option, place_orders.len());
2456
2457        client
2458            .batch_place_orders_with_ids(place_orders, place_req_ids.clone())
2459            .await
2460            .unwrap();
2461
2462        let command = cmd_rx.recv().await.expect("expected place batch command");
2463        let HandlerCommand::SendOrders { commands } = command else {
2464            panic!("expected atomic place batch command, was {command:?}");
2465        };
2466        assert_option_batch_commands(
2467            &commands,
2468            &place_req_ids,
2469            BybitWsOrderRequestOp::CreateBatch,
2470            &place_order_link_ids,
2471            batch_nested_items,
2472        );
2473
2474        let amend_template = BybitWsAmendOrderParams {
2475            category: BybitProductType::Option,
2476            symbol: Ustr::from("BTC-30JUN25-100000-C"),
2477            order_id: Some("venue-option-amend".to_string()),
2478            order_link_id: None,
2479            qty: Some("0.23".to_string()),
2480            price: Some("510.5".to_string()),
2481            trigger_price: Some("505.5".to_string()),
2482            take_profit: Some("530.5".to_string()),
2483            stop_loss: Some("490.5".to_string()),
2484            tp_trigger_by: Some(crate::common::enums::BybitTriggerType::MarkPrice),
2485            sl_trigger_by: Some(crate::common::enums::BybitTriggerType::IndexPrice),
2486            order_iv: Some("0.91".to_string()),
2487        };
2488        let amend_order_link_ids = (0..6)
2489            .map(|index| format!("option-amend-{index}"))
2490            .collect::<Vec<_>>();
2491        let amend_orders = amend_order_link_ids
2492            .iter()
2493            .map(|order_link_id| BybitWsAmendOrderParams {
2494                order_link_id: Some(order_link_id.clone()),
2495                ..amend_template.clone()
2496            })
2497            .collect::<Vec<_>>();
2498        let amend_req_ids =
2499            BybitWebSocketClient::batch_request_ids(BybitProductType::Option, amend_orders.len());
2500
2501        client
2502            .batch_amend_orders_with_ids(amend_orders, amend_req_ids.clone())
2503            .await
2504            .unwrap();
2505
2506        let command = cmd_rx.recv().await.expect("expected amend batch command");
2507        let HandlerCommand::SendOrders { commands } = command else {
2508            panic!("expected atomic amend batch command, was {command:?}");
2509        };
2510        assert_option_batch_commands(
2511            &commands,
2512            &amend_req_ids,
2513            BybitWsOrderRequestOp::AmendBatch,
2514            &amend_order_link_ids,
2515            batch_nested_items,
2516        );
2517        assert!(commands.iter().all(|command| command.args.len() == 1));
2518        assert!(
2519            commands
2520                .iter()
2521                .all(|command| command.args[0]["category"] == "option")
2522        );
2523        assert_eq!(
2524            commands[0].args[0]["request"][0],
2525            serde_json::json!({
2526                "symbol": "BTC-30JUN25-100000-C",
2527                "orderId": "venue-option-amend",
2528                "orderLinkId": "option-amend-0",
2529                "qty": "0.23",
2530                "price": "510.5",
2531                "triggerPrice": "505.5",
2532                "takeProfit": "530.5",
2533                "stopLoss": "490.5",
2534                "tpTriggerBy": "MarkPrice",
2535                "slTriggerBy": "IndexPrice",
2536                "orderIv": "0.91",
2537            })
2538        );
2539        assert!(
2540            commands
2541                .iter()
2542                .flat_map(batch_nested_items)
2543                .all(|order| order.get("category").is_none())
2544        );
2545
2546        let cancel_order_link_ids = (0..6)
2547            .map(|index| format!("option-cancel-{index}"))
2548            .collect::<Vec<_>>();
2549        let cancel_orders = cancel_order_link_ids
2550            .iter()
2551            .enumerate()
2552            .map(|(index, order_link_id)| BybitWsCancelOrderParams {
2553                category: BybitProductType::Option,
2554                symbol: Ustr::from("BTC-30JUN25-100000-C"),
2555                order_id: Some(format!("venue-option-{index}")),
2556                order_link_id: Some(order_link_id.clone()),
2557            })
2558            .collect::<Vec<_>>();
2559        let cancel_req_ids =
2560            BybitWebSocketClient::batch_request_ids(BybitProductType::Option, cancel_orders.len());
2561
2562        client
2563            .batch_cancel_orders_with_ids(cancel_orders, cancel_req_ids.clone())
2564            .await
2565            .unwrap();
2566
2567        let command = cmd_rx.recv().await.expect("expected cancel batch command");
2568        let HandlerCommand::SendOrders { commands } = command else {
2569            panic!("expected atomic cancel batch command, was {command:?}");
2570        };
2571        assert_option_batch_commands(
2572            &commands,
2573            &cancel_req_ids,
2574            BybitWsOrderRequestOp::CancelBatch,
2575            &cancel_order_link_ids,
2576            batch_nested_items,
2577        );
2578        assert!(cmd_rx.try_recv().is_err());
2579    }
2580
2581    fn assert_option_batch_commands(
2582        commands: &[BybitWsOrderCommand],
2583        req_ids: &[String],
2584        op: BybitWsOrderRequestOp,
2585        order_link_ids: &[String],
2586        items: for<'a> fn(&'a BybitWsOrderCommand) -> &'a [Value],
2587    ) {
2588        assert_eq!(req_ids.len(), 2);
2589        assert_ne!(req_ids[0], req_ids[1]);
2590        assert_eq!(commands.len(), 2);
2591        assert_eq!(
2592            commands
2593                .iter()
2594                .map(|command| command.req_id.as_str())
2595                .collect::<Vec<_>>(),
2596            req_ids.iter().map(String::as_str).collect::<Vec<_>>()
2597        );
2598        assert!(commands.iter().all(|command| command.op == op));
2599        assert_eq!(
2600            commands
2601                .iter()
2602                .map(|command| items(command).len())
2603                .collect::<Vec<_>>(),
2604            vec![5, 1]
2605        );
2606        assert_eq!(
2607            commands
2608                .iter()
2609                .flat_map(items)
2610                .map(|order| order["orderLinkId"].as_str().unwrap().to_string())
2611                .collect::<Vec<_>>(),
2612            order_link_ids
2613        );
2614        assert!(commands.iter().all(|command| command.weight == 1));
2615    }
2616
2617    fn batch_nested_items(command: &BybitWsOrderCommand) -> &[Value] {
2618        command.args[0]["request"].as_array().unwrap()
2619    }
2620
2621    #[rstest]
2622    fn test_race_duplicate_subscribe_messages_idempotent() {
2623        let subscriptions = SubscriptionState::new(BYBIT_WS_TOPIC_DELIMITER);
2624        let topic = "publicTrade.BTCUSDT";
2625
2626        subscriptions.mark_subscribe(topic);
2627        subscriptions.confirm_subscribe(topic);
2628        assert_eq!(subscriptions.len(), 1);
2629
2630        subscriptions.mark_subscribe(topic);
2631        assert!(subscriptions.pending_subscribe_topics().is_empty());
2632        assert_eq!(subscriptions.len(), 1);
2633
2634        subscriptions.confirm_subscribe(topic);
2635        assert_eq!(subscriptions.len(), 1);
2636
2637        let all = subscriptions.all_topics();
2638        assert_eq!(all.len(), 1);
2639        assert_eq!(all[0], topic);
2640    }
2641
2642    #[rstest]
2643    #[case::spot_with_leverage(BybitProductType::Spot, true, Some(1))]
2644    #[case::spot_without_leverage(BybitProductType::Spot, false, Some(0))]
2645    #[case::linear_with_leverage(BybitProductType::Linear, true, None)]
2646    #[case::linear_without_leverage(BybitProductType::Linear, false, None)]
2647    #[case::inverse_with_leverage(BybitProductType::Inverse, true, None)]
2648    #[case::option_with_leverage(BybitProductType::Option, true, None)]
2649    fn test_is_leverage_parameter(
2650        #[case] product_type: BybitProductType,
2651        #[case] is_leverage: bool,
2652        #[case] expected: Option<i32>,
2653    ) {
2654        let symbol = match product_type {
2655            BybitProductType::Spot => "BTCUSDT-SPOT.BYBIT",
2656            BybitProductType::Linear => "ETHUSDT-LINEAR.BYBIT",
2657            BybitProductType::Inverse => "BTCUSD-INVERSE.BYBIT",
2658            BybitProductType::Option => "BTC-31MAY24-50000-C-OPTION.BYBIT",
2659        };
2660
2661        let instrument_id = InstrumentId::from(symbol);
2662        let client_order_id = ClientOrderId::from("test-order-1");
2663        let quantity = Quantity::from("1.0");
2664
2665        let client = BybitWebSocketClient::new_trade(
2666            BybitEnvironment::Testnet,
2667            Some("test-key".to_string()),
2668            Some("test-secret".to_string()),
2669            None,
2670            20,
2671            TransportBackend::default(),
2672            None,
2673        );
2674
2675        let params = client
2676            .build_place_order_params(
2677                product_type,
2678                instrument_id,
2679                client_order_id,
2680                OrderSide::Buy,
2681                OrderType::Limit,
2682                quantity,
2683                false,
2684                Some(TimeInForce::Gtc),
2685                Some(Price::from("50000.0")),
2686                None,
2687                None,
2688                None,
2689                None,
2690                is_leverage,
2691                None,
2692                None,
2693                None,
2694                None,
2695                None,
2696            )
2697            .expect("Failed to build params");
2698
2699        assert_eq!(params.is_leverage, expected);
2700    }
2701
2702    #[rstest]
2703    #[case::spot_market_quote_quantity(
2704        BybitProductType::Spot,
2705        OrderType::Market,
2706        true,
2707        Some(BybitMarketUnit::QuoteCoin)
2708    )]
2709    #[case::spot_market_base_quantity(
2710        BybitProductType::Spot,
2711        OrderType::Market,
2712        false,
2713        Some(BybitMarketUnit::BaseCoin)
2714    )]
2715    #[case::spot_limit_no_unit(BybitProductType::Spot, OrderType::Limit, false, None)]
2716    #[case::spot_limit_quote(BybitProductType::Spot, OrderType::Limit, true, None)]
2717    #[case::linear_market_no_unit(BybitProductType::Linear, OrderType::Market, false, None)]
2718    #[case::inverse_market_no_unit(BybitProductType::Inverse, OrderType::Market, true, None)]
2719    fn test_is_quote_quantity_parameter(
2720        #[case] product_type: BybitProductType,
2721        #[case] order_type: OrderType,
2722        #[case] is_quote_quantity: bool,
2723        #[case] expected: Option<BybitMarketUnit>,
2724    ) {
2725        let symbol = match product_type {
2726            BybitProductType::Spot => "BTCUSDT-SPOT.BYBIT",
2727            BybitProductType::Linear => "ETHUSDT-LINEAR.BYBIT",
2728            BybitProductType::Inverse => "BTCUSD-INVERSE.BYBIT",
2729            BybitProductType::Option => "BTC-31MAY24-50000-C-OPTION.BYBIT",
2730        };
2731
2732        let instrument_id = InstrumentId::from(symbol);
2733        let client_order_id = ClientOrderId::from("test-order-1");
2734        let quantity = Quantity::from("1.0");
2735
2736        let client = BybitWebSocketClient::new_trade(
2737            BybitEnvironment::Testnet,
2738            Some("test-key".to_string()),
2739            Some("test-secret".to_string()),
2740            None,
2741            20,
2742            TransportBackend::default(),
2743            None,
2744        );
2745
2746        let params = client
2747            .build_place_order_params(
2748                product_type,
2749                instrument_id,
2750                client_order_id,
2751                OrderSide::Buy,
2752                order_type,
2753                quantity,
2754                is_quote_quantity,
2755                Some(TimeInForce::Gtc),
2756                if order_type == OrderType::Market {
2757                    None
2758                } else {
2759                    Some(Price::from("50000.0"))
2760                },
2761                None,
2762                None,
2763                None,
2764                None,
2765                false,
2766                None,
2767                None,
2768                None,
2769                None,
2770                None,
2771            )
2772            .expect("Failed to build params");
2773
2774        assert_eq!(params.market_unit, expected);
2775    }
2776
2777    #[rstest]
2778    fn test_build_place_order_params_with_bbo_omits_price() {
2779        let client = BybitWebSocketClient::new_trade(
2780            BybitEnvironment::Testnet,
2781            Some("test-key".to_string()),
2782            Some("test-secret".to_string()),
2783            None,
2784            20,
2785            TransportBackend::default(),
2786            None,
2787        );
2788
2789        let params = client
2790            .build_place_order_params(
2791                BybitProductType::Linear,
2792                InstrumentId::from("ETHUSDT-LINEAR.BYBIT"),
2793                ClientOrderId::from("test-bbo-order-1"),
2794                OrderSide::Buy,
2795                OrderType::Limit,
2796                Quantity::from("1.0"),
2797                false,
2798                Some(TimeInForce::Gtc),
2799                Some(Price::from("50000.0")),
2800                None,
2801                None,
2802                None,
2803                None,
2804                false,
2805                None,
2806                None,
2807                None,
2808                Some(BybitBboSideType::Queue),
2809                Some("2".to_string()),
2810            )
2811            .expect("Failed to build params");
2812
2813        assert_eq!(params.price, None);
2814        assert_eq!(params.bbo_side_type, Some(BybitBboSideType::Queue));
2815        assert_eq!(params.bbo_level.as_deref(), Some("2"));
2816    }
2817}
nautilus_bybit/websocket/client.rs

nautilus_bybit/websocket/
client.rs
