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

nautilus_bitmex/python/
submitter.rs
