nautilus_bitmex/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 BitMEX WebSocket client.
17//!
18//! [`PyBitmexWebSocketClient`] wraps the Rust [`BitmexWebSocketClient`] and adds an
19//! instrument cache at the Python boundary. The inner client is a pure network component
20//! that emits venue-specific types; this wrapper parses them into Nautilus domain objects
21//! before passing them to Python callbacks.
22//!
23//! The instrument cache is shared via `Arc<AtomicMap>` so that:
24//! - Python can inject instruments at any time via `cache_instrument`.
25//! - The spawned stream task reads from the same cache for parsing.
26//! - Instrument table messages from the venue update the cache automatically.
27
28use std::{fmt::Debug, sync::Arc};
29
30use ahash::AHashMap;
31use futures_util::StreamExt;
32use nautilus_common::{cache::quote::QuoteCache, live::get_runtime};
33use nautilus_core::{
34    AtomicMap, UUID4, UnixNanos,
35    python::{call_python_threadsafe, to_pyruntime_err, to_pyvalue_err},
36    time::get_atomic_clock_realtime,
37};
38use nautilus_model::{
39    data::{Data, InstrumentStatus, bar::BarType},
40    enums::{MarketStatusAction, OrderSide, OrderType},
41    events::{OrderAccepted, OrderUpdated},
42    identifiers::{AccountId, ClientOrderId, InstrumentId, StrategyId, TraderId, VenueOrderId},
43    instruments::{Instrument, InstrumentAny},
44    python::{
45        data::data_to_pycapsule,
46        instruments::{instrument_any_to_pyobject, pyobject_to_instrument_any},
47    },
48    types::Price,
49};
50use nautilus_network::websocket::TransportBackend;
51use pyo3::{conversion::IntoPyObjectExt, prelude::*};
52use ustr::Ustr;
53
54use crate::{
55    common::{
56        enums::{
57            BitmexEnvironment, BitmexExecType, BitmexInstrumentState, BitmexOrderType,
58            BitmexPegPriceType,
59        },
60        parse::{
61            parse_contracts_quantity, parse_instrument_id, parse_optional_datetime_to_unix_nanos,
62        },
63    },
64    http::parse::{InstrumentParseResult, parse_instrument_any},
65    websocket::{
66        BitmexWebSocketClient,
67        dispatch::{OrderIdentity, WsDispatchState, fill_report_to_order_filled},
68        enums::{BitmexAction, BitmexWsTopic},
69        messages::{
70            BitmexExecutionMsg, BitmexInstrumentMsg, BitmexQuoteMsg, BitmexTableMessage,
71            BitmexWsMessage, OrderData,
72        },
73        parse::{
74            ParsedOrderEvent, parse_book_msg_vec, parse_book10_msg_vec, parse_execution_msg,
75            parse_funding_msg, parse_instrument_msg, parse_order_event, parse_order_msg,
76            parse_order_update_msg, parse_position_msg, parse_trade_bin_msg_vec,
77            parse_trade_msg_vec, parse_wallet_msg,
78        },
79    },
80};
81
82/// Python wrapper around [`BitmexWebSocketClient`] that holds an instrument cache
83/// at the Python boundary for parsing venue messages into Nautilus domain types.
84#[pyclass(
85    name = "BitmexWebSocketClient",
86    module = "nautilus_trader.core.nautilus_pyo3.bitmex"
87)]
88#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bitmex")]
89pub struct PyBitmexWebSocketClient {
90    inner: BitmexWebSocketClient,
91    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
92    ws_dispatch_state: Arc<WsDispatchState>,
93}
94
95impl Debug for PyBitmexWebSocketClient {
96    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
97        f.debug_struct(stringify!(PyBitmexWebSocketClient))
98            .field("inner", &self.inner)
99            .finish_non_exhaustive()
100    }
101}
102
103#[pymethods]
104#[pyo3_stub_gen::derive::gen_stub_pymethods]
105impl PyBitmexWebSocketClient {
106    #[new]
107    #[pyo3(signature = (url=None, api_key=None, api_secret=None, account_id=None, heartbeat=5, environment=BitmexEnvironment::Mainnet, proxy_url=None))]
108    fn py_new(
109        url: Option<String>,
110        api_key: Option<String>,
111        api_secret: Option<String>,
112        account_id: Option<AccountId>,
113        heartbeat: u64,
114        environment: BitmexEnvironment,
115        proxy_url: Option<String>,
116    ) -> PyResult<Self> {
117        let inner = BitmexWebSocketClient::new_with_env(
118            url,
119            api_key,
120            api_secret,
121            account_id,
122            heartbeat,
123            environment,
124            TransportBackend::default(),
125            proxy_url,
126        )
127        .map_err(to_pyvalue_err)?;
128        Ok(Self {
129            inner,
130            instruments_cache: Arc::new(AtomicMap::new()),
131            ws_dispatch_state: Arc::new(WsDispatchState::default()),
132        })
133    }
134
135    #[staticmethod]
136    #[pyo3(name = "from_env")]
137    fn py_from_env() -> PyResult<Self> {
138        let inner = BitmexWebSocketClient::from_env().map_err(to_pyvalue_err)?;
139        Ok(Self {
140            inner,
141            instruments_cache: Arc::new(AtomicMap::new()),
142            ws_dispatch_state: Arc::new(WsDispatchState::default()),
143        })
144    }
145
146    #[getter]
147    #[pyo3(name = "url")]
148    #[must_use]
149    fn py_url(&self) -> &str {
150        self.inner.url()
151    }
152
153    #[getter]
154    #[pyo3(name = "api_key")]
155    #[must_use]
156    fn py_api_key(&self) -> Option<&str> {
157        self.inner.api_key()
158    }
159
160    #[getter]
161    #[pyo3(name = "api_key_masked")]
162    #[must_use]
163    fn py_api_key_masked(&self) -> Option<String> {
164        self.inner.api_key_masked()
165    }
166
167    #[pyo3(name = "is_active")]
168    fn py_is_active(&mut self) -> bool {
169        self.inner.is_active()
170    }
171
172    #[pyo3(name = "is_closed")]
173    fn py_is_closed(&mut self) -> bool {
174        self.inner.is_closed()
175    }
176
177    #[pyo3(name = "get_subscriptions")]
178    fn py_get_subscriptions(&self, instrument_id: InstrumentId) -> Vec<String> {
179        self.inner.get_subscriptions(instrument_id)
180    }
181
182    #[pyo3(name = "set_account_id")]
183    fn py_set_account_id(&mut self, account_id: AccountId) {
184        self.inner.set_account_id(account_id);
185    }
186
187    #[pyo3(name = "register_order_identity")]
188    fn py_register_order_identity(
189        &self,
190        client_order_id: ClientOrderId,
191        instrument_id: InstrumentId,
192        strategy_id: StrategyId,
193        order_side: OrderSide,
194        order_type: OrderType,
195    ) {
196        self.ws_dispatch_state.order_identities.insert(
197            client_order_id,
198            OrderIdentity {
199                instrument_id,
200                strategy_id,
201                order_side,
202                order_type,
203            },
204        );
205    }
206
207    #[pyo3(name = "remove_order_identity")]
208    fn py_remove_order_identity(&self, client_order_id: ClientOrderId) {
209        self.ws_dispatch_state
210            .order_identities
211            .remove(&client_order_id);
212    }
213
214    #[pyo3(name = "cache_instrument")]
215    fn py_cache_instrument(&self, py: Python, instrument: Py<PyAny>) -> PyResult<()> {
216        let inst = pyobject_to_instrument_any(py, instrument)?;
217        let symbol = inst.symbol().inner();
218        self.instruments_cache.insert(symbol, inst);
219        Ok(())
220    }
221
222    #[pyo3(name = "connect")]
223    #[pyo3(signature = (loop_, instruments, callback, trader_id=None))]
224    #[expect(clippy::needless_pass_by_value)]
225    fn py_connect<'py>(
226        &mut self,
227        py: Python<'py>,
228        loop_: Py<PyAny>,
229        instruments: Vec<Py<PyAny>>,
230        callback: Py<PyAny>,
231        trader_id: Option<TraderId>,
232    ) -> PyResult<Bound<'py, PyAny>> {
233        let call_soon: Py<PyAny> = loop_.getattr(py, "call_soon_threadsafe")?;
234
235        let cache = Arc::clone(&self.instruments_cache);
236        {
237            let mut initial: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
238
239            for inst_py in instruments {
240                let inst = pyobject_to_instrument_any(py, inst_py)?;
241                initial.insert(inst.symbol().inner(), inst);
242            }
243            cache.rcu(|m| {
244                for (k, v) in &initial {
245                    m.insert(*k, v.clone());
246                }
247            });
248        }
249
250        let clock = get_atomic_clock_realtime();
251        let mut client = self.inner.clone();
252        let account_id = self.inner.account_id();
253        let dispatch_state = Arc::clone(&self.ws_dispatch_state);
254        let trader_id = trader_id.unwrap_or(TraderId::from("TRADER-000"));
255
256        pyo3_async_runtimes::tokio::future_into_py(py, async move {
257            client.connect().await.map_err(to_pyruntime_err)?;
258
259            let stream = client.stream();
260
261            get_runtime().spawn(async move {
262                let _client = client; // Keep client alive for the entire duration
263                tokio::pin!(stream);
264
265                let mut quote_cache = QuoteCache::new();
266                let mut order_type_cache: AHashMap<ClientOrderId, OrderType> = AHashMap::new();
267                let mut order_symbol_cache: AHashMap<ClientOrderId, Ustr> = AHashMap::new();
268
269                while let Some(msg) = stream.next().await {
270                    let ts_init = clock.get_time_ns();
271
272                    match msg {
273                        BitmexWsMessage::Table(table_msg) => {
274                            handle_table_message(
275                                table_msg,
276                                &cache,
277                                &mut quote_cache,
278                                &mut order_type_cache,
279                                &mut order_symbol_cache,
280                                &dispatch_state,
281                                trader_id,
282                                account_id,
283                                ts_init,
284                                &call_soon,
285                                &callback,
286                            );
287                        }
288                        BitmexWsMessage::Reconnected => {
289                            quote_cache.clear();
290                            order_type_cache.clear();
291                            order_symbol_cache.clear();
292                        }
293                        BitmexWsMessage::Authenticated => {}
294                    }
295                }
296            });
297
298            Ok(())
299        })
300    }
301
302    #[pyo3(name = "wait_until_active")]
303    fn py_wait_until_active<'py>(
304        &self,
305        py: Python<'py>,
306        timeout_secs: f64,
307    ) -> PyResult<Bound<'py, PyAny>> {
308        let client = self.inner.clone();
309
310        pyo3_async_runtimes::tokio::future_into_py(py, async move {
311            client
312                .wait_until_active(timeout_secs)
313                .await
314                .map_err(to_pyruntime_err)?;
315            Ok(())
316        })
317    }
318
319    #[pyo3(name = "close")]
320    fn py_close<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
321        let mut client = self.inner.clone();
322
323        pyo3_async_runtimes::tokio::future_into_py(py, async move {
324            if let Err(e) = client.close().await {
325                log::warn!("Error on close: {e}");
326            }
327            Ok(())
328        })
329    }
330
331    #[pyo3(name = "subscribe_instruments")]
332    fn py_subscribe_instruments<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
333        let client = self.inner.clone();
334
335        pyo3_async_runtimes::tokio::future_into_py(py, async move {
336            if let Err(e) = client.subscribe_instruments().await {
337                log::error!("Failed to subscribe to instruments: {e}");
338            }
339            Ok(())
340        })
341    }
342
343    #[pyo3(name = "subscribe_instrument")]
344    fn py_subscribe_instrument<'py>(
345        &self,
346        py: Python<'py>,
347        instrument_id: InstrumentId,
348    ) -> PyResult<Bound<'py, PyAny>> {
349        let client = self.inner.clone();
350
351        pyo3_async_runtimes::tokio::future_into_py(py, async move {
352            if let Err(e) = client.subscribe_instrument(instrument_id).await {
353                log::error!("Failed to subscribe to instrument: {e}");
354            }
355            Ok(())
356        })
357    }
358
359    #[pyo3(name = "subscribe_book")]
360    fn py_subscribe_book<'py>(
361        &self,
362        py: Python<'py>,
363        instrument_id: InstrumentId,
364    ) -> PyResult<Bound<'py, PyAny>> {
365        let client = self.inner.clone();
366
367        pyo3_async_runtimes::tokio::future_into_py(py, async move {
368            if let Err(e) = client.subscribe_book(instrument_id).await {
369                log::error!("Failed to subscribe to order book: {e}");
370            }
371            Ok(())
372        })
373    }
374
375    #[pyo3(name = "subscribe_book_25")]
376    fn py_subscribe_book_25<'py>(
377        &self,
378        py: Python<'py>,
379        instrument_id: InstrumentId,
380    ) -> PyResult<Bound<'py, PyAny>> {
381        let client = self.inner.clone();
382
383        pyo3_async_runtimes::tokio::future_into_py(py, async move {
384            if let Err(e) = client.subscribe_book_25(instrument_id).await {
385                log::error!("Failed to subscribe to order book 25: {e}");
386            }
387            Ok(())
388        })
389    }
390
391    #[pyo3(name = "subscribe_book_depth10")]
392    fn py_subscribe_book_depth10<'py>(
393        &self,
394        py: Python<'py>,
395        instrument_id: InstrumentId,
396    ) -> PyResult<Bound<'py, PyAny>> {
397        let client = self.inner.clone();
398
399        pyo3_async_runtimes::tokio::future_into_py(py, async move {
400            if let Err(e) = client.subscribe_book_depth10(instrument_id).await {
401                log::error!("Failed to subscribe to order book depth 10: {e}");
402            }
403            Ok(())
404        })
405    }
406
407    #[pyo3(name = "subscribe_quotes")]
408    fn py_subscribe_quotes<'py>(
409        &self,
410        py: Python<'py>,
411        instrument_id: InstrumentId,
412    ) -> PyResult<Bound<'py, PyAny>> {
413        let client = self.inner.clone();
414
415        pyo3_async_runtimes::tokio::future_into_py(py, async move {
416            if let Err(e) = client.subscribe_quotes(instrument_id).await {
417                log::error!("Failed to subscribe to quotes: {e}");
418            }
419            Ok(())
420        })
421    }
422
423    #[pyo3(name = "subscribe_trades")]
424    fn py_subscribe_trades<'py>(
425        &self,
426        py: Python<'py>,
427        instrument_id: InstrumentId,
428    ) -> PyResult<Bound<'py, PyAny>> {
429        let client = self.inner.clone();
430
431        pyo3_async_runtimes::tokio::future_into_py(py, async move {
432            if let Err(e) = client.subscribe_trades(instrument_id).await {
433                log::error!("Failed to subscribe to trades: {e}");
434            }
435            Ok(())
436        })
437    }
438
439    #[pyo3(name = "subscribe_mark_prices")]
440    fn py_subscribe_mark_prices<'py>(
441        &self,
442        py: Python<'py>,
443        instrument_id: InstrumentId,
444    ) -> PyResult<Bound<'py, PyAny>> {
445        let client = self.inner.clone();
446
447        pyo3_async_runtimes::tokio::future_into_py(py, async move {
448            if let Err(e) = client.subscribe_mark_prices(instrument_id).await {
449                log::error!("Failed to subscribe to mark prices: {e}");
450            }
451            Ok(())
452        })
453    }
454
455    #[pyo3(name = "subscribe_index_prices")]
456    fn py_subscribe_index_prices<'py>(
457        &self,
458        py: Python<'py>,
459        instrument_id: InstrumentId,
460    ) -> PyResult<Bound<'py, PyAny>> {
461        let client = self.inner.clone();
462
463        pyo3_async_runtimes::tokio::future_into_py(py, async move {
464            if let Err(e) = client.subscribe_index_prices(instrument_id).await {
465                log::error!("Failed to subscribe to index prices: {e}");
466            }
467            Ok(())
468        })
469    }
470
471    #[pyo3(name = "subscribe_funding_rates")]
472    fn py_subscribe_funding_rates<'py>(
473        &self,
474        py: Python<'py>,
475        instrument_id: InstrumentId,
476    ) -> PyResult<Bound<'py, PyAny>> {
477        let client = self.inner.clone();
478
479        pyo3_async_runtimes::tokio::future_into_py(py, async move {
480            if let Err(e) = client.subscribe_funding_rates(instrument_id).await {
481                log::error!("Failed to subscribe to funding: {e}");
482            }
483            Ok(())
484        })
485    }
486
487    #[pyo3(name = "subscribe_bars")]
488    fn py_subscribe_bars<'py>(
489        &self,
490        py: Python<'py>,
491        bar_type: BarType,
492    ) -> PyResult<Bound<'py, PyAny>> {
493        let client = self.inner.clone();
494
495        pyo3_async_runtimes::tokio::future_into_py(py, async move {
496            if let Err(e) = client.subscribe_bars(bar_type).await {
497                log::error!("Failed to subscribe to bars: {e}");
498            }
499            Ok(())
500        })
501    }
502
503    #[pyo3(name = "unsubscribe_instruments")]
504    fn py_unsubscribe_instruments<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
505        let client = self.inner.clone();
506
507        pyo3_async_runtimes::tokio::future_into_py(py, async move {
508            if let Err(e) = client.unsubscribe_instruments().await {
509                log::error!("Failed to unsubscribe from instruments: {e}");
510            }
511            Ok(())
512        })
513    }
514
515    #[pyo3(name = "unsubscribe_instrument")]
516    fn py_unsubscribe_instrument<'py>(
517        &self,
518        py: Python<'py>,
519        instrument_id: InstrumentId,
520    ) -> PyResult<Bound<'py, PyAny>> {
521        let client = self.inner.clone();
522
523        pyo3_async_runtimes::tokio::future_into_py(py, async move {
524            if let Err(e) = client.unsubscribe_instrument(instrument_id).await {
525                log::error!("Failed to unsubscribe from instrument: {e}");
526            }
527            Ok(())
528        })
529    }
530
531    #[pyo3(name = "unsubscribe_book")]
532    fn py_unsubscribe_book<'py>(
533        &self,
534        py: Python<'py>,
535        instrument_id: InstrumentId,
536    ) -> PyResult<Bound<'py, PyAny>> {
537        let client = self.inner.clone();
538
539        pyo3_async_runtimes::tokio::future_into_py(py, async move {
540            if let Err(e) = client.unsubscribe_book(instrument_id).await {
541                log::error!("Failed to unsubscribe from order book: {e}");
542            }
543            Ok(())
544        })
545    }
546
547    #[pyo3(name = "unsubscribe_book_25")]
548    fn py_unsubscribe_book_25<'py>(
549        &self,
550        py: Python<'py>,
551        instrument_id: InstrumentId,
552    ) -> PyResult<Bound<'py, PyAny>> {
553        let client = self.inner.clone();
554
555        pyo3_async_runtimes::tokio::future_into_py(py, async move {
556            if let Err(e) = client.unsubscribe_book_25(instrument_id).await {
557                log::error!("Failed to unsubscribe from order book 25: {e}");
558            }
559            Ok(())
560        })
561    }
562
563    #[pyo3(name = "unsubscribe_book_depth10")]
564    fn py_unsubscribe_book_depth10<'py>(
565        &self,
566        py: Python<'py>,
567        instrument_id: InstrumentId,
568    ) -> PyResult<Bound<'py, PyAny>> {
569        let client = self.inner.clone();
570
571        pyo3_async_runtimes::tokio::future_into_py(py, async move {
572            if let Err(e) = client.unsubscribe_book_depth10(instrument_id).await {
573                log::error!("Failed to unsubscribe from order book depth 10: {e}");
574            }
575            Ok(())
576        })
577    }
578
579    #[pyo3(name = "unsubscribe_quotes")]
580    fn py_unsubscribe_quotes<'py>(
581        &self,
582        py: Python<'py>,
583        instrument_id: InstrumentId,
584    ) -> PyResult<Bound<'py, PyAny>> {
585        let client = self.inner.clone();
586
587        pyo3_async_runtimes::tokio::future_into_py(py, async move {
588            if let Err(e) = client.unsubscribe_quotes(instrument_id).await {
589                log::error!("Failed to unsubscribe from quotes: {e}");
590            }
591            Ok(())
592        })
593    }
594
595    #[pyo3(name = "unsubscribe_trades")]
596    fn py_unsubscribe_trades<'py>(
597        &self,
598        py: Python<'py>,
599        instrument_id: InstrumentId,
600    ) -> PyResult<Bound<'py, PyAny>> {
601        let client = self.inner.clone();
602
603        pyo3_async_runtimes::tokio::future_into_py(py, async move {
604            if let Err(e) = client.unsubscribe_trades(instrument_id).await {
605                log::error!("Failed to unsubscribe from trades: {e}");
606            }
607            Ok(())
608        })
609    }
610
611    #[pyo3(name = "unsubscribe_mark_prices")]
612    fn py_unsubscribe_mark_prices<'py>(
613        &self,
614        py: Python<'py>,
615        instrument_id: InstrumentId,
616    ) -> PyResult<Bound<'py, PyAny>> {
617        let client = self.inner.clone();
618
619        pyo3_async_runtimes::tokio::future_into_py(py, async move {
620            if let Err(e) = client.unsubscribe_mark_prices(instrument_id).await {
621                log::error!("Failed to unsubscribe from mark prices: {e}");
622            }
623            Ok(())
624        })
625    }
626
627    #[pyo3(name = "unsubscribe_index_prices")]
628    fn py_unsubscribe_index_prices<'py>(
629        &self,
630        py: Python<'py>,
631        instrument_id: InstrumentId,
632    ) -> PyResult<Bound<'py, PyAny>> {
633        let client = self.inner.clone();
634
635        pyo3_async_runtimes::tokio::future_into_py(py, async move {
636            if let Err(e) = client.unsubscribe_index_prices(instrument_id).await {
637                log::error!("Failed to unsubscribe from index prices: {e}");
638            }
639            Ok(())
640        })
641    }
642
643    #[pyo3(name = "unsubscribe_funding_rates")]
644    fn py_unsubscribe_funding_rates<'py>(
645        &self,
646        py: Python<'py>,
647        instrument_id: InstrumentId,
648    ) -> PyResult<Bound<'py, PyAny>> {
649        let client = self.inner.clone();
650        pyo3_async_runtimes::tokio::future_into_py(py, async move {
651            if let Err(e) = client.unsubscribe_funding_rates(instrument_id).await {
652                log::error!("Failed to unsubscribe from funding rates: {e}");
653            }
654            Ok(())
655        })
656    }
657
658    #[pyo3(name = "unsubscribe_bars")]
659    fn py_unsubscribe_bars<'py>(
660        &self,
661        py: Python<'py>,
662        bar_type: BarType,
663    ) -> PyResult<Bound<'py, PyAny>> {
664        let client = self.inner.clone();
665
666        pyo3_async_runtimes::tokio::future_into_py(py, async move {
667            if let Err(e) = client.unsubscribe_bars(bar_type).await {
668                log::error!("Failed to unsubscribe from bars: {e}");
669            }
670            Ok(())
671        })
672    }
673
674    #[pyo3(name = "subscribe_orders")]
675    fn py_subscribe_orders<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
676        let client = self.inner.clone();
677
678        pyo3_async_runtimes::tokio::future_into_py(py, async move {
679            if let Err(e) = client.subscribe_orders().await {
680                log::error!("Failed to subscribe to orders: {e}");
681            }
682            Ok(())
683        })
684    }
685
686    #[pyo3(name = "subscribe_executions")]
687    fn py_subscribe_executions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
688        let client = self.inner.clone();
689
690        pyo3_async_runtimes::tokio::future_into_py(py, async move {
691            if let Err(e) = client.subscribe_executions().await {
692                log::error!("Failed to subscribe to executions: {e}");
693            }
694            Ok(())
695        })
696    }
697
698    #[pyo3(name = "subscribe_positions")]
699    fn py_subscribe_positions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
700        let client = self.inner.clone();
701
702        pyo3_async_runtimes::tokio::future_into_py(py, async move {
703            if let Err(e) = client.subscribe_positions().await {
704                log::error!("Failed to subscribe to positions: {e}");
705            }
706            Ok(())
707        })
708    }
709
710    #[pyo3(name = "subscribe_margin")]
711    fn py_subscribe_margin<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
712        let client = self.inner.clone();
713
714        pyo3_async_runtimes::tokio::future_into_py(py, async move {
715            if let Err(e) = client.subscribe_margin().await {
716                log::error!("Failed to subscribe to margin: {e}");
717            }
718            Ok(())
719        })
720    }
721
722    #[pyo3(name = "subscribe_wallet")]
723    fn py_subscribe_wallet<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
724        let client = self.inner.clone();
725
726        pyo3_async_runtimes::tokio::future_into_py(py, async move {
727            if let Err(e) = client.subscribe_wallet().await {
728                log::error!("Failed to subscribe to wallet: {e}");
729            }
730            Ok(())
731        })
732    }
733
734    #[pyo3(name = "unsubscribe_orders")]
735    fn py_unsubscribe_orders<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
736        let client = self.inner.clone();
737
738        pyo3_async_runtimes::tokio::future_into_py(py, async move {
739            if let Err(e) = client.unsubscribe_orders().await {
740                log::error!("Failed to unsubscribe from orders: {e}");
741            }
742            Ok(())
743        })
744    }
745
746    #[pyo3(name = "unsubscribe_executions")]
747    fn py_unsubscribe_executions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
748        let client = self.inner.clone();
749
750        pyo3_async_runtimes::tokio::future_into_py(py, async move {
751            if let Err(e) = client.unsubscribe_executions().await {
752                log::error!("Failed to unsubscribe from executions: {e}");
753            }
754            Ok(())
755        })
756    }
757
758    #[pyo3(name = "unsubscribe_positions")]
759    fn py_unsubscribe_positions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
760        let client = self.inner.clone();
761
762        pyo3_async_runtimes::tokio::future_into_py(py, async move {
763            if let Err(e) = client.unsubscribe_positions().await {
764                log::error!("Failed to unsubscribe from positions: {e}");
765            }
766            Ok(())
767        })
768    }
769
770    #[pyo3(name = "unsubscribe_margin")]
771    fn py_unsubscribe_margin<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
772        let client = self.inner.clone();
773
774        pyo3_async_runtimes::tokio::future_into_py(py, async move {
775            if let Err(e) = client.unsubscribe_margin().await {
776                log::error!("Failed to unsubscribe from margin: {e}");
777            }
778            Ok(())
779        })
780    }
781
782    #[pyo3(name = "unsubscribe_wallet")]
783    fn py_unsubscribe_wallet<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
784        let client = self.inner.clone();
785
786        pyo3_async_runtimes::tokio::future_into_py(py, async move {
787            if let Err(e) = client.unsubscribe_wallet().await {
788                log::error!("Failed to unsubscribe from wallet: {e}");
789            }
790            Ok(())
791        })
792    }
793}
794
795#[expect(clippy::too_many_arguments)]
796fn handle_table_message(
797    table_msg: BitmexTableMessage,
798    instruments_cache: &Arc<AtomicMap<Ustr, InstrumentAny>>,
799    quote_cache: &mut QuoteCache,
800    order_type_cache: &mut AHashMap<ClientOrderId, OrderType>,
801    order_symbol_cache: &mut AHashMap<ClientOrderId, Ustr>,
802    dispatch_state: &WsDispatchState,
803    trader_id: TraderId,
804    account_id: AccountId,
805    ts_init: UnixNanos,
806    call_soon: &Py<PyAny>,
807    callback: &Py<PyAny>,
808) {
809    if let BitmexTableMessage::Instrument { action, data } = table_msg {
810        handle_instrument_messages(
811            action,
812            data,
813            instruments_cache,
814            ts_init,
815            call_soon,
816            callback,
817        );
818        return;
819    }
820
821    let instruments = instruments_cache.load();
822
823    match table_msg {
824        BitmexTableMessage::OrderBookL2 { action, data }
825        | BitmexTableMessage::OrderBookL2_25 { action, data } => {
826            if !data.is_empty() {
827                for d in parse_book_msg_vec(data, action, &instruments, ts_init) {
828                    send_data_to_python(d, call_soon, callback);
829                }
830            }
831        }
832        BitmexTableMessage::OrderBook10 { data, .. } => {
833            if !data.is_empty() {
834                for d in parse_book10_msg_vec(data, &instruments, ts_init) {
835                    send_data_to_python(d, call_soon, callback);
836                }
837            }
838        }
839        BitmexTableMessage::Quote { data, .. } => {
840            handle_quote_messages(
841                data,
842                &instruments,
843                quote_cache,
844                ts_init,
845                call_soon,
846                callback,
847            );
848        }
849        BitmexTableMessage::Trade { data, .. } => {
850            if !data.is_empty() {
851                for d in parse_trade_msg_vec(data, &instruments, ts_init) {
852                    send_data_to_python(d, call_soon, callback);
853                }
854            }
855        }
856        BitmexTableMessage::TradeBin1m { action, data } => {
857            if action != BitmexAction::Partial && !data.is_empty() {
858                for d in
859                    parse_trade_bin_msg_vec(data, &BitmexWsTopic::TradeBin1m, &instruments, ts_init)
860                {
861                    send_data_to_python(d, call_soon, callback);
862                }
863            }
864        }
865        BitmexTableMessage::TradeBin5m { action, data } => {
866            if action != BitmexAction::Partial && !data.is_empty() {
867                for d in
868                    parse_trade_bin_msg_vec(data, &BitmexWsTopic::TradeBin5m, &instruments, ts_init)
869                {
870                    send_data_to_python(d, call_soon, callback);
871                }
872            }
873        }
874        BitmexTableMessage::TradeBin1h { action, data } => {
875            if action != BitmexAction::Partial && !data.is_empty() {
876                for d in
877                    parse_trade_bin_msg_vec(data, &BitmexWsTopic::TradeBin1h, &instruments, ts_init)
878                {
879                    send_data_to_python(d, call_soon, callback);
880                }
881            }
882        }
883        BitmexTableMessage::TradeBin1d { action, data } => {
884            if action != BitmexAction::Partial && !data.is_empty() {
885                for d in
886                    parse_trade_bin_msg_vec(data, &BitmexWsTopic::TradeBin1d, &instruments, ts_init)
887                {
888                    send_data_to_python(d, call_soon, callback);
889                }
890            }
891        }
892        BitmexTableMessage::Funding { data, .. } => {
893            for msg in data {
894                send_to_python(parse_funding_msg(&msg, ts_init), call_soon, callback);
895            }
896        }
897        BitmexTableMessage::Order { data, .. } => {
898            handle_order_messages(
899                data,
900                &instruments,
901                order_type_cache,
902                order_symbol_cache,
903                dispatch_state,
904                trader_id,
905                account_id,
906                ts_init,
907                call_soon,
908                callback,
909            );
910        }
911        BitmexTableMessage::Execution { data, .. } => {
912            handle_execution_messages(
913                data,
914                &instruments,
915                order_symbol_cache,
916                dispatch_state,
917                trader_id,
918                account_id,
919                ts_init,
920                call_soon,
921                callback,
922            );
923        }
924        BitmexTableMessage::Position { data, .. } => {
925            for msg in data {
926                let Some(instrument) = instruments.get(&msg.symbol) else {
927                    log::warn!("Instrument cache miss for position symbol={}", msg.symbol);
928                    continue;
929                };
930
931                let mut report = parse_position_msg(&msg, instrument, ts_init);
932                report.account_id = account_id;
933                send_to_python(report, call_soon, callback);
934            }
935        }
936        BitmexTableMessage::Wallet { data, .. } => {
937            for msg in data {
938                let mut account_state = parse_wallet_msg(&msg, ts_init);
939                account_state.account_id = account_id;
940                send_to_python(account_state, call_soon, callback);
941            }
942        }
943        BitmexTableMessage::Margin { .. } => {}
944        _ => {
945            log::debug!("Unhandled table message type in Python WebSocket client");
946        }
947    }
948}
949
950fn handle_quote_messages(
951    data: Vec<BitmexQuoteMsg>,
952    instruments: &AHashMap<Ustr, InstrumentAny>,
953    quote_cache: &mut QuoteCache,
954    ts_init: UnixNanos,
955    call_soon: &Py<PyAny>,
956    callback: &Py<PyAny>,
957) {
958    for msg in data {
959        let Some(instrument) = instruments.get(&msg.symbol) else {
960            log::error!(
961                "Instrument cache miss: quote dropped for symbol={}",
962                msg.symbol,
963            );
964            continue;
965        };
966
967        let instrument_id = instrument.id();
968        let price_precision = instrument.price_precision();
969
970        let bid_price = msg.bid_price.map(|p| Price::new(p, price_precision));
971        let ask_price = msg.ask_price.map(|p| Price::new(p, price_precision));
972        let bid_size = msg
973            .bid_size
974            .map(|s| parse_contracts_quantity(s, instrument));
975        let ask_size = msg
976            .ask_size
977            .map(|s| parse_contracts_quantity(s, instrument));
978        let ts_event = UnixNanos::from(msg.timestamp);
979
980        match quote_cache.process(
981            instrument_id,
982            bid_price,
983            ask_price,
984            bid_size,
985            ask_size,
986            ts_event,
987            ts_init,
988        ) {
989            Ok(quote) => send_data_to_python(Data::Quote(quote), call_soon, callback),
990            Err(e) => {
991                log::warn!("Failed to process quote for {}: {e}", msg.symbol);
992            }
993        }
994    }
995}
996
997fn handle_instrument_messages(
998    action: BitmexAction,
999    data: Vec<BitmexInstrumentMsg>,
1000    instruments_cache: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1001    ts_init: UnixNanos,
1002    call_soon: &Py<PyAny>,
1003    callback: &Py<PyAny>,
1004) {
1005    if action == BitmexAction::Partial || action == BitmexAction::Insert {
1006        let data_for_prices = data.clone();
1007
1008        let mut new_instruments: Vec<(Ustr, InstrumentAny)> = Vec::new();
1009
1010        for msg in data {
1011            match msg.try_into() {
1012                Ok(http_inst) => match parse_instrument_any(&http_inst, ts_init) {
1013                    InstrumentParseResult::Ok(boxed) => {
1014                        let inst = *boxed;
1015                        let symbol = inst.symbol().inner();
1016                        new_instruments.push((symbol, inst));
1017                    }
1018                    InstrumentParseResult::Unsupported { .. }
1019                    | InstrumentParseResult::Inactive { .. } => {}
1020                    InstrumentParseResult::Failed { symbol, error, .. } => {
1021                        log::warn!("Failed to parse instrument {symbol}: {error}");
1022                    }
1023                },
1024                Err(e) => {
1025                    log::debug!("Skipping instrument (missing required fields): {e}");
1026                }
1027            }
1028        }
1029
1030        instruments_cache.rcu(|m| {
1031            for (symbol, inst) in &new_instruments {
1032                m.insert(*symbol, inst.clone());
1033            }
1034        });
1035
1036        for (_, inst) in &new_instruments {
1037            Python::attach(|py| {
1038                if let Ok(py_obj) = instrument_any_to_pyobject(py, inst.clone()) {
1039                    call_python_threadsafe(py, call_soon, callback, py_obj);
1040                }
1041            });
1042        }
1043
1044        let cache = instruments_cache.load();
1045        for msg in data_for_prices {
1046            for d in parse_instrument_msg(&msg, &cache, ts_init) {
1047                send_data_to_python(d, call_soon, callback);
1048            }
1049        }
1050    } else {
1051        for msg in &data {
1052            if let Some(state_str) = &msg.state
1053                && let Ok(state) =
1054                    serde_json::from_str::<BitmexInstrumentState>(&format!("\"{state_str}\""))
1055            {
1056                let instrument_id = parse_instrument_id(msg.symbol);
1057                let action = MarketStatusAction::from(&state);
1058                let is_trading = Some(state == BitmexInstrumentState::Open);
1059                let ts_event =
1060                    parse_optional_datetime_to_unix_nanos(&Some(msg.timestamp), "timestamp");
1061                let status = InstrumentStatus::new(
1062                    instrument_id,
1063                    action,
1064                    ts_event,
1065                    ts_init,
1066                    None,
1067                    None,
1068                    is_trading,
1069                    None,
1070                    None,
1071                );
1072                send_to_python(status, call_soon, callback);
1073            }
1074        }
1075
1076        let cache = instruments_cache.load();
1077        for msg in data {
1078            for d in parse_instrument_msg(&msg, &cache, ts_init) {
1079                send_data_to_python(d, call_soon, callback);
1080            }
1081        }
1082    }
1083}
1084
1085#[expect(clippy::too_many_arguments)]
1086fn handle_order_messages(
1087    data: Vec<OrderData>,
1088    instruments: &AHashMap<Ustr, InstrumentAny>,
1089    order_type_cache: &mut AHashMap<ClientOrderId, OrderType>,
1090    order_symbol_cache: &mut AHashMap<ClientOrderId, Ustr>,
1091    dispatch_state: &WsDispatchState,
1092    trader_id: TraderId,
1093    account_id: AccountId,
1094    ts_init: UnixNanos,
1095    call_soon: &Py<PyAny>,
1096    callback: &Py<PyAny>,
1097) {
1098    for order_data in data {
1099        match order_data {
1100            OrderData::Full(order_msg) => {
1101                let Some(instrument) = instruments.get(&order_msg.symbol) else {
1102                    log::warn!(
1103                        "Instrument cache miss for order symbol={}",
1104                        order_msg.symbol
1105                    );
1106                    continue;
1107                };
1108
1109                let client_order_id = order_msg.cl_ord_id.map(ClientOrderId::new);
1110
1111                if let Some(ref cid) = client_order_id {
1112                    if let Some(ord_type) = &order_msg.ord_type {
1113                        let order_type: OrderType = if *ord_type == BitmexOrderType::Pegged
1114                            && order_msg.peg_price_type == Some(BitmexPegPriceType::TrailingStopPeg)
1115                        {
1116                            if order_msg.price.is_some() {
1117                                OrderType::TrailingStopLimit
1118                            } else {
1119                                OrderType::TrailingStopMarket
1120                            }
1121                        } else {
1122                            (*ord_type).into()
1123                        };
1124                        order_type_cache.insert(*cid, order_type);
1125                    }
1126                    order_symbol_cache.insert(*cid, order_msg.symbol);
1127                }
1128
1129                let identity = client_order_id.and_then(|cid| {
1130                    dispatch_state
1131                        .order_identities
1132                        .get(&cid)
1133                        .map(|r| (cid, r.clone()))
1134                });
1135
1136                if let Some((cid, ident)) = identity {
1137                    if let Some(event) = parse_order_event(
1138                        &order_msg,
1139                        cid,
1140                        account_id,
1141                        trader_id,
1142                        ident.strategy_id,
1143                        ts_init,
1144                    ) {
1145                        let venue_order_id = VenueOrderId::new(order_msg.order_id.to_string());
1146                        dispatch_order_event_to_python(
1147                            event,
1148                            cid,
1149                            account_id,
1150                            venue_order_id,
1151                            &ident,
1152                            dispatch_state,
1153                            trader_id,
1154                            ts_init,
1155                            call_soon,
1156                            callback,
1157                        );
1158                    }
1159
1160                    if order_msg.ord_status.is_terminal() {
1161                        order_type_cache.remove(&cid);
1162                        order_symbol_cache.remove(&cid);
1163                    }
1164                } else {
1165                    match parse_order_msg(&order_msg, instrument, order_type_cache, ts_init) {
1166                        Ok(mut report) => {
1167                            if report.order_status.is_closed()
1168                                && let Some(cid) = report.client_order_id
1169                            {
1170                                order_type_cache.remove(&cid);
1171                                order_symbol_cache.remove(&cid);
1172                            }
1173                            report.account_id = account_id;
1174                            send_to_python(report, call_soon, callback);
1175                        }
1176                        Err(e) => log::error!("Failed to parse order message: {e}"),
1177                    }
1178                }
1179            }
1180            OrderData::Update(msg) => {
1181                if let Some(cl_ord_id) = &msg.cl_ord_id {
1182                    let cid = ClientOrderId::new(cl_ord_id);
1183                    order_symbol_cache.insert(cid, msg.symbol);
1184                }
1185
1186                let Some(instrument) = instruments.get(&msg.symbol) else {
1187                    log::warn!(
1188                        "Instrument cache miss for order update symbol={}",
1189                        msg.symbol,
1190                    );
1191                    continue;
1192                };
1193
1194                let identity = msg.cl_ord_id.as_ref().and_then(|cl| {
1195                    let cid = ClientOrderId::new(cl);
1196                    dispatch_state
1197                        .order_identities
1198                        .get(&cid)
1199                        .map(|r| (cid, r.clone()))
1200                });
1201
1202                if let Some((cid, ident)) = identity {
1203                    if let Some(event) =
1204                        parse_order_update_msg(&msg, instrument, account_id, ts_init)
1205                    {
1206                        let enriched = OrderUpdated::new(
1207                            trader_id,
1208                            ident.strategy_id,
1209                            event.instrument_id,
1210                            cid,
1211                            event.quantity,
1212                            event.event_id,
1213                            event.ts_event,
1214                            event.ts_init,
1215                            false,
1216                            event.venue_order_id,
1217                            Some(account_id),
1218                            event.price,
1219                            event.trigger_price,
1220                            event.protection_price,
1221                            false, // is_quote_quantity
1222                        );
1223                        let venue_order_id = enriched
1224                            .venue_order_id
1225                            .unwrap_or_else(|| VenueOrderId::new(msg.order_id.to_string()));
1226                        ensure_accepted_to_python(
1227                            cid,
1228                            account_id,
1229                            venue_order_id,
1230                            &ident,
1231                            dispatch_state,
1232                            trader_id,
1233                            ts_init,
1234                            call_soon,
1235                            callback,
1236                        );
1237                        send_to_python(enriched, call_soon, callback);
1238                    }
1239                } else {
1240                    log::debug!(
1241                        "Skipping order update for untracked order: order_id={}",
1242                        msg.order_id,
1243                    );
1244                }
1245            }
1246        }
1247    }
1248}
1249
1250#[expect(clippy::too_many_arguments)]
1251fn handle_execution_messages(
1252    data: Vec<BitmexExecutionMsg>,
1253    instruments: &AHashMap<Ustr, InstrumentAny>,
1254    order_symbol_cache: &AHashMap<ClientOrderId, Ustr>,
1255    dispatch_state: &WsDispatchState,
1256    trader_id: TraderId,
1257    account_id: AccountId,
1258    ts_init: UnixNanos,
1259    call_soon: &Py<PyAny>,
1260    callback: &Py<PyAny>,
1261) {
1262    for exec_msg in data {
1263        let symbol = exec_msg.symbol.or_else(|| {
1264            exec_msg
1265                .cl_ord_id
1266                .map(ClientOrderId::new)
1267                .and_then(|cid| order_symbol_cache.get(&cid).copied())
1268        });
1269
1270        let Some(symbol) = symbol else {
1271            if let Some(cl_ord_id) = &exec_msg.cl_ord_id {
1272                if exec_msg.exec_type == Some(BitmexExecType::Trade) {
1273                    log::warn!(
1274                        "Execution missing symbol and not in cache: \
1275                        cl_ord_id={cl_ord_id}, exec_id={:?}",
1276                        exec_msg.exec_id,
1277                    );
1278                } else {
1279                    log::debug!(
1280                        "Execution missing symbol and not in cache: \
1281                        cl_ord_id={cl_ord_id}, exec_type={:?}",
1282                        exec_msg.exec_type,
1283                    );
1284                }
1285            } else if exec_msg.exec_type == Some(BitmexExecType::CancelReject) {
1286                log::debug!(
1287                    "CancelReject missing symbol/clOrdID: exec_id={:?}, order_id={:?}",
1288                    exec_msg.exec_id,
1289                    exec_msg.order_id,
1290                );
1291            } else {
1292                log::warn!(
1293                    "Execution missing both symbol and clOrdID: \
1294                    exec_id={:?}, order_id={:?}, exec_type={:?}",
1295                    exec_msg.exec_id,
1296                    exec_msg.order_id,
1297                    exec_msg.exec_type,
1298                );
1299            }
1300            continue;
1301        };
1302
1303        let Some(instrument) = instruments.get(&symbol) else {
1304            log::warn!("Instrument cache miss for execution symbol={symbol}");
1305            continue;
1306        };
1307
1308        let Some(mut fill) = parse_execution_msg(exec_msg, instrument, ts_init) else {
1309            continue;
1310        };
1311        fill.account_id = account_id;
1312
1313        let identity = fill.client_order_id.and_then(|cid| {
1314            dispatch_state
1315                .order_identities
1316                .get(&cid)
1317                .map(|r| (cid, r.clone()))
1318        });
1319
1320        if let Some((cid, ident)) = identity {
1321            let venue_order_id = fill.venue_order_id;
1322            ensure_accepted_to_python(
1323                cid,
1324                fill.account_id,
1325                venue_order_id,
1326                &ident,
1327                dispatch_state,
1328                trader_id,
1329                ts_init,
1330                call_soon,
1331                callback,
1332            );
1333            dispatch_state.insert_filled(cid);
1334            dispatch_state.remove_triggered(&cid);
1335            let filled =
1336                fill_report_to_order_filled(&fill, trader_id, &ident, instrument.quote_currency());
1337            send_to_python(filled, call_soon, callback);
1338        } else {
1339            send_to_python(fill, call_soon, callback);
1340        }
1341    }
1342}
1343
1344/// Dispatches a parsed order event to Python with lifecycle synthesis and deduplication.
1345#[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
1346fn dispatch_order_event_to_python(
1347    event: ParsedOrderEvent,
1348    client_order_id: ClientOrderId,
1349    account_id: AccountId,
1350    venue_order_id: VenueOrderId,
1351    identity: &OrderIdentity,
1352    state: &WsDispatchState,
1353    trader_id: TraderId,
1354    ts_init: UnixNanos,
1355    call_soon: &Py<PyAny>,
1356    callback: &Py<PyAny>,
1357) {
1358    let is_terminal;
1359
1360    match event {
1361        ParsedOrderEvent::Accepted(e) => {
1362            if state.accepted_contains(&client_order_id)
1363                || state.filled_contains(&client_order_id)
1364                || state.triggered_contains(&client_order_id)
1365            {
1366                log::debug!("Skipping duplicate Accepted for {client_order_id}");
1367                return;
1368            }
1369            state.insert_accepted(client_order_id);
1370            is_terminal = false;
1371            send_to_python(e, call_soon, callback);
1372        }
1373        ParsedOrderEvent::Triggered(e) => {
1374            if state.filled_contains(&client_order_id) {
1375                log::debug!("Skipping stale Triggered for {client_order_id} (already filled)");
1376                return;
1377            }
1378            ensure_accepted_to_python(
1379                client_order_id,
1380                account_id,
1381                venue_order_id,
1382                identity,
1383                state,
1384                trader_id,
1385                ts_init,
1386                call_soon,
1387                callback,
1388            );
1389            state.insert_triggered(client_order_id);
1390            is_terminal = false;
1391            send_to_python(e, call_soon, callback);
1392        }
1393        ParsedOrderEvent::Canceled(e) => {
1394            ensure_accepted_to_python(
1395                client_order_id,
1396                account_id,
1397                venue_order_id,
1398                identity,
1399                state,
1400                trader_id,
1401                ts_init,
1402                call_soon,
1403                callback,
1404            );
1405            state.remove_triggered(&client_order_id);
1406            state.remove_filled(&client_order_id);
1407            is_terminal = true;
1408            send_to_python(e, call_soon, callback);
1409        }
1410        ParsedOrderEvent::Expired(e) => {
1411            ensure_accepted_to_python(
1412                client_order_id,
1413                account_id,
1414                venue_order_id,
1415                identity,
1416                state,
1417                trader_id,
1418                ts_init,
1419                call_soon,
1420                callback,
1421            );
1422            state.remove_triggered(&client_order_id);
1423            state.remove_filled(&client_order_id);
1424            is_terminal = true;
1425            send_to_python(e, call_soon, callback);
1426        }
1427        ParsedOrderEvent::Rejected(e) => {
1428            state.remove_triggered(&client_order_id);
1429            state.remove_filled(&client_order_id);
1430            is_terminal = true;
1431            send_to_python(e, call_soon, callback);
1432        }
1433    }
1434
1435    if is_terminal {
1436        state.order_identities.remove(&client_order_id);
1437        state.remove_accepted(&client_order_id);
1438    }
1439}
1440
1441/// Synthesizes and sends `OrderAccepted` to Python if one has not yet been emitted.
1442#[expect(clippy::too_many_arguments)]
1443fn ensure_accepted_to_python(
1444    client_order_id: ClientOrderId,
1445    account_id: AccountId,
1446    venue_order_id: VenueOrderId,
1447    identity: &OrderIdentity,
1448    state: &WsDispatchState,
1449    trader_id: TraderId,
1450    ts_init: UnixNanos,
1451    call_soon: &Py<PyAny>,
1452    callback: &Py<PyAny>,
1453) {
1454    if state.accepted_contains(&client_order_id) {
1455        return;
1456    }
1457    state.insert_accepted(client_order_id);
1458    let accepted = OrderAccepted::new(
1459        trader_id,
1460        identity.strategy_id,
1461        identity.instrument_id,
1462        client_order_id,
1463        venue_order_id,
1464        account_id,
1465        UUID4::new(),
1466        ts_init,
1467        ts_init,
1468        false,
1469    );
1470    send_to_python(accepted, call_soon, callback);
1471}
1472
1473fn send_data_to_python(data: Data, call_soon: &Py<PyAny>, callback: &Py<PyAny>) {
1474    Python::attach(|py| {
1475        let py_obj = data_to_pycapsule(py, data);
1476        call_python_threadsafe(py, call_soon, callback, py_obj);
1477    });
1478}
1479
1480fn send_to_python<T: for<'py> IntoPyObjectExt<'py>>(
1481    value: T,
1482    call_soon: &Py<PyAny>,
1483    callback: &Py<PyAny>,
1484) {
1485    Python::attach(|py| {
1486        if let Ok(py_obj) = value.into_py_any(py) {
1487            call_python_threadsafe(py, call_soon, callback, py_obj);
1488        }
1489    });
1490}
nautilus_bitmex/python/websocket.rs

nautilus_bitmex/python/
websocket.rs
