nautilus_bitmex/python/
canceller.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 cancel broadcaster.
17
18use nautilus_core::{python::to_pyvalue_err, string::secret::SecretString};
19use nautilus_model::{
20    enums::OrderSide,
21    identifiers::{ClientOrderId, InstrumentId, VenueOrderId},
22    python::instruments::pyobject_to_instrument_any,
23};
24use pyo3::{conversion::IntoPyObjectExt, prelude::*, types::PyDict};
25
26use crate::{
27    broadcast::canceller::{CancelBroadcaster, CancelBroadcasterConfig},
28    common::enums::BitmexEnvironment,
29};
30
31#[pymethods]
32#[pyo3_stub_gen::derive::gen_stub_pymethods]
33impl CancelBroadcaster {
34    /// Broadcasts cancel requests to multiple HTTP clients for redundancy.
35    ///
36    /// This broadcaster fans out cancel requests to multiple pre-warmed HTTP clients
37    /// in parallel, short-circuits when the first successful acknowledgement is received,
38    /// and handles expected rejection patterns with appropriate log levels.
39    ///
40    /// The client pool must contain `[1, 16]` clients.
41    #[new]
42    #[pyo3(signature = (
43        pool_size,
44        api_key=None,
45        api_secret=None,
46        base_url=None,
47        environment=BitmexEnvironment::Mainnet,
48        timeout_secs=60,
49        max_retries=3,
50        retry_delay_ms=1_000,
51        retry_delay_max_ms=5_000,
52        recv_window_ms=10_000,
53        max_requests_per_second=10,
54        max_requests_per_minute=120,
55        health_check_interval_secs=30,
56        health_check_timeout_secs=5,
57        expected_reject_patterns=None,
58        idempotent_success_patterns=None,
59        proxy_urls=None
60    ))]
61    #[expect(clippy::too_many_arguments)]
62    fn py_new(
63        pool_size: usize,
64        api_key: Option<String>,
65        api_secret: Option<String>,
66        base_url: Option<String>,
67        environment: BitmexEnvironment,
68        timeout_secs: u64,
69        max_retries: u32,
70        retry_delay_ms: u64,
71        retry_delay_max_ms: u64,
72        recv_window_ms: u64,
73        max_requests_per_second: u32,
74        max_requests_per_minute: u32,
75        health_check_interval_secs: u64,
76        health_check_timeout_secs: u64,
77        expected_reject_patterns: Option<Vec<String>>,
78        idempotent_success_patterns: Option<Vec<String>>,
79        proxy_urls: Option<Vec<Option<String>>>,
80    ) -> PyResult<Self> {
81        let config = CancelBroadcasterConfig {
82            pool_size,
83            api_key: api_key.map(SecretString::from),
84            api_secret: api_secret.map(SecretString::from),
85            base_url,
86            environment,
87            timeout_secs,
88            max_retries,
89            retry_delay_ms,
90            retry_delay_max_ms,
91            recv_window_ms,
92            max_requests_per_second,
93            max_requests_per_minute,
94            health_check_interval_secs,
95            health_check_timeout_secs,
96            expected_reject_patterns: expected_reject_patterns
97                .unwrap_or_else(|| CancelBroadcasterConfig::default().expected_reject_patterns),
98            idempotent_success_patterns: idempotent_success_patterns
99                .unwrap_or_else(|| CancelBroadcasterConfig::default().idempotent_success_patterns),
100            proxy_urls: proxy_urls
101                .unwrap_or_default()
102                .into_iter()
103                .map(|value| value.map(SecretString::from))
104                .collect(),
105        };
106
107        Self::new(config).map_err(to_pyvalue_err)
108    }
109
110    /// Caches an instrument in all HTTP clients in the pool.
111    #[pyo3(name = "cache_instrument")]
112    fn py_cache_instrument(&self, py: Python, instrument: Py<PyAny>) -> PyResult<()> {
113        let inst_any = pyobject_to_instrument_any(py, instrument)?;
114        self.cache_instrument(&inst_any);
115        Ok(())
116    }
117
118    /// Starts the broadcaster and health check loop.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if the broadcaster is already running.
123    #[pyo3(name = "start")]
124    fn py_start<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
125        let broadcaster = self.clone_for_async();
126        pyo3_async_runtimes::tokio::future_into_py(py, async move {
127            broadcaster.start().await.map_err(to_pyvalue_err)
128        })
129    }
130
131    /// Stops the broadcaster and health check loop.
132    #[pyo3(name = "stop")]
133    fn py_stop<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
134        let broadcaster = self.clone_for_async();
135        pyo3_async_runtimes::tokio::future_into_py(py, async move {
136            broadcaster.stop().await;
137            Ok(())
138        })
139    }
140
141    /// Broadcasts a single cancel request to all healthy clients in parallel.
142    ///
143    /// # Returns
144    ///
145    /// - `Ok(Some(report))` if successfully cancelled with a report.
146    /// - `Ok(None)` if the order was already cancelled (idempotent success).
147    /// - `Err` if all requests failed.
148    ///
149    /// # Errors
150    ///
151    /// Returns an error if all cancel requests fail or no healthy clients are available.
152    #[pyo3(name = "broadcast_cancel")]
153    fn py_broadcast_cancel<'py>(
154        &self,
155        py: Python<'py>,
156        instrument_id: InstrumentId,
157        client_order_id: Option<ClientOrderId>,
158        venue_order_id: Option<VenueOrderId>,
159    ) -> PyResult<Bound<'py, PyAny>> {
160        let broadcaster = self.clone_for_async();
161        pyo3_async_runtimes::tokio::future_into_py(py, async move {
162            let report = broadcaster
163                .broadcast_cancel(instrument_id, client_order_id, venue_order_id)
164                .await
165                .map_err(to_pyvalue_err)?;
166
167            Python::attach(|py| match report {
168                Some(r) => r.into_py_any(py),
169                None => Ok(py.None()),
170            })
171        })
172    }
173
174    /// Broadcasts a batch cancel request to all healthy clients in parallel.
175    ///
176    /// # Errors
177    ///
178    /// Returns an error if all cancel requests fail or no healthy clients are available.
179    #[pyo3(name = "broadcast_batch_cancel")]
180    fn py_broadcast_batch_cancel<'py>(
181        &self,
182        py: Python<'py>,
183        instrument_id: InstrumentId,
184        client_order_ids: Option<Vec<ClientOrderId>>,
185        venue_order_ids: Option<Vec<VenueOrderId>>,
186    ) -> PyResult<Bound<'py, PyAny>> {
187        let broadcaster = self.clone_for_async();
188        pyo3_async_runtimes::tokio::future_into_py(py, async move {
189            let reports = broadcaster
190                .broadcast_batch_cancel(instrument_id, client_order_ids, venue_order_ids)
191                .await
192                .map_err(to_pyvalue_err)?;
193
194            Python::attach(|py| {
195                let py_reports: PyResult<Vec<_>> = reports
196                    .into_iter()
197                    .map(|report| report.into_py_any(py))
198                    .collect();
199                let pylist = pyo3::types::PyList::new(py, py_reports?)?
200                    .into_any()
201                    .unbind();
202                Ok(pylist)
203            })
204        })
205    }
206
207    /// Broadcasts a cancel all request to all healthy clients in parallel.
208    ///
209    /// # Errors
210    ///
211    /// Returns an error if all cancel requests fail or no healthy clients are available.
212    #[pyo3(name = "broadcast_cancel_all")]
213    fn py_broadcast_cancel_all<'py>(
214        &self,
215        py: Python<'py>,
216        instrument_id: InstrumentId,
217        order_side: Option<OrderSide>,
218    ) -> PyResult<Bound<'py, PyAny>> {
219        let broadcaster = self.clone_for_async();
220        pyo3_async_runtimes::tokio::future_into_py(py, async move {
221            let reports = broadcaster
222                .broadcast_cancel_all(instrument_id, order_side)
223                .await
224                .map_err(to_pyvalue_err)?;
225
226            Python::attach(|py| {
227                let py_reports: PyResult<Vec<_>> = reports
228                    .into_iter()
229                    .map(|report| report.into_py_any(py))
230                    .collect();
231                let pylist = pyo3::types::PyList::new(py, py_reports?)?
232                    .into_any()
233                    .unbind();
234                Ok(pylist)
235            })
236        })
237    }
238
239    /// Gets broadcaster metrics.
240    #[pyo3(name = "get_metrics")]
241    fn py_get_metrics(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
242        let metrics = self.get_metrics();
243        let dict = PyDict::new(py);
244        dict.set_item("total_cancels", metrics.total_cancels)?;
245        dict.set_item("successful_cancels", metrics.successful_cancels)?;
246        dict.set_item("failed_cancels", metrics.failed_cancels)?;
247        dict.set_item("expected_rejects", metrics.expected_rejects)?;
248        dict.set_item("idempotent_successes", metrics.idempotent_successes)?;
249        dict.set_item("healthy_clients", metrics.healthy_clients)?;
250        dict.set_item("total_clients", metrics.total_clients)?;
251        Ok(dict.into())
252    }
253
254    /// Gets per-client statistics.
255    #[pyo3(name = "get_client_stats")]
256    fn py_get_client_stats(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
257        let stats = self.get_client_stats();
258        let list = pyo3::types::PyList::empty(py);
259        for stat in stats {
260            let dict = PyDict::new(py);
261            dict.set_item("client_id", stat.client_id.clone())?;
262            dict.set_item("healthy", stat.healthy)?;
263            dict.set_item("cancel_count", stat.cancel_count)?;
264            dict.set_item("error_count", stat.error_count)?;
265            list.append(dict)?;
266        }
267        Ok(list.into())
268    }
269}
nautilus_bitmex/python/canceller.rs

nautilus_bitmex/python/
canceller.rs
