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

nautilus_bybit/websocket/
client.rs
