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

nautilus_databento/python/
live.rs
