nautilus_databento/python/
live.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 Databento live client.
17
18use std::path::PathBuf;
19
20use nautilus_core::python::{to_pyruntime_err, to_pyvalue_err};
21use nautilus_model::{
22    identifiers::InstrumentId,
23    python::{data::data_to_pyobject, instruments::instrument_any_to_pyobject},
24};
25use pyo3::{IntoPyObjectExt, prelude::*};
26
27use super::types::DatabentoSubscriptionAck;
28pub use crate::live::DatabentoLiveClient;
29use crate::live::{DatabentoMessage, is_command_send_error};
30
31impl DatabentoLiveClient {
32    async fn process_messages(
33        mut msg_rx: tokio::sync::mpsc::UnboundedReceiver<DatabentoMessage>,
34        callback: Py<PyAny>,
35        callback_pyo3: Py<PyAny>,
36    ) -> PyResult<()> {
37        log::debug!("Processing messages...");
38        // Continue to process messages until channel is hung up
39        while let Some(msg) = msg_rx.recv().await {
40            log::trace!("Received message: {msg:?}");
41
42            match msg {
43                DatabentoMessage::Data(data) => Python::attach(|py| -> PyResult<()> {
44                    let py_obj = data_to_pyobject(py, data)?;
45                    call_python(py, &callback, py_obj);
46                    Ok(())
47                })?,
48                DatabentoMessage::Instrument(data) => {
49                    Python::attach(|py| match instrument_any_to_pyobject(py, *data) {
50                        Ok(py_obj) => call_python(py, &callback, py_obj),
51                        Err(e) => log::error!("Failed creating instrument: {e}"),
52                    });
53                }
54                DatabentoMessage::Status(data) => {
55                    Python::attach(|py| -> PyResult<()> {
56                        call_python(py, &callback_pyo3, data.into_py_any(py)?);
57                        Ok(())
58                    })?;
59                }
60                DatabentoMessage::Imbalance(data) => {
61                    Python::attach(|py| -> PyResult<()> {
62                        call_python(py, &callback_pyo3, data.into_py_any(py)?);
63                        Ok(())
64                    })?;
65                }
66                DatabentoMessage::Statistics(data) => {
67                    Python::attach(|py| -> PyResult<()> {
68                        call_python(py, &callback_pyo3, data.into_py_any(py)?);
69                        Ok(())
70                    })?;
71                }
72                DatabentoMessage::SubscriptionAck(ack) => {
73                    Python::attach(|py| -> PyResult<()> {
74                        let py_obj = DatabentoSubscriptionAck::from(ack).into_py_any(py)?;
75                        call_python(py, &callback_pyo3, py_obj);
76                        Ok(())
77                    })?;
78                }
79                DatabentoMessage::Close => {
80                    // Graceful close
81                    break;
82                }
83                DatabentoMessage::Error(e) => {
84                    // Return error to Python
85                    return Err(to_pyruntime_err(e));
86                }
87            }
88        }
89
90        msg_rx.close();
91        log::debug!("Closed message receiver");
92
93        Ok(())
94    }
95}
96
97fn call_python(py: Python, callback: &Py<PyAny>, py_obj: Py<PyAny>) {
98    if let Err(e) = callback.call1(py, (py_obj,)) {
99        // TODO: Improve this by checking for the actual exception type
100        if !e.to_string().contains("CancelledError") {
101            log::error!("Error calling Python: {e}");
102        }
103    }
104}
105
106#[pymethods]
107#[pyo3_stub_gen::derive::gen_stub_pymethods]
108impl DatabentoLiveClient {
109    /// Creates a new `DatabentoLiveClient` instance.
110    ///
111    /// # Errors
112    ///
113    /// Returns an error if reading or parsing the publishers file fails.
114    #[new]
115    #[pyo3(signature = (key, dataset, publishers_filepath, use_exchange_as_venue, bars_timestamp_on_close=None, reconnect_timeout_mins=None))]
116    pub fn py_new(
117        key: String,
118        dataset: String,
119        publishers_filepath: PathBuf,
120        use_exchange_as_venue: bool,
121        bars_timestamp_on_close: Option<bool>,
122        reconnect_timeout_mins: Option<i64>,
123    ) -> PyResult<Self> {
124        Self::new(
125            key,
126            dataset,
127            publishers_filepath,
128            use_exchange_as_venue,
129            bars_timestamp_on_close,
130            reconnect_timeout_mins,
131        )
132        .map_err(to_pyvalue_err)
133    }
134
135    #[getter]
136    fn dataset(&self) -> &str {
137        self.dataset.as_str()
138    }
139
140    #[pyo3(name = "is_running")]
141    const fn py_is_running(&self) -> bool {
142        self.is_running()
143    }
144
145    #[pyo3(name = "is_closed")]
146    const fn py_is_closed(&self) -> bool {
147        self.is_closed()
148    }
149
150    /// Subscribes to Databento live data for the requested instruments.
151    ///
152    /// # Errors
153    ///
154    /// Returns an error if symbology, schema, timestamp, or precision inputs are invalid,
155    /// or if the command cannot be sent to the feed handler.
156    #[pyo3(name = "subscribe")]
157    #[pyo3(signature = (schema, instrument_ids, start=None, snapshot=None, price_precisions=None, stype_in=None))]
158    fn py_subscribe(
159        &mut self,
160        schema: String,
161        instrument_ids: Vec<InstrumentId>,
162        start: Option<u64>,
163        snapshot: Option<bool>,
164        price_precisions: Option<Vec<Option<u8>>>,
165        stype_in: Option<String>,
166    ) -> PyResult<()> {
167        if let Err(e) = self.subscribe(
168            schema,
169            instrument_ids,
170            start,
171            snapshot,
172            price_precisions,
173            stype_in,
174        ) {
175            return if is_command_send_error(&e) {
176                Err(to_pyruntime_err(e))
177            } else {
178                Err(to_pyvalue_err(e))
179            };
180        }
181
182        Ok(())
183    }
184
185    /// Starts the live feed handler and returns its message receiver.
186    ///
187    /// # Errors
188    ///
189    /// Returns an error if the client is already closed, already running, or cannot start.
190    #[pyo3(name = "start")]
191    fn py_start<'py>(
192        &mut self,
193        py: Python<'py>,
194        callback: Py<PyAny>,
195        callback_pyo3: Py<PyAny>,
196    ) -> PyResult<Bound<'py, PyAny>> {
197        let (mut feed_handler, msg_rx) = self.start().map_err(to_pyruntime_err)?;
198
199        pyo3_async_runtimes::tokio::future_into_py(py, async move {
200            let (proc_handle, feed_handle) = tokio::join!(
201                Self::process_messages(msg_rx, callback, callback_pyo3),
202                feed_handler.run(),
203            );
204
205            if let Err(e) = proc_handle {
206                log::error!("Message processor error: {e}");
207                return Err(e);
208            }
209
210            if let Err(e) = feed_handle {
211                log::error!("Feed handler error: {e}");
212                return Err(to_pyruntime_err(e));
213            }
214
215            log::debug!("Live client completed");
216            Ok(())
217        })
218    }
219
220    /// Closes the live client.
221    ///
222    /// # Errors
223    ///
224    /// Returns an error if the client was never started, is already closed, or cannot send
225    /// the close command to the feed handler.
226    #[pyo3(name = "close")]
227    fn py_close(&mut self) -> PyResult<()> {
228        self.close().map_err(to_pyruntime_err)
229    }
230}
231
232#[cfg(test)]
233mod tests {
234    use std::path::PathBuf;
235
236    use pyo3::exceptions::{PyRuntimeError, PyValueError};
237    use rstest::rstest;
238
239    use super::*;
240
241    fn create_test_client() -> DatabentoLiveClient {
242        DatabentoLiveClient::new(
243            "test-api-key".to_string(),
244            "GLBX.MDP3".to_string(),
245            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("publishers.json"),
246            true,
247            None,
248            None,
249        )
250        .unwrap()
251    }
252
253    #[rstest]
254    fn test_py_subscribe_maps_invalid_input_to_value_error() {
255        Python::initialize();
256        let mut client = create_test_client();
257
258        let err = client
259            .py_subscribe(
260                "definition".to_string(),
261                vec![InstrumentId::from("ES.FUT.GLBX")],
262                None,
263                None,
264                None,
265                Some("not-a-stype".to_string()),
266            )
267            .unwrap_err();
268
269        Python::attach(|py| {
270            assert!(err.is_instance_of::<PyValueError>(py));
271        });
272    }
273
274    #[rstest]
275    fn test_py_subscribe_maps_command_send_error_to_runtime_error() {
276        Python::initialize();
277        let mut client = create_test_client();
278        let (feed_handler, msg_rx) = client.start().unwrap();
279        drop(feed_handler);
280        drop(msg_rx);
281
282        let err = client
283            .py_subscribe(
284                "definition".to_string(),
285                vec![InstrumentId::from("ES.FUT.GLBX")],
286                None,
287                None,
288                None,
289                Some("parent".to_string()),
290            )
291            .unwrap_err();
292
293        Python::attach(|py| {
294            assert!(err.is_instance_of::<PyRuntimeError>(py));
295        });
296    }
297}
nautilus_databento/python/live.rs

nautilus_databento/python/
live.rs
