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, string::secret::SecretString};
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    ///
41    /// The client pool must contain `[1, 16]` clients.
42    #[new]
43    #[pyo3(signature = (
44        pool_size,
45        api_key=None,
46        api_secret=None,
47        base_url=None,
48        environment=BitmexEnvironment::Mainnet,
49        timeout_secs=60,
50        max_retries=3,
51        retry_delay_ms=1_000,
52        retry_delay_max_ms=5_000,
53        recv_window_ms=10_000,
54        max_requests_per_second=10,
55        max_requests_per_minute=120,
56        health_check_interval_secs=30,
57        health_check_timeout_secs=5,
58        expected_reject_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        proxy_urls: Option<Vec<Option<String>>>,
79    ) -> PyResult<Self> {
80        let config = SubmitBroadcasterConfig {
81            pool_size,
82            api_key: api_key.map(SecretString::from),
83            api_secret: api_secret.map(SecretString::from),
84            base_url,
85            environment,
86            timeout_secs,
87            max_retries,
88            retry_delay_ms,
89            retry_delay_max_ms,
90            recv_window_ms,
91            max_requests_per_second,
92            max_requests_per_minute,
93            health_check_interval_secs,
94            health_check_timeout_secs,
95            expected_reject_patterns: expected_reject_patterns
96                .unwrap_or_else(|| SubmitBroadcasterConfig::default().expected_reject_patterns),
97            proxy_urls: proxy_urls
98                .unwrap_or_default()
99                .into_iter()
100                .map(|value| value.map(SecretString::from))
101                .collect(),
102        };
103
104        Self::new(config).map_err(to_pyvalue_err)
105    }
106
107    /// Starts the broadcaster and health check loop.
108    ///
109    /// # Errors
110    ///
111    /// Returns an error if the broadcaster is already running.
112    #[pyo3(name = "start")]
113    fn py_start<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
114        let broadcaster = self.clone_for_async();
115        pyo3_async_runtimes::tokio::future_into_py(py, async move {
116            broadcaster.start().await.map_err(to_pyvalue_err)
117        })
118    }
119
120    /// Stops the broadcaster and health check loop.
121    #[pyo3(name = "stop")]
122    fn py_stop<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
123        let broadcaster = self.clone_for_async();
124        pyo3_async_runtimes::tokio::future_into_py(py, async move {
125            broadcaster.stop().await;
126            Ok(())
127        })
128    }
129
130    /// Broadcasts a submit request to all healthy clients in parallel.
131    ///
132    /// # Returns
133    ///
134    /// - `Ok(report)` if successfully submitted with a report.
135    /// - `Err` if all requests failed.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if all submit requests fail or no healthy clients are available.
140    #[pyo3(name = "broadcast_submit")]
141    #[pyo3(signature = (
142        instrument_id,
143        client_order_id,
144        order_side,
145        order_type,
146        quantity,
147        time_in_force,
148        price=None,
149        trigger_price=None,
150        trigger_type=None,
151        trailing_offset=None,
152        trailing_offset_type=None,
153        display_qty=None,
154        post_only=false,
155        reduce_only=false,
156        order_list_id=None,
157        contingency_type=None,
158        submit_tries=None,
159        peg_price_type=None,
160        peg_offset_value=None
161    ))]
162    #[expect(clippy::too_many_arguments)]
163    fn py_broadcast_submit<'py>(
164        &self,
165        py: Python<'py>,
166        instrument_id: InstrumentId,
167        client_order_id: ClientOrderId,
168        order_side: OrderSide,
169        order_type: OrderType,
170        quantity: Quantity,
171        time_in_force: TimeInForce,
172        price: Option<Price>,
173        trigger_price: Option<Price>,
174        trigger_type: Option<TriggerType>,
175        trailing_offset: Option<f64>,
176        trailing_offset_type: Option<TrailingOffsetType>,
177        display_qty: Option<Quantity>,
178        post_only: bool,
179        reduce_only: bool,
180        order_list_id: Option<OrderListId>,
181        contingency_type: Option<ContingencyType>,
182        submit_tries: Option<usize>,
183        peg_price_type: Option<String>,
184        peg_offset_value: Option<f64>,
185    ) -> PyResult<Bound<'py, PyAny>> {
186        let broadcaster = self.clone_for_async();
187
188        let peg_price_type: Option<BitmexPegPriceType> = peg_price_type
189            .map(|s| {
190                s.parse::<BitmexPegPriceType>()
191                    .map_err(|_| to_pyvalue_err(format!("Invalid peg_price_type: {s}")))
192            })
193            .transpose()?;
194
195        pyo3_async_runtimes::tokio::future_into_py(py, async move {
196            let report = broadcaster
197                .broadcast_submit(
198                    instrument_id,
199                    client_order_id,
200                    order_side,
201                    order_type,
202                    quantity,
203                    time_in_force,
204                    price,
205                    trigger_price,
206                    trigger_type,
207                    trailing_offset,
208                    trailing_offset_type,
209                    display_qty,
210                    post_only,
211                    reduce_only,
212                    order_list_id,
213                    contingency_type,
214                    submit_tries,
215                    peg_price_type,
216                    peg_offset_value,
217                )
218                .await
219                .map_err(to_pyvalue_err)?;
220
221            Python::attach(|py| report.into_py_any(py))
222        })
223    }
224
225    /// Gets broadcaster metrics.
226    #[pyo3(name = "get_metrics")]
227    fn py_get_metrics(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
228        let metrics = self.get_metrics();
229        let dict = PyDict::new(py);
230        dict.set_item("total_submits", metrics.total_submits)?;
231        dict.set_item("successful_submits", metrics.successful_submits)?;
232        dict.set_item("failed_submits", metrics.failed_submits)?;
233        dict.set_item("expected_rejects", metrics.expected_rejects)?;
234        dict.set_item("healthy_clients", metrics.healthy_clients)?;
235        dict.set_item("total_clients", metrics.total_clients)?;
236        Ok(dict.into())
237    }
238
239    /// Gets per-client statistics.
240    #[pyo3(name = "get_client_stats")]
241    fn py_get_client_stats(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
242        let stats = self.get_client_stats();
243        let list = pyo3::types::PyList::empty(py);
244        for stat in stats {
245            let dict = PyDict::new(py);
246            dict.set_item("client_id", stat.client_id.clone())?;
247            dict.set_item("healthy", stat.healthy)?;
248            dict.set_item("submit_count", stat.submit_count)?;
249            dict.set_item("error_count", stat.error_count)?;
250            list.append(dict)?;
251        }
252        Ok(list.into())
253    }
254
255    /// Caches an instrument in all HTTP clients in the pool.
256    #[pyo3(name = "cache_instrument")]
257    fn py_cache_instrument(&self, py: Python, instrument: Py<PyAny>) -> PyResult<()> {
258        let inst_any = pyobject_to_instrument_any(py, instrument)?;
259        self.cache_instrument(&inst_any);
260        Ok(())
261    }
262}
nautilus_bitmex/python/submitter.rs

nautilus_bitmex/python/
submitter.rs
