nautilus_bybit/python/
websocket.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//! Python bindings for the Bybit WebSocket client.
17
18use std::sync::Arc;
19
20use ahash::AHashMap;
21use dashmap::DashMap;
22use futures_util::StreamExt;
23use nautilus_common::live::get_runtime;
24use nautilus_core::{
25    AtomicMap, AtomicSet, UUID4, UnixNanos,
26    python::{call_python_threadsafe, to_pyruntime_err, to_pyvalue_err},
27    time::{AtomicTime, get_atomic_clock_realtime},
28};
29use nautilus_model::{
30    data::{BarType, Data, OrderBookDeltas_API, QuoteTick},
31    enums::{
32        AggregationSource, BarAggregation, OrderSide, OrderType, PriceType, TimeInForce,
33        TriggerType,
34    },
35    events::{OrderCancelRejected, OrderModifyRejected, OrderRejected},
36    identifiers::{
37        AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TraderId, VenueOrderId,
38    },
39    instruments::{Instrument, InstrumentAny},
40    python::{data::data_to_pycapsule, instruments::pyobject_to_instrument_any},
41    types::{Price, Quantity},
42};
43use nautilus_network::websocket::TransportBackend;
44use pyo3::{IntoPyObjectExt, prelude::*};
45use ustr::Ustr;
46
47use crate::{
48    common::{
49        consts::BYBIT_VENUE,
50        enums::{BybitEnvironment, BybitPositionIdx, BybitProductType},
51        parse::{make_bybit_symbol, parse_bbo_level, parse_bbo_side_type},
52    },
53    python::params::{BybitWsAmendOrderParams, BybitWsCancelOrderParams, BybitWsPlaceOrderParams},
54    websocket::{
55        client::{BATCH_PROCESSING_LIMIT, BybitWebSocketClient, PendingPyRequest},
56        dispatch::PendingOperation,
57        messages::{BybitWebSocketError, BybitWsMessage},
58        parse::{
59            parse_kline_topic, parse_millis_i64, parse_orderbook_deltas, parse_orderbook_quote,
60            parse_ticker_linear_funding, parse_ticker_linear_index_price,
61            parse_ticker_linear_mark_price, parse_ticker_linear_quote, parse_ticker_option_greeks,
62            parse_ticker_option_index_price, parse_ticker_option_mark_price,
63            parse_ticker_option_quote, parse_ws_account_state, parse_ws_fill_report,
64            parse_ws_fill_report_fast, parse_ws_kline_bar, parse_ws_order_status_report,
65            parse_ws_position_status_report, parse_ws_trade_tick,
66        },
67    },
68};
69
70fn validate_bar_type(bar_type: &BarType) -> anyhow::Result<()> {
71    let spec = bar_type.spec();
72
73    if spec.price_type != PriceType::Last {
74        anyhow::bail!(
75            "Invalid bar type: Bybit bars only support LAST price type, received {}",
76            spec.price_type
77        );
78    }
79
80    if bar_type.aggregation_source() != AggregationSource::External {
81        anyhow::bail!(
82            "Invalid bar type: Bybit bars only support EXTERNAL aggregation source, received {}",
83            bar_type.aggregation_source()
84        );
85    }
86
87    let step = spec.step.get();
88    if spec.aggregation == BarAggregation::Minute && step >= 60 {
89        let hours = step / 60;
90        anyhow::bail!("Invalid bar type: {step}-MINUTE not supported, use {hours}-HOUR instead");
91    }
92
93    Ok(())
94}
95
96#[pymethods]
97#[pyo3_stub_gen::derive::gen_stub_pymethods]
98impl BybitWebSocketError {
99    fn __repr__(&self) -> String {
100        format!(
101            "BybitWebSocketError(code={}, message='{}', conn_id={:?}, topic={:?})",
102            self.code, self.message, self.conn_id, self.topic
103        )
104    }
105
106    #[getter]
107    pub fn code(&self) -> i64 {
108        self.code
109    }
110
111    #[getter]
112    pub fn message(&self) -> &str {
113        &self.message
114    }
115
116    #[getter]
117    pub fn conn_id(&self) -> Option<&str> {
118        self.conn_id.as_deref()
119    }
120
121    #[getter]
122    pub fn topic(&self) -> Option<&str> {
123        self.topic.as_deref()
124    }
125
126    #[getter]
127    pub fn req_id(&self) -> Option<&str> {
128        self.req_id.as_deref()
129    }
130}
131
132#[pymethods]
133#[pyo3_stub_gen::derive::gen_stub_pymethods]
134impl BybitWebSocketClient {
135    /// Creates a new Bybit public WebSocket client.
136    #[staticmethod]
137    #[pyo3(name = "new_public")]
138    #[pyo3(signature = (product_type, environment, url=None, heartbeat=20, proxy_url=None))]
139    fn py_new_public(
140        product_type: BybitProductType,
141        environment: BybitEnvironment,
142        url: Option<String>,
143        heartbeat: u64,
144        proxy_url: Option<String>,
145    ) -> Self {
146        Self::new_public_with(
147            product_type,
148            environment,
149            url,
150            heartbeat,
151            TransportBackend::default(),
152            proxy_url,
153        )
154    }
155
156    /// Creates a new Bybit private WebSocket client.
157    ///
158    /// If `api_key` or `api_secret` are not provided, they will be loaded from
159    /// environment variables based on the environment:
160    /// - Demo: `BYBIT_DEMO_API_KEY`, `BYBIT_DEMO_API_SECRET`
161    /// - Testnet: `BYBIT_TESTNET_API_KEY`, `BYBIT_TESTNET_API_SECRET`
162    /// - Mainnet: `BYBIT_API_KEY`, `BYBIT_API_SECRET`
163    #[staticmethod]
164    #[pyo3(name = "new_private")]
165    #[pyo3(signature = (environment, api_key=None, api_secret=None, url=None, heartbeat=20, proxy_url=None))]
166    fn py_new_private(
167        environment: BybitEnvironment,
168        api_key: Option<String>,
169        api_secret: Option<String>,
170        url: Option<String>,
171        heartbeat: u64,
172        proxy_url: Option<String>,
173    ) -> Self {
174        Self::new_private(
175            environment,
176            api_key,
177            api_secret,
178            url,
179            heartbeat,
180            TransportBackend::default(),
181            proxy_url,
182        )
183    }
184
185    /// Creates a new Bybit trade WebSocket client for order operations.
186    ///
187    /// If `api_key` or `api_secret` are not provided, they will be loaded from
188    /// environment variables based on the environment:
189    /// - Demo: `BYBIT_DEMO_API_KEY`, `BYBIT_DEMO_API_SECRET`
190    /// - Testnet: `BYBIT_TESTNET_API_KEY`, `BYBIT_TESTNET_API_SECRET`
191    /// - Mainnet: `BYBIT_API_KEY`, `BYBIT_API_SECRET`
192    #[staticmethod]
193    #[pyo3(name = "new_trade")]
194    #[pyo3(signature = (environment, api_key=None, api_secret=None, url=None, heartbeat=20, proxy_url=None))]
195    fn py_new_trade(
196        environment: BybitEnvironment,
197        api_key: Option<String>,
198        api_secret: Option<String>,
199        url: Option<String>,
200        heartbeat: u64,
201        proxy_url: Option<String>,
202    ) -> Self {
203        Self::new_trade(
204            environment,
205            api_key,
206            api_secret,
207            url,
208            heartbeat,
209            TransportBackend::default(),
210            proxy_url,
211        )
212    }
213
214    #[getter]
215    #[pyo3(name = "api_key_masked")]
216    #[must_use]
217    pub fn py_api_key_masked(&self) -> Option<String> {
218        self.credential().map(|c| c.api_key_masked())
219    }
220
221    /// Returns a value indicating whether the client is active.
222    #[pyo3(name = "is_active")]
223    fn py_is_active(&self) -> bool {
224        self.is_active()
225    }
226
227    /// Returns a value indicating whether the client is closed.
228    #[pyo3(name = "is_closed")]
229    fn py_is_closed(&self) -> bool {
230        self.is_closed()
231    }
232
233    /// Returns the number of currently registered subscriptions.
234    #[pyo3(name = "subscription_count")]
235    fn py_subscription_count(&self) -> usize {
236        self.subscription_count()
237    }
238
239    /// Adds an instrument to the shared instruments cache.
240    #[pyo3(name = "cache_instrument")]
241    fn py_cache_instrument(&self, py: Python<'_>, instrument: Py<PyAny>) -> PyResult<()> {
242        self.cache_instrument(pyobject_to_instrument_any(py, instrument)?);
243        Ok(())
244    }
245
246    /// Sets the account ID for account message parsing.
247    #[pyo3(name = "set_account_id")]
248    fn py_set_account_id(&mut self, account_id: AccountId) {
249        self.set_account_id(account_id);
250    }
251
252    /// Sets the account market maker level.
253    #[pyo3(name = "set_mm_level")]
254    fn py_set_mm_level(&self, mm_level: u8) {
255        self.set_mm_level(mm_level);
256    }
257
258    /// Sets whether bar timestamps use the close time.
259    #[pyo3(name = "set_bars_timestamp_on_close")]
260    fn py_set_bars_timestamp_on_close(&self, value: bool) {
261        self.set_bars_timestamp_on_close(value);
262    }
263
264    /// Adds an instrument ID to the option greeks subscription set.
265    #[pyo3(name = "add_option_greeks_sub")]
266    fn py_add_option_greeks_sub(&self, instrument_id: InstrumentId) {
267        self.add_option_greeks_sub(instrument_id);
268    }
269
270    /// Removes an instrument ID from the option greeks subscription set.
271    #[pyo3(name = "remove_option_greeks_sub")]
272    fn py_remove_option_greeks_sub(&self, instrument_id: InstrumentId) {
273        self.remove_option_greeks_sub(&instrument_id);
274    }
275
276    /// Disconnects the WebSocket client and stops the background task.
277    #[pyo3(name = "connect")]
278    #[expect(clippy::needless_pass_by_value)] // PyO3 extracted parameter
279    fn py_connect<'py>(
280        &mut self,
281        py: Python<'py>,
282        loop_: Py<PyAny>,
283        callback: Py<PyAny>,
284    ) -> PyResult<Bound<'py, PyAny>> {
285        let call_soon: Py<PyAny> = loop_.getattr(py, "call_soon_threadsafe")?;
286        let mut client = self.clone();
287
288        pyo3_async_runtimes::tokio::future_into_py(py, async move {
289            client.connect().await.map_err(to_pyruntime_err)?;
290
291            let stream = client.stream();
292            let clock = get_atomic_clock_realtime();
293            let product_type = client.product_type();
294            let account_id = client.account_id();
295            let bar_types_cache = client.bar_types_cache().clone();
296            let trade_subs = client.trade_subs().clone();
297            let option_greeks_subs = client.option_greeks_subs().clone();
298            let bars_timestamp_on_close = client.bars_timestamp_on_close();
299            let instruments = Arc::clone(client.instruments_cache_ref());
300            let pending_py_requests = Arc::clone(client.pending_py_requests());
301
302            get_runtime().spawn(async move {
303                let mut quote_cache = AHashMap::new();
304                let mut funding_cache: AHashMap<Ustr, (Option<String>, Option<String>)> =
305                    AHashMap::new();
306                let _client = client;
307
308                tokio::pin!(stream);
309
310                while let Some(msg) = stream.next().await {
311                    match msg {
312                        BybitWsMessage::Orderbook(ref msg) => {
313                            handle_orderbook(
314                                msg,
315                                product_type,
316                                &instruments,
317                                &mut quote_cache,
318                                clock,
319                                &call_soon,
320                                &callback,
321                            );
322                        }
323                        BybitWsMessage::Trade(ref msg) => {
324                            handle_trade(
325                                msg,
326                                product_type,
327                                &instruments,
328                                &trade_subs,
329                                clock,
330                                &call_soon,
331                                &callback,
332                            );
333                        }
334                        BybitWsMessage::Kline(ref msg) => {
335                            handle_kline(
336                                msg,
337                                product_type,
338                                &instruments,
339                                &bar_types_cache,
340                                bars_timestamp_on_close,
341                                clock,
342                                &call_soon,
343                                &callback,
344                            );
345                        }
346                        BybitWsMessage::TickerLinear(ref msg) => {
347                            handle_ticker_linear(
348                                msg,
349                                product_type,
350                                &instruments,
351                                &mut quote_cache,
352                                &mut funding_cache,
353                                clock,
354                                &call_soon,
355                                &callback,
356                            );
357                        }
358                        BybitWsMessage::TickerOption(ref msg) => {
359                            handle_ticker_option(
360                                msg,
361                                product_type,
362                                &instruments,
363                                &mut quote_cache,
364                                &option_greeks_subs,
365                                clock,
366                                &call_soon,
367                                &callback,
368                            );
369                        }
370                        BybitWsMessage::AccountOrder(ref msg) => {
371                            handle_account_order(
372                                msg,
373                                &instruments,
374                                account_id,
375                                clock,
376                                &call_soon,
377                                &callback,
378                            );
379                        }
380                        BybitWsMessage::AccountExecution(ref msg) => {
381                            handle_account_execution(
382                                msg,
383                                &instruments,
384                                account_id,
385                                clock,
386                                &call_soon,
387                                &callback,
388                            );
389                        }
390                        BybitWsMessage::AccountExecutionFast(ref msg) => {
391                            handle_account_execution_fast(
392                                msg,
393                                &instruments,
394                                account_id,
395                                clock,
396                                &call_soon,
397                                &callback,
398                            );
399                        }
400                        BybitWsMessage::AccountWallet(ref msg) => {
401                            handle_account_wallet(msg, account_id, clock, &call_soon, &callback);
402                        }
403                        BybitWsMessage::AccountPosition(ref msg) => {
404                            handle_account_position(
405                                msg,
406                                &instruments,
407                                account_id,
408                                clock,
409                                &call_soon,
410                                &callback,
411                            );
412                        }
413                        BybitWsMessage::OrderResponse(ref resp) => {
414                            handle_order_response(
415                                resp,
416                                &pending_py_requests,
417                                account_id,
418                                clock,
419                                &call_soon,
420                                &callback,
421                            );
422                        }
423                        BybitWsMessage::Error(err) => {
424                            send_to_python(err, &call_soon, &callback);
425                        }
426                        BybitWsMessage::Reconnected => {
427                            quote_cache.clear();
428                            funding_cache.clear();
429                            log::info!("WebSocket reconnected");
430                        }
431                        BybitWsMessage::Auth(_) => {
432                            log::debug!("WebSocket authenticated");
433                        }
434                    }
435                }
436            });
437
438            Ok(())
439        })
440    }
441
442    #[pyo3(name = "close")]
443    fn py_close<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
444        let mut client = self.clone();
445
446        pyo3_async_runtimes::tokio::future_into_py(py, async move {
447            if let Err(e) = client.close().await {
448                log::warn!("Error on close: {e}");
449            }
450            Ok(())
451        })
452    }
453
454    /// Subscribe to the provided topic strings.
455    #[pyo3(name = "subscribe")]
456    fn py_subscribe<'py>(
457        &self,
458        py: Python<'py>,
459        topics: Vec<String>,
460    ) -> PyResult<Bound<'py, PyAny>> {
461        let client = self.clone();
462
463        pyo3_async_runtimes::tokio::future_into_py(py, async move {
464            client.subscribe(topics).await.map_err(to_pyruntime_err)?;
465            Ok(())
466        })
467    }
468
469    /// Unsubscribe from the provided topics.
470    #[pyo3(name = "unsubscribe")]
471    fn py_unsubscribe<'py>(
472        &self,
473        py: Python<'py>,
474        topics: Vec<String>,
475    ) -> PyResult<Bound<'py, PyAny>> {
476        let client = self.clone();
477
478        pyo3_async_runtimes::tokio::future_into_py(py, async move {
479            client.unsubscribe(topics).await.map_err(to_pyruntime_err)?;
480            Ok(())
481        })
482    }
483
484    /// Subscribes to orderbook updates for a specific instrument.
485    ///
486    /// # References
487    ///
488    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook>
489    #[pyo3(name = "subscribe_orderbook")]
490    fn py_subscribe_orderbook<'py>(
491        &self,
492        py: Python<'py>,
493        instrument_id: InstrumentId,
494        depth: u32,
495    ) -> PyResult<Bound<'py, PyAny>> {
496        let client = self.clone();
497
498        pyo3_async_runtimes::tokio::future_into_py(py, async move {
499            client
500                .subscribe_orderbook(instrument_id, depth)
501                .await
502                .map_err(to_pyruntime_err)?;
503            Ok(())
504        })
505    }
506
507    /// Unsubscribes from orderbook updates for a specific instrument.
508    #[pyo3(name = "unsubscribe_orderbook")]
509    fn py_unsubscribe_orderbook<'py>(
510        &self,
511        py: Python<'py>,
512        instrument_id: InstrumentId,
513        depth: u32,
514    ) -> PyResult<Bound<'py, PyAny>> {
515        let client = self.clone();
516
517        pyo3_async_runtimes::tokio::future_into_py(py, async move {
518            client
519                .unsubscribe_orderbook(instrument_id, depth)
520                .await
521                .map_err(to_pyruntime_err)?;
522            Ok(())
523        })
524    }
525
526    /// Subscribes to public trade updates for a specific instrument.
527    ///
528    /// # References
529    ///
530    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/trade>
531    #[pyo3(name = "subscribe_trades")]
532    fn py_subscribe_trades<'py>(
533        &self,
534        py: Python<'py>,
535        instrument_id: InstrumentId,
536    ) -> PyResult<Bound<'py, PyAny>> {
537        let client = self.clone();
538
539        pyo3_async_runtimes::tokio::future_into_py(py, async move {
540            client
541                .subscribe_trades(instrument_id)
542                .await
543                .map_err(to_pyruntime_err)?;
544            Ok(())
545        })
546    }
547
548    /// Unsubscribes from public trade updates for a specific instrument.
549    #[pyo3(name = "unsubscribe_trades")]
550    fn py_unsubscribe_trades<'py>(
551        &self,
552        py: Python<'py>,
553        instrument_id: InstrumentId,
554    ) -> PyResult<Bound<'py, PyAny>> {
555        let client = self.clone();
556
557        pyo3_async_runtimes::tokio::future_into_py(py, async move {
558            client
559                .unsubscribe_trades(instrument_id)
560                .await
561                .map_err(to_pyruntime_err)?;
562            Ok(())
563        })
564    }
565
566    /// Subscribes to ticker updates for a specific instrument.
567    ///
568    /// # References
569    ///
570    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/ticker>
571    #[pyo3(name = "subscribe_ticker")]
572    fn py_subscribe_ticker<'py>(
573        &self,
574        py: Python<'py>,
575        instrument_id: InstrumentId,
576    ) -> PyResult<Bound<'py, PyAny>> {
577        let client = self.clone();
578
579        pyo3_async_runtimes::tokio::future_into_py(py, async move {
580            client
581                .subscribe_ticker(instrument_id)
582                .await
583                .map_err(to_pyruntime_err)?;
584            Ok(())
585        })
586    }
587
588    #[pyo3(name = "subscribe_option_greeks")]
589    fn py_subscribe_option_greeks<'py>(
590        &self,
591        py: Python<'py>,
592        instrument_id: InstrumentId,
593    ) -> PyResult<Bound<'py, PyAny>> {
594        self.add_option_greeks_sub(instrument_id);
595        let client = self.clone();
596
597        pyo3_async_runtimes::tokio::future_into_py(py, async move {
598            client
599                .subscribe_ticker(instrument_id)
600                .await
601                .map_err(to_pyruntime_err)?;
602            Ok(())
603        })
604    }
605
606    #[pyo3(name = "unsubscribe_option_greeks")]
607    fn py_unsubscribe_option_greeks<'py>(
608        &self,
609        py: Python<'py>,
610        instrument_id: InstrumentId,
611    ) -> PyResult<Bound<'py, PyAny>> {
612        self.remove_option_greeks_sub(&instrument_id);
613        let client = self.clone();
614
615        pyo3_async_runtimes::tokio::future_into_py(py, async move {
616            client
617                .unsubscribe_ticker(instrument_id)
618                .await
619                .map_err(to_pyruntime_err)?;
620            Ok(())
621        })
622    }
623
624    /// Unsubscribes from ticker updates for a specific instrument.
625    #[pyo3(name = "unsubscribe_ticker")]
626    fn py_unsubscribe_ticker<'py>(
627        &self,
628        py: Python<'py>,
629        instrument_id: InstrumentId,
630    ) -> PyResult<Bound<'py, PyAny>> {
631        let client = self.clone();
632
633        pyo3_async_runtimes::tokio::future_into_py(py, async move {
634            client
635                .unsubscribe_ticker(instrument_id)
636                .await
637                .map_err(to_pyruntime_err)?;
638            Ok(())
639        })
640    }
641
642    /// Subscribes to kline/candlestick updates for a specific instrument.
643    ///
644    /// # References
645    ///
646    /// <https://bybit-exchange.github.io/docs/v5/websocket/public/kline>
647    #[pyo3(name = "subscribe_bars")]
648    fn py_subscribe_bars<'py>(
649        &self,
650        py: Python<'py>,
651        bar_type: BarType,
652    ) -> PyResult<Bound<'py, PyAny>> {
653        validate_bar_type(&bar_type).map_err(to_pyvalue_err)?;
654
655        let client = self.clone();
656        pyo3_async_runtimes::tokio::future_into_py(py, async move {
657            client
658                .subscribe_bars(bar_type)
659                .await
660                .map_err(to_pyruntime_err)?;
661            Ok(())
662        })
663    }
664
665    /// Unsubscribes from kline/candlestick updates for a specific instrument.
666    #[pyo3(name = "unsubscribe_bars")]
667    fn py_unsubscribe_bars<'py>(
668        &self,
669        py: Python<'py>,
670        bar_type: BarType,
671    ) -> PyResult<Bound<'py, PyAny>> {
672        validate_bar_type(&bar_type).map_err(to_pyvalue_err)?;
673
674        let client = self.clone();
675        pyo3_async_runtimes::tokio::future_into_py(py, async move {
676            client
677                .unsubscribe_bars(bar_type)
678                .await
679                .map_err(to_pyruntime_err)?;
680            Ok(())
681        })
682    }
683
684    /// Subscribes to order updates.
685    ///
686    /// # Errors
687    ///
688    /// Returns an error if the subscription request fails or if not authenticated.
689    ///
690    /// # References
691    ///
692    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/order>
693    #[pyo3(name = "subscribe_orders")]
694    fn py_subscribe_orders<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
695        let client = self.clone();
696
697        pyo3_async_runtimes::tokio::future_into_py(py, async move {
698            client.subscribe_orders().await.map_err(to_pyruntime_err)?;
699            Ok(())
700        })
701    }
702
703    /// Unsubscribes from order updates.
704    #[pyo3(name = "unsubscribe_orders")]
705    fn py_unsubscribe_orders<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
706        let client = self.clone();
707
708        pyo3_async_runtimes::tokio::future_into_py(py, async move {
709            client
710                .unsubscribe_orders()
711                .await
712                .map_err(to_pyruntime_err)?;
713            Ok(())
714        })
715    }
716
717    /// Subscribes to execution/fill updates.
718    ///
719    /// # Errors
720    ///
721    /// Returns an error if the subscription request fails or if not authenticated.
722    ///
723    /// # References
724    ///
725    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/execution>
726    #[pyo3(name = "subscribe_executions")]
727    fn py_subscribe_executions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
728        let client = self.clone();
729
730        pyo3_async_runtimes::tokio::future_into_py(py, async move {
731            client
732                .subscribe_executions()
733                .await
734                .map_err(to_pyruntime_err)?;
735            Ok(())
736        })
737    }
738
739    /// Unsubscribes from execution/fill updates.
740    #[pyo3(name = "unsubscribe_executions")]
741    fn py_unsubscribe_executions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
742        let client = self.clone();
743
744        pyo3_async_runtimes::tokio::future_into_py(py, async move {
745            client
746                .unsubscribe_executions()
747                .await
748                .map_err(to_pyruntime_err)?;
749            Ok(())
750        })
751    }
752
753    /// Subscribes to position updates.
754    ///
755    /// # Errors
756    ///
757    /// Returns an error if the subscription request fails or if not authenticated.
758    ///
759    /// # References
760    ///
761    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/position>
762    #[pyo3(name = "subscribe_positions")]
763    fn py_subscribe_positions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
764        let client = self.clone();
765
766        pyo3_async_runtimes::tokio::future_into_py(py, async move {
767            client
768                .subscribe_positions()
769                .await
770                .map_err(to_pyruntime_err)?;
771            Ok(())
772        })
773    }
774
775    /// Unsubscribes from position updates.
776    #[pyo3(name = "unsubscribe_positions")]
777    fn py_unsubscribe_positions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
778        let client = self.clone();
779
780        pyo3_async_runtimes::tokio::future_into_py(py, async move {
781            client
782                .unsubscribe_positions()
783                .await
784                .map_err(to_pyruntime_err)?;
785            Ok(())
786        })
787    }
788
789    /// Subscribes to wallet/balance updates.
790    ///
791    /// # Errors
792    ///
793    /// Returns an error if the subscription request fails or if not authenticated.
794    ///
795    /// # References
796    ///
797    /// <https://bybit-exchange.github.io/docs/v5/websocket/private/wallet>
798    #[pyo3(name = "subscribe_wallet")]
799    fn py_subscribe_wallet<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
800        let client = self.clone();
801
802        pyo3_async_runtimes::tokio::future_into_py(py, async move {
803            client.subscribe_wallet().await.map_err(to_pyruntime_err)?;
804            Ok(())
805        })
806    }
807
808    /// Unsubscribes from wallet/balance updates.
809    #[pyo3(name = "unsubscribe_wallet")]
810    fn py_unsubscribe_wallet<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
811        let client = self.clone();
812
813        pyo3_async_runtimes::tokio::future_into_py(py, async move {
814            client
815                .unsubscribe_wallet()
816                .await
817                .map_err(to_pyruntime_err)?;
818            Ok(())
819        })
820    }
821
822    /// Waits until the WebSocket client becomes active or times out.
823    #[pyo3(name = "wait_until_active")]
824    fn py_wait_until_active<'py>(
825        &self,
826        py: Python<'py>,
827        timeout_secs: f64,
828    ) -> PyResult<Bound<'py, PyAny>> {
829        let client = self.clone();
830
831        pyo3_async_runtimes::tokio::future_into_py(py, async move {
832            client
833                .wait_until_active(timeout_secs)
834                .await
835                .map_err(to_pyruntime_err)?;
836            Ok(())
837        })
838    }
839
840    /// Submits an order using Nautilus domain objects.
841    #[pyo3(name = "submit_order")]
842    #[pyo3(signature = (
843        product_type,
844        trader_id,
845        strategy_id,
846        instrument_id,
847        client_order_id,
848        order_side,
849        order_type,
850        quantity,
851        is_quote_quantity=false,
852        time_in_force=None,
853        price=None,
854        trigger_price=None,
855        trigger_type=None,
856        post_only=None,
857        reduce_only=None,
858        is_leverage=false,
859        position_idx=None,
860        bbo_side_type=None,
861        bbo_level=None,
862    ))]
863    #[expect(clippy::too_many_arguments)]
864    fn py_submit_order<'py>(
865        &self,
866        py: Python<'py>,
867        product_type: BybitProductType,
868        trader_id: TraderId,
869        strategy_id: StrategyId,
870        instrument_id: InstrumentId,
871        client_order_id: ClientOrderId,
872        order_side: OrderSide,
873        order_type: OrderType,
874        quantity: Quantity,
875        is_quote_quantity: bool,
876        time_in_force: Option<TimeInForce>,
877        price: Option<Price>,
878        trigger_price: Option<Price>,
879        trigger_type: Option<TriggerType>,
880        post_only: Option<bool>,
881        reduce_only: Option<bool>,
882        is_leverage: bool,
883        position_idx: Option<BybitPositionIdx>,
884        bbo_side_type: Option<String>,
885        bbo_level: Option<String>,
886    ) -> PyResult<Bound<'py, PyAny>> {
887        let client = self.clone();
888        let pending_py_requests = Arc::clone(self.pending_py_requests());
889        let bbo_side_type = bbo_side_type
890            .map(|value| parse_bbo_side_type(&value))
891            .transpose()
892            .map_err(to_pyvalue_err)?;
893        let bbo_level = bbo_level
894            .map(parse_bbo_level)
895            .transpose()
896            .map_err(to_pyvalue_err)?;
897        if bbo_side_type.is_some() != bbo_level.is_some() {
898            return Err(to_pyvalue_err(anyhow::anyhow!(
899                "'bbo_side_type' and 'bbo_level' must be provided together"
900            )));
901        }
902
903        pyo3_async_runtimes::tokio::future_into_py(py, async move {
904            let req_id = client
905                .submit_order(
906                    product_type,
907                    instrument_id,
908                    client_order_id,
909                    order_side,
910                    order_type,
911                    quantity,
912                    is_quote_quantity,
913                    time_in_force,
914                    price,
915                    trigger_price,
916                    trigger_type,
917                    post_only,
918                    reduce_only,
919                    is_leverage,
920                    position_idx,
921                    bbo_side_type,
922                    bbo_level,
923                )
924                .await
925                .map_err(to_pyruntime_err)?;
926            pending_py_requests.insert(
927                req_id,
928                vec![PendingPyRequest {
929                    client_order_id,
930                    operation: PendingOperation::Place,
931                    trader_id,
932                    strategy_id,
933                    instrument_id,
934                    venue_order_id: None,
935                }],
936            );
937            Ok(())
938        })
939    }
940
941    /// Modifies an existing order using Nautilus domain objects.
942    #[pyo3(name = "modify_order")]
943    #[pyo3(signature = (
944        product_type,
945        trader_id,
946        strategy_id,
947        instrument_id,
948        client_order_id,
949        venue_order_id=None,
950        quantity=None,
951        price=None,
952    ))]
953    #[expect(clippy::too_many_arguments)]
954    fn py_modify_order<'py>(
955        &self,
956        py: Python<'py>,
957        product_type: BybitProductType,
958        trader_id: TraderId,
959        strategy_id: StrategyId,
960        instrument_id: InstrumentId,
961        client_order_id: ClientOrderId,
962        venue_order_id: Option<VenueOrderId>,
963        quantity: Option<Quantity>,
964        price: Option<Price>,
965    ) -> PyResult<Bound<'py, PyAny>> {
966        let client = self.clone();
967        let pending_py_requests = Arc::clone(self.pending_py_requests());
968
969        pyo3_async_runtimes::tokio::future_into_py(py, async move {
970            let req_id = client
971                .modify_order(
972                    product_type,
973                    instrument_id,
974                    client_order_id,
975                    venue_order_id,
976                    quantity,
977                    price,
978                )
979                .await
980                .map_err(to_pyruntime_err)?;
981            pending_py_requests.insert(
982                req_id,
983                vec![PendingPyRequest {
984                    client_order_id,
985                    operation: PendingOperation::Amend,
986                    trader_id,
987                    strategy_id,
988                    instrument_id,
989                    venue_order_id,
990                }],
991            );
992            Ok(())
993        })
994    }
995
996    /// Cancels an order via WebSocket, returning the request ID for correlation.
997    #[pyo3(name = "cancel_order")]
998    #[pyo3(signature = (
999        product_type,
1000        trader_id,
1001        strategy_id,
1002        instrument_id,
1003        client_order_id,
1004        venue_order_id=None,
1005    ))]
1006    #[expect(clippy::too_many_arguments)]
1007    fn py_cancel_order<'py>(
1008        &self,
1009        py: Python<'py>,
1010        product_type: BybitProductType,
1011        trader_id: TraderId,
1012        strategy_id: StrategyId,
1013        instrument_id: InstrumentId,
1014        client_order_id: ClientOrderId,
1015        venue_order_id: Option<VenueOrderId>,
1016    ) -> PyResult<Bound<'py, PyAny>> {
1017        let client = self.clone();
1018        let pending_py_requests = Arc::clone(self.pending_py_requests());
1019
1020        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1021            let req_id = client
1022                .cancel_order_by_id(product_type, instrument_id, client_order_id, venue_order_id)
1023                .await
1024                .map_err(to_pyruntime_err)?;
1025            pending_py_requests.insert(
1026                req_id,
1027                vec![PendingPyRequest {
1028                    client_order_id,
1029                    operation: PendingOperation::Cancel,
1030                    trader_id,
1031                    strategy_id,
1032                    instrument_id,
1033                    venue_order_id,
1034                }],
1035            );
1036            Ok(())
1037        })
1038    }
1039
1040    /// Builds order params for placing an order.
1041    #[pyo3(name = "build_place_order_params")]
1042    #[pyo3(signature = (
1043        product_type,
1044        instrument_id,
1045        client_order_id,
1046        order_side,
1047        order_type,
1048        quantity,
1049        is_quote_quantity=false,
1050        time_in_force=None,
1051        price=None,
1052        trigger_price=None,
1053        trigger_type=None,
1054        post_only=None,
1055        reduce_only=None,
1056        is_leverage=false,
1057        take_profit=None,
1058        stop_loss=None,
1059        position_idx=None,
1060        bbo_side_type=None,
1061        bbo_level=None,
1062    ))]
1063    #[expect(clippy::too_many_arguments)]
1064    fn py_build_place_order_params(
1065        &self,
1066        product_type: BybitProductType,
1067        instrument_id: InstrumentId,
1068        client_order_id: ClientOrderId,
1069        order_side: OrderSide,
1070        order_type: OrderType,
1071        quantity: Quantity,
1072        is_quote_quantity: bool,
1073        time_in_force: Option<TimeInForce>,
1074        price: Option<Price>,
1075        trigger_price: Option<Price>,
1076        trigger_type: Option<TriggerType>,
1077        post_only: Option<bool>,
1078        reduce_only: Option<bool>,
1079        is_leverage: bool,
1080        take_profit: Option<Price>,
1081        stop_loss: Option<Price>,
1082        position_idx: Option<BybitPositionIdx>,
1083        bbo_side_type: Option<String>,
1084        bbo_level: Option<String>,
1085    ) -> PyResult<BybitWsPlaceOrderParams> {
1086        let bbo_side_type = bbo_side_type
1087            .map(|value| parse_bbo_side_type(&value))
1088            .transpose()
1089            .map_err(to_pyvalue_err)?;
1090        let bbo_level = bbo_level
1091            .map(parse_bbo_level)
1092            .transpose()
1093            .map_err(to_pyvalue_err)?;
1094        if bbo_side_type.is_some() != bbo_level.is_some() {
1095            return Err(to_pyvalue_err(anyhow::anyhow!(
1096                "'bbo_side_type' and 'bbo_level' must be provided together"
1097            )));
1098        }
1099
1100        let params = self
1101            .build_place_order_params(
1102                product_type,
1103                instrument_id,
1104                client_order_id,
1105                order_side,
1106                order_type,
1107                quantity,
1108                is_quote_quantity,
1109                time_in_force,
1110                price,
1111                trigger_price,
1112                trigger_type,
1113                post_only,
1114                reduce_only,
1115                is_leverage,
1116                take_profit,
1117                stop_loss,
1118                position_idx,
1119                bbo_side_type,
1120                bbo_level,
1121            )
1122            .map_err(to_pyruntime_err)?;
1123        Ok(params.into())
1124    }
1125
1126    /// Batch cancels multiple orders via WebSocket, returning the request ID for correlation.
1127    #[pyo3(name = "batch_cancel_orders")]
1128    fn py_batch_cancel_orders<'py>(
1129        &self,
1130        py: Python<'py>,
1131        trader_id: TraderId,
1132        strategy_id: StrategyId,
1133        orders: Vec<BybitWsCancelOrderParams>,
1134    ) -> PyResult<Bound<'py, PyAny>> {
1135        let client = self.clone();
1136        let pending_py_requests = Arc::clone(self.pending_py_requests());
1137
1138        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1139            let order_params: Vec<crate::websocket::messages::BybitWsCancelOrderParams> = orders
1140                .into_iter()
1141                .map(|p| p.try_into())
1142                .collect::<Result<Vec<_>, _>>()
1143                .map_err(to_pyruntime_err)?;
1144
1145            let per_order = build_pending_entries(
1146                &order_params,
1147                PendingOperation::Cancel,
1148                trader_id,
1149                strategy_id,
1150            );
1151
1152            let req_ids = client
1153                .batch_cancel_orders(order_params)
1154                .await
1155                .map_err(to_pyruntime_err)?;
1156
1157            register_batch_pending(req_ids, &per_order, &pending_py_requests);
1158            Ok(())
1159        })
1160    }
1161
1162    /// Builds order params for amending an order.
1163    #[pyo3(name = "build_amend_order_params")]
1164    fn py_build_amend_order_params(
1165        &self,
1166        product_type: BybitProductType,
1167        instrument_id: InstrumentId,
1168        venue_order_id: Option<VenueOrderId>,
1169        client_order_id: Option<ClientOrderId>,
1170        quantity: Option<Quantity>,
1171        price: Option<Price>,
1172    ) -> PyResult<crate::python::params::BybitWsAmendOrderParams> {
1173        let params = self
1174            .build_amend_order_params(
1175                product_type,
1176                instrument_id,
1177                venue_order_id,
1178                client_order_id,
1179                quantity,
1180                price,
1181            )
1182            .map_err(to_pyruntime_err)?;
1183        Ok(params.into())
1184    }
1185
1186    /// Builds order params for canceling an order via WebSocket.
1187    #[pyo3(name = "build_cancel_order_params")]
1188    fn py_build_cancel_order_params(
1189        &self,
1190        product_type: BybitProductType,
1191        instrument_id: InstrumentId,
1192        venue_order_id: Option<VenueOrderId>,
1193        client_order_id: Option<ClientOrderId>,
1194    ) -> PyResult<crate::python::params::BybitWsCancelOrderParams> {
1195        let params = self
1196            .build_cancel_order_params(product_type, instrument_id, venue_order_id, client_order_id)
1197            .map_err(to_pyruntime_err)?;
1198        Ok(params.into())
1199    }
1200
1201    #[pyo3(name = "batch_modify_orders")]
1202    fn py_batch_modify_orders<'py>(
1203        &self,
1204        py: Python<'py>,
1205        trader_id: TraderId,
1206        strategy_id: StrategyId,
1207        orders: Vec<BybitWsAmendOrderParams>,
1208    ) -> PyResult<Bound<'py, PyAny>> {
1209        let client = self.clone();
1210        let pending_py_requests = Arc::clone(self.pending_py_requests());
1211
1212        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1213            let order_params: Vec<crate::websocket::messages::BybitWsAmendOrderParams> = orders
1214                .into_iter()
1215                .map(|p| p.try_into())
1216                .collect::<Result<Vec<_>, _>>()
1217                .map_err(to_pyruntime_err)?;
1218
1219            let per_order = build_pending_entries(
1220                &order_params,
1221                PendingOperation::Amend,
1222                trader_id,
1223                strategy_id,
1224            );
1225
1226            let req_ids = client
1227                .batch_amend_orders(order_params)
1228                .await
1229                .map_err(to_pyruntime_err)?;
1230
1231            register_batch_pending(req_ids, &per_order, &pending_py_requests);
1232            Ok(())
1233        })
1234    }
1235
1236    /// Batch creates multiple orders via WebSocket, returning the request ID for correlation.
1237    #[pyo3(name = "batch_place_orders")]
1238    fn py_batch_place_orders<'py>(
1239        &self,
1240        py: Python<'py>,
1241        trader_id: TraderId,
1242        strategy_id: StrategyId,
1243        orders: Vec<BybitWsPlaceOrderParams>,
1244    ) -> PyResult<Bound<'py, PyAny>> {
1245        let client = self.clone();
1246        let pending_py_requests = Arc::clone(self.pending_py_requests());
1247
1248        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1249            let order_params: Vec<crate::websocket::messages::BybitWsPlaceOrderParams> = orders
1250                .into_iter()
1251                .map(|p| p.try_into())
1252                .collect::<Result<Vec<_>, _>>()
1253                .map_err(to_pyruntime_err)?;
1254
1255            let per_order = build_pending_entries(
1256                &order_params,
1257                PendingOperation::Place,
1258                trader_id,
1259                strategy_id,
1260            );
1261
1262            let req_ids = client
1263                .batch_place_orders(order_params)
1264                .await
1265                .map_err(to_pyruntime_err)?;
1266
1267            register_batch_pending(req_ids, &per_order, &pending_py_requests);
1268            Ok(())
1269        })
1270    }
1271}
1272
1273trait BatchOrderParams {
1274    fn order_link_id(&self) -> Option<&str>;
1275    fn symbol(&self) -> Ustr;
1276    fn category(&self) -> BybitProductType;
1277    fn venue_order_id(&self) -> Option<VenueOrderId>;
1278}
1279
1280impl BatchOrderParams for crate::websocket::messages::BybitWsCancelOrderParams {
1281    fn order_link_id(&self) -> Option<&str> {
1282        self.order_link_id.as_deref()
1283    }
1284    fn symbol(&self) -> Ustr {
1285        self.symbol
1286    }
1287    fn category(&self) -> BybitProductType {
1288        self.category
1289    }
1290    fn venue_order_id(&self) -> Option<VenueOrderId> {
1291        self.order_id.as_ref().map(VenueOrderId::new)
1292    }
1293}
1294
1295impl BatchOrderParams for crate::websocket::messages::BybitWsAmendOrderParams {
1296    fn order_link_id(&self) -> Option<&str> {
1297        self.order_link_id.as_deref()
1298    }
1299    fn symbol(&self) -> Ustr {
1300        self.symbol
1301    }
1302    fn category(&self) -> BybitProductType {
1303        self.category
1304    }
1305    fn venue_order_id(&self) -> Option<VenueOrderId> {
1306        self.order_id.as_ref().map(VenueOrderId::new)
1307    }
1308}
1309
1310impl BatchOrderParams for crate::websocket::messages::BybitWsPlaceOrderParams {
1311    fn order_link_id(&self) -> Option<&str> {
1312        self.order_link_id.as_deref()
1313    }
1314    fn symbol(&self) -> Ustr {
1315        self.symbol
1316    }
1317    fn category(&self) -> BybitProductType {
1318        self.category
1319    }
1320    fn venue_order_id(&self) -> Option<VenueOrderId> {
1321        None
1322    }
1323}
1324
1325fn build_pending_entries<P: BatchOrderParams>(
1326    params: &[P],
1327    operation: PendingOperation,
1328    trader_id: TraderId,
1329    strategy_id: StrategyId,
1330) -> Vec<PendingPyRequest> {
1331    params
1332        .iter()
1333        .map(|p| PendingPyRequest {
1334            client_order_id: p
1335                .order_link_id()
1336                .filter(|s| !s.is_empty())
1337                .map_or(ClientOrderId::from("UNKNOWN"), ClientOrderId::new),
1338            operation,
1339            trader_id,
1340            strategy_id,
1341            instrument_id: InstrumentId::new(
1342                Symbol::new(make_bybit_symbol(p.symbol().as_str(), p.category()).as_str()),
1343                *BYBIT_VENUE,
1344            ),
1345            venue_order_id: p.venue_order_id(),
1346        })
1347        .collect()
1348}
1349
1350fn register_batch_pending(
1351    req_ids: Vec<String>,
1352    per_order: &[PendingPyRequest],
1353    pending_py_requests: &DashMap<String, Vec<PendingPyRequest>>,
1354) {
1355    for (req_id, chunk) in req_ids
1356        .into_iter()
1357        .zip(per_order.chunks(BATCH_PROCESSING_LIMIT))
1358    {
1359        pending_py_requests.insert(req_id, chunk.to_vec());
1360    }
1361}
1362
1363fn resolve_instrument_from_snapshot<'a>(
1364    raw_symbol: &Ustr,
1365    product_type: Option<BybitProductType>,
1366    instruments: &'a AHashMap<Ustr, InstrumentAny>,
1367) -> Option<&'a InstrumentAny> {
1368    let key = product_type.map_or(*raw_symbol, |pt| make_bybit_symbol(raw_symbol, pt));
1369    instruments.get(&key)
1370}
1371
1372fn send_data_to_python(data: Data, call_soon: &Py<PyAny>, callback: &Py<PyAny>) {
1373    Python::attach(|py| {
1374        let py_obj = data_to_pycapsule(py, data);
1375        call_python_threadsafe(py, call_soon, callback, py_obj);
1376    });
1377}
1378
1379fn send_to_python<T: for<'py> IntoPyObjectExt<'py>>(
1380    value: T,
1381    call_soon: &Py<PyAny>,
1382    callback: &Py<PyAny>,
1383) {
1384    Python::attach(|py| {
1385        if let Ok(py_obj) = value.into_py_any(py) {
1386            call_python_threadsafe(py, call_soon, callback, py_obj);
1387        }
1388    });
1389}
1390
1391fn handle_orderbook(
1392    msg: &crate::websocket::messages::BybitWsOrderbookDepthMsg,
1393    product_type: Option<BybitProductType>,
1394    instruments: &AtomicMap<Ustr, InstrumentAny>,
1395    quote_cache: &mut AHashMap<InstrumentId, QuoteTick>,
1396    clock: &AtomicTime,
1397    call_soon: &Py<PyAny>,
1398    callback: &Py<PyAny>,
1399) {
1400    let instruments_snapshot = instruments.load();
1401    let Some(instrument) =
1402        resolve_instrument_from_snapshot(&msg.data.s, product_type, &instruments_snapshot)
1403    else {
1404        return;
1405    };
1406    let ts_init = clock.get_time_ns();
1407
1408    match parse_orderbook_deltas(msg, instrument, ts_init) {
1409        Ok(deltas) => {
1410            send_data_to_python(
1411                Data::Deltas(OrderBookDeltas_API::new(deltas)),
1412                call_soon,
1413                callback,
1414            );
1415        }
1416        Err(e) => log::error!("Failed to parse orderbook deltas: {e}"),
1417    }
1418
1419    let instrument_id = instrument.id();
1420    let last_quote = quote_cache.get(&instrument_id);
1421
1422    match parse_orderbook_quote(msg, instrument, last_quote, ts_init) {
1423        Ok(quote) => {
1424            quote_cache.insert(instrument_id, quote);
1425            send_data_to_python(Data::Quote(quote), call_soon, callback);
1426        }
1427        Err(e) => log::error!("Failed to parse orderbook quote: {e}"),
1428    }
1429}
1430
1431fn handle_trade(
1432    msg: &crate::websocket::messages::BybitWsTradeMsg,
1433    product_type: Option<BybitProductType>,
1434    instruments: &AtomicMap<Ustr, InstrumentAny>,
1435    trade_subs: &AtomicSet<InstrumentId>,
1436    clock: &AtomicTime,
1437    call_soon: &Py<PyAny>,
1438    callback: &Py<PyAny>,
1439) {
1440    let ts_init = clock.get_time_ns();
1441    let instruments_snapshot = instruments.load();
1442
1443    for trade in &msg.data {
1444        let Some(instrument) =
1445            resolve_instrument_from_snapshot(&trade.s, product_type, &instruments_snapshot)
1446        else {
1447            continue;
1448        };
1449
1450        if product_type == Some(BybitProductType::Option)
1451            && !trade_subs.is_empty()
1452            && !trade_subs.contains(&instrument.id())
1453        {
1454            continue;
1455        }
1456
1457        match parse_ws_trade_tick(trade, instrument, ts_init) {
1458            Ok(tick) => send_data_to_python(Data::Trade(tick), call_soon, callback),
1459            Err(e) => log::error!("Failed to parse trade tick: {e}"),
1460        }
1461    }
1462}
1463
1464#[expect(clippy::too_many_arguments)]
1465fn handle_kline(
1466    msg: &crate::websocket::messages::BybitWsKlineMsg,
1467    product_type: Option<BybitProductType>,
1468    instruments: &AtomicMap<Ustr, InstrumentAny>,
1469    bar_types_cache: &AtomicMap<String, BarType>,
1470    bars_timestamp_on_close: bool,
1471    clock: &AtomicTime,
1472    call_soon: &Py<PyAny>,
1473    callback: &Py<PyAny>,
1474) {
1475    let Ok((_, raw_symbol)) = parse_kline_topic(msg.topic.as_str()) else {
1476        return;
1477    };
1478    let ustr_symbol = Ustr::from(raw_symbol);
1479    let instruments_snapshot = instruments.load();
1480    let Some(instrument) =
1481        resolve_instrument_from_snapshot(&ustr_symbol, product_type, &instruments_snapshot)
1482    else {
1483        return;
1484    };
1485    let Some(bar_type) = bar_types_cache.load().get(msg.topic.as_str()).copied() else {
1486        return;
1487    };
1488
1489    let ts_init = clock.get_time_ns();
1490
1491    for kline in &msg.data {
1492        if !kline.confirm {
1493            continue;
1494        }
1495
1496        match parse_ws_kline_bar(
1497            kline,
1498            instrument,
1499            bar_type,
1500            bars_timestamp_on_close,
1501            ts_init,
1502        ) {
1503            Ok(bar) => send_data_to_python(Data::Bar(bar), call_soon, callback),
1504            Err(e) => log::error!("Failed to parse kline bar: {e}"),
1505        }
1506    }
1507}
1508
1509#[expect(clippy::too_many_arguments)]
1510fn handle_ticker_linear(
1511    msg: &crate::websocket::messages::BybitWsTickerLinearMsg,
1512    product_type: Option<BybitProductType>,
1513    instruments: &AtomicMap<Ustr, InstrumentAny>,
1514    quote_cache: &mut AHashMap<InstrumentId, QuoteTick>,
1515    funding_cache: &mut AHashMap<Ustr, (Option<String>, Option<String>)>,
1516    clock: &AtomicTime,
1517    call_soon: &Py<PyAny>,
1518    callback: &Py<PyAny>,
1519) {
1520    let instruments_snapshot = instruments.load();
1521    let Some(instrument) =
1522        resolve_instrument_from_snapshot(&msg.data.symbol, product_type, &instruments_snapshot)
1523    else {
1524        return;
1525    };
1526    let instrument_id = instrument.id();
1527    let ts_init = clock.get_time_ns();
1528
1529    if msg.data.bid1_price.is_some() {
1530        match parse_ticker_linear_quote(msg, instrument, ts_init) {
1531            Ok(quote) => {
1532                let last = quote_cache.get(&instrument_id);
1533
1534                if last.is_none_or(|q| *q != quote) {
1535                    quote_cache.insert(instrument_id, quote);
1536                    send_data_to_python(Data::Quote(quote), call_soon, callback);
1537                }
1538            }
1539            Err(e) => log::debug!("Skipping partial ticker update: {e}"),
1540        }
1541    }
1542
1543    let ts_event = match parse_millis_i64(msg.ts, "ticker.ts") {
1544        Ok(ts) => ts,
1545        Err(e) => {
1546            log::error!("Failed to parse ticker timestamp: {e}");
1547            return;
1548        }
1549    };
1550
1551    let cache_entry = funding_cache.entry(msg.data.symbol).or_insert((None, None));
1552    let mut changed = false;
1553
1554    if let Some(rate) = &msg.data.funding_rate
1555        && cache_entry.0.as_ref() != Some(rate)
1556    {
1557        cache_entry.0 = Some(rate.clone());
1558        changed = true;
1559    }
1560
1561    if let Some(next_time) = &msg.data.next_funding_time
1562        && cache_entry.1.as_ref() != Some(next_time)
1563    {
1564        cache_entry.1 = Some(next_time.clone());
1565        changed = true;
1566    }
1567
1568    if changed {
1569        match parse_ticker_linear_funding(&msg.data, instrument_id, ts_event, ts_init) {
1570            Ok(update) => send_to_python(update, call_soon, callback),
1571            Err(e) => log::debug!("Skipping funding rate update: {e}"),
1572        }
1573    }
1574
1575    if msg.data.mark_price.is_some() {
1576        match parse_ticker_linear_mark_price(&msg.data, instrument, ts_event, ts_init) {
1577            Ok(update) => send_to_python(update, call_soon, callback),
1578            Err(e) => log::debug!("Skipping mark price update: {e}"),
1579        }
1580    }
1581
1582    if msg.data.index_price.is_some() {
1583        match parse_ticker_linear_index_price(&msg.data, instrument, ts_event, ts_init) {
1584            Ok(update) => send_to_python(update, call_soon, callback),
1585            Err(e) => log::debug!("Skipping index price update: {e}"),
1586        }
1587    }
1588}
1589
1590#[expect(clippy::too_many_arguments)]
1591fn handle_ticker_option(
1592    msg: &crate::websocket::messages::BybitWsTickerOptionMsg,
1593    product_type: Option<BybitProductType>,
1594    instruments: &AtomicMap<Ustr, InstrumentAny>,
1595    quote_cache: &mut AHashMap<InstrumentId, QuoteTick>,
1596    option_greeks_subs: &AtomicSet<InstrumentId>,
1597    clock: &AtomicTime,
1598    call_soon: &Py<PyAny>,
1599    callback: &Py<PyAny>,
1600) {
1601    let instruments_snapshot = instruments.load();
1602    let Some(instrument) =
1603        resolve_instrument_from_snapshot(&msg.data.symbol, product_type, &instruments_snapshot)
1604    else {
1605        return;
1606    };
1607    let instrument_id = instrument.id();
1608    let ts_init = clock.get_time_ns();
1609
1610    match parse_ticker_option_quote(msg, instrument, ts_init) {
1611        Ok(quote) => {
1612            let last = quote_cache.get(&instrument_id);
1613
1614            if last.is_none_or(|q| *q != quote) {
1615                quote_cache.insert(instrument_id, quote);
1616                send_data_to_python(Data::Quote(quote), call_soon, callback);
1617            }
1618        }
1619        Err(e) => log::error!("Failed to parse ticker option quote: {e}"),
1620    }
1621
1622    match parse_ticker_option_mark_price(msg, instrument, ts_init) {
1623        Ok(update) => send_to_python(update, call_soon, callback),
1624        Err(e) => log::error!("Failed to parse ticker option mark price: {e}"),
1625    }
1626
1627    match parse_ticker_option_index_price(msg, instrument, ts_init) {
1628        Ok(update) => send_to_python(update, call_soon, callback),
1629        Err(e) => log::error!("Failed to parse ticker option index price: {e}"),
1630    }
1631
1632    if option_greeks_subs.contains(&instrument_id) {
1633        match parse_ticker_option_greeks(msg, instrument, ts_init) {
1634            Ok(greeks) => send_to_python(greeks, call_soon, callback),
1635            Err(e) => log::error!("Failed to parse option greeks: {e}"),
1636        }
1637    }
1638}
1639
1640fn handle_account_order(
1641    msg: &crate::websocket::messages::BybitWsAccountOrderMsg,
1642    instruments: &AtomicMap<Ustr, InstrumentAny>,
1643    account_id: Option<AccountId>,
1644    clock: &AtomicTime,
1645    call_soon: &Py<PyAny>,
1646    callback: &Py<PyAny>,
1647) {
1648    let ts_init = clock.get_time_ns();
1649    let instruments_snapshot = instruments.load();
1650
1651    for order in &msg.data {
1652        let symbol = make_bybit_symbol(order.symbol, order.category);
1653        let Some(instrument) = instruments_snapshot.get(&symbol) else {
1654            log::warn!("No instrument for order update: {symbol}");
1655            continue;
1656        };
1657        let Some(account_id) = account_id else {
1658            continue;
1659        };
1660
1661        match parse_ws_order_status_report(order, instrument, account_id, ts_init) {
1662            Ok(report) => send_to_python(report, call_soon, callback),
1663            Err(e) => log::error!("Failed to parse order status report: {e}"),
1664        }
1665    }
1666}
1667
1668fn handle_account_execution(
1669    msg: &crate::websocket::messages::BybitWsAccountExecutionMsg,
1670    instruments: &AtomicMap<Ustr, InstrumentAny>,
1671    account_id: Option<AccountId>,
1672    clock: &AtomicTime,
1673    call_soon: &Py<PyAny>,
1674    callback: &Py<PyAny>,
1675) {
1676    let ts_init = clock.get_time_ns();
1677    let instruments_snapshot = instruments.load();
1678
1679    for exec in &msg.data {
1680        let symbol = make_bybit_symbol(exec.symbol, exec.category);
1681        let Some(instrument) = instruments_snapshot.get(&symbol) else {
1682            log::warn!("No instrument for execution update: {symbol}");
1683            continue;
1684        };
1685        let Some(account_id) = account_id else {
1686            continue;
1687        };
1688
1689        match parse_ws_fill_report(exec, account_id, instrument, ts_init) {
1690            Ok(report) => send_to_python(report, call_soon, callback),
1691            Err(e) => log::error!("Failed to parse fill report: {e}"),
1692        }
1693    }
1694}
1695
1696fn handle_account_execution_fast(
1697    msg: &crate::websocket::messages::BybitWsAccountExecutionFastMsg,
1698    instruments: &AtomicMap<Ustr, InstrumentAny>,
1699    account_id: Option<AccountId>,
1700    clock: &AtomicTime,
1701    call_soon: &Py<PyAny>,
1702    callback: &Py<PyAny>,
1703) {
1704    let ts_init = clock.get_time_ns();
1705    let instruments_snapshot = instruments.load();
1706
1707    for exec in &msg.data {
1708        let symbol = make_bybit_symbol(exec.symbol, exec.category);
1709        let Some(instrument) = instruments_snapshot.get(&symbol) else {
1710            log::warn!("No instrument for fast-execution update: {symbol}");
1711            continue;
1712        };
1713        let Some(account_id) = account_id else {
1714            continue;
1715        };
1716
1717        match parse_ws_fill_report_fast(exec, account_id, instrument, None, ts_init) {
1718            Ok(report) => send_to_python(report, call_soon, callback),
1719            Err(e) => log::error!("Failed to parse fast fill report: {e}"),
1720        }
1721    }
1722}
1723
1724fn handle_account_wallet(
1725    msg: &crate::websocket::messages::BybitWsAccountWalletMsg,
1726    account_id: Option<AccountId>,
1727    clock: &AtomicTime,
1728    call_soon: &Py<PyAny>,
1729    callback: &Py<PyAny>,
1730) {
1731    let ts_init = clock.get_time_ns();
1732    let ts_event = parse_millis_i64(msg.creation_time, "wallet.creation_time").unwrap_or(ts_init);
1733    let Some(account_id) = account_id else {
1734        return;
1735    };
1736
1737    for wallet in &msg.data {
1738        match parse_ws_account_state(wallet, account_id, ts_event, ts_init) {
1739            Ok(state) => send_to_python(state, call_soon, callback),
1740            Err(e) => log::error!("Failed to parse account state: {e}"),
1741        }
1742    }
1743}
1744
1745fn handle_account_position(
1746    msg: &crate::websocket::messages::BybitWsAccountPositionMsg,
1747    instruments: &AtomicMap<Ustr, InstrumentAny>,
1748    account_id: Option<AccountId>,
1749    clock: &AtomicTime,
1750    call_soon: &Py<PyAny>,
1751    callback: &Py<PyAny>,
1752) {
1753    let ts_init = clock.get_time_ns();
1754    let instruments_snapshot = instruments.load();
1755
1756    for position in &msg.data {
1757        let symbol = make_bybit_symbol(position.symbol, position.category);
1758        let Some(instrument) = instruments_snapshot.get(&symbol) else {
1759            log::warn!("No instrument for position update: {symbol}");
1760            continue;
1761        };
1762        let Some(account_id) = account_id else {
1763            continue;
1764        };
1765
1766        match parse_ws_position_status_report(position, account_id, instrument, ts_init) {
1767            Ok(report) => send_to_python(report, call_soon, callback),
1768            Err(e) => log::error!("Failed to parse position status report: {e}"),
1769        }
1770    }
1771}
1772
1773fn handle_order_response(
1774    resp: &crate::websocket::messages::BybitWsOrderResponse,
1775    pending_py_requests: &DashMap<String, Vec<PendingPyRequest>>,
1776    account_id: Option<AccountId>,
1777    clock: &AtomicTime,
1778    call_soon: &Py<PyAny>,
1779    callback: &Py<PyAny>,
1780) {
1781    if resp.ret_code == 0 {
1782        let entries = resp
1783            .req_id
1784            .as_ref()
1785            .and_then(|rid| pending_py_requests.remove(rid))
1786            .map(|(_, v)| v);
1787
1788        // Check for per-order failures in batch retExtInfo
1789        if let Some(entries) = entries {
1790            let batch_errors = resp.extract_batch_errors();
1791            let data_array = resp.data.as_array();
1792            let ts_init = clock.get_time_ns();
1793
1794            for (idx, error) in batch_errors.iter().enumerate() {
1795                if error.code == 0 {
1796                    continue;
1797                }
1798
1799                let pending = data_array
1800                    .and_then(|arr| arr.get(idx))
1801                    .and_then(|item| item.get("orderLinkId"))
1802                    .and_then(|v| v.as_str())
1803                    .filter(|s| !s.is_empty())
1804                    .and_then(|oli| {
1805                        let cid = ClientOrderId::new(oli);
1806                        entries.iter().find(|e| e.client_order_id == cid)
1807                    })
1808                    .or_else(|| entries.get(idx));
1809
1810                if let Some(pending) = pending {
1811                    let reason = Ustr::from(&error.msg);
1812                    emit_rejection(pending, reason, account_id, ts_init, call_soon, callback);
1813                } else {
1814                    log::warn!(
1815                        "Batch error at index {idx} without correlation: code={}, msg={}",
1816                        error.code,
1817                        error.msg,
1818                    );
1819                }
1820            }
1821        }
1822        return;
1823    }
1824
1825    // Try to find the pending entries by req_id, then by orderLinkId
1826    let entries = resp
1827        .req_id
1828        .as_ref()
1829        .and_then(|rid| pending_py_requests.remove(rid))
1830        .map(|(_, v)| v)
1831        .or_else(|| {
1832            // Bybit sometimes omits req_id, search by orderLinkId instead
1833            let order_link_id = resp
1834                .data
1835                .get("orderLinkId")
1836                .and_then(|v| v.as_str())
1837                .filter(|s| !s.is_empty())?;
1838            let cid = ClientOrderId::new(order_link_id);
1839            let key = pending_py_requests
1840                .iter()
1841                .find(|entry| entry.value().iter().any(|e| e.client_order_id == cid))
1842                .map(|entry| entry.key().clone())?;
1843            pending_py_requests.remove(&key).map(|(_, v)| v)
1844        });
1845
1846    let Some(entries) = entries else {
1847        log::warn!(
1848            "Unmatched order response: ret_code={}, ret_msg={}",
1849            resp.ret_code,
1850            resp.ret_msg,
1851        );
1852        return;
1853    };
1854
1855    let ts_init = clock.get_time_ns();
1856    let reason = Ustr::from(&resp.ret_msg);
1857
1858    for pending in &entries {
1859        emit_rejection(pending, reason, account_id, ts_init, call_soon, callback);
1860    }
1861}
1862
1863fn emit_rejection(
1864    pending: &PendingPyRequest,
1865    reason: Ustr,
1866    account_id: Option<AccountId>,
1867    ts_init: UnixNanos,
1868    call_soon: &Py<PyAny>,
1869    callback: &Py<PyAny>,
1870) {
1871    match pending.operation {
1872        PendingOperation::Place => {
1873            let event = OrderRejected::new(
1874                pending.trader_id,
1875                pending.strategy_id,
1876                pending.instrument_id,
1877                pending.client_order_id,
1878                account_id.unwrap_or(AccountId::from("BYBIT-000")),
1879                reason,
1880                UUID4::new(),
1881                ts_init,
1882                ts_init,
1883                false,
1884                false,
1885            );
1886            send_to_python(event, call_soon, callback);
1887        }
1888        PendingOperation::Cancel => {
1889            let event = OrderCancelRejected::new(
1890                pending.trader_id,
1891                pending.strategy_id,
1892                pending.instrument_id,
1893                pending.client_order_id,
1894                reason,
1895                UUID4::new(),
1896                ts_init,
1897                ts_init,
1898                false,
1899                pending.venue_order_id,
1900                account_id,
1901            );
1902            send_to_python(event, call_soon, callback);
1903        }
1904        PendingOperation::Amend => {
1905            let event = OrderModifyRejected::new(
1906                pending.trader_id,
1907                pending.strategy_id,
1908                pending.instrument_id,
1909                pending.client_order_id,
1910                reason,
1911                UUID4::new(),
1912                ts_init,
1913                ts_init,
1914                false,
1915                pending.venue_order_id,
1916                account_id,
1917            );
1918            send_to_python(event, call_soon, callback);
1919        }
1920    }
1921}
nautilus_bybit/python/websocket.rs

nautilus_bybit/python/
websocket.rs
