nautilus_databento/python/
historical.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 historical client.
17
18use std::{fmt::Debug, path::PathBuf};
19
20use nautilus_core::{
21    python::{IntoPyObjectNautilusExt, to_pyexception, to_pyvalue_err},
22    time::get_atomic_clock_realtime,
23};
24use nautilus_model::{
25    enums::BarAggregation,
26    identifiers::{InstrumentId, Symbol},
27    python::instruments::instrument_any_to_pyobject,
28};
29use pyo3::{
30    IntoPyObjectExt,
31    prelude::*,
32    types::{PyDict, PyList},
33};
34
35use crate::{
36    common::Credential,
37    historical::{DatabentoHistoricalClient as CoreDatabentoHistoricalClient, RangeQueryParams},
38};
39
40/// Python wrapper for the core Databento historical client.
41#[cfg_attr(
42    feature = "python",
43    pyo3::pyclass(module = "nautilus_trader.adapters.databento")
44)]
45#[cfg_attr(
46    feature = "python",
47    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.databento")
48)]
49pub struct DatabentoHistoricalClient {
50    inner: CoreDatabentoHistoricalClient,
51}
52
53impl Debug for DatabentoHistoricalClient {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct(stringify!(DatabentoHistoricalClient))
56            .field("inner", &self.inner)
57            .finish()
58    }
59}
60
61#[pymethods]
62#[pyo3_stub_gen::derive::gen_stub_pymethods]
63impl DatabentoHistoricalClient {
64    /// Core Databento historical client for fetching historical market data.
65    ///
66    /// This client provides both synchronous and asynchronous interfaces for fetching
67    /// various types of historical market data from Databento.
68    #[new]
69    fn py_new(
70        key: String,
71        publishers_filepath: PathBuf,
72        use_exchange_as_venue: bool,
73    ) -> PyResult<Self> {
74        let clock = get_atomic_clock_realtime();
75        let inner = CoreDatabentoHistoricalClient::new(
76            Credential::new(key),
77            publishers_filepath,
78            clock,
79            use_exchange_as_venue,
80        )
81        .map_err(to_pyvalue_err)?;
82
83        Ok(Self { inner })
84    }
85
86    /// Returns the API key from the stored credential.
87    #[getter]
88    #[pyo3(name = "api_key")]
89    fn py_api_key(&self) -> &str {
90        self.inner.api_key()
91    }
92
93    /// Caches a `price_precision` for the given `symbol`.
94    ///
95    /// When market data is fetched without an explicit `price_precision`, the
96    /// client resolves precision per record from this cache. Instruments
97    /// returned by `Self.get_range_instruments` are inserted automatically.
98    #[pyo3(name = "set_price_precision")]
99    fn py_set_price_precision(&self, symbol: &str, price_precision: u8) {
100        self.inner
101            .set_price_precision(Symbol::from(symbol), price_precision);
102    }
103
104    /// Gets the date range for a specific dataset.
105    ///
106    /// # Errors
107    ///
108    /// Returns an error if the API request fails.
109    #[pyo3(name = "get_dataset_range")]
110    fn py_get_dataset_range<'py>(
111        &self,
112        py: Python<'py>,
113        dataset: String,
114    ) -> PyResult<Bound<'py, PyAny>> {
115        let inner = self.inner.clone();
116
117        pyo3_async_runtimes::tokio::future_into_py(py, async move {
118            let response = inner.get_dataset_range(&dataset).await;
119            match response {
120                Ok(res) => Python::attach(|py| {
121                    let dict = PyDict::new(py);
122                    dict.set_item("start", res.start)?;
123                    dict.set_item("end", res.end)?;
124                    dict.into_py_any(py)
125                }),
126                Err(e) => Err(to_pyexception(format!("Error handling response: {e}"))),
127            }
128        })
129    }
130
131    /// Fetches instrument definitions for the given parameters.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error if the API request or data processing fails.
136    #[pyo3(name = "get_range_instruments")]
137    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
138    #[expect(clippy::needless_pass_by_value)]
139    fn py_get_range_instruments<'py>(
140        &self,
141        py: Python<'py>,
142        dataset: String,
143        instrument_ids: Vec<InstrumentId>,
144        start: u64,
145        end: Option<u64>,
146        limit: Option<u64>,
147    ) -> PyResult<Bound<'py, PyAny>> {
148        let inner = self.inner.clone();
149        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
150
151        let params = RangeQueryParams {
152            dataset,
153            symbols,
154            start: start.into(),
155            end: end.map(Into::into),
156            limit,
157            price_precision: None,
158        };
159
160        pyo3_async_runtimes::tokio::future_into_py(py, async move {
161            let instruments = inner
162                .get_range_instruments(params)
163                .await
164                .map_err(to_pyvalue_err)?;
165
166            Python::attach(|py| -> PyResult<Py<PyAny>> {
167                let objs: Vec<Py<PyAny>> = instruments
168                    .into_iter()
169                    .map(|inst| instrument_any_to_pyobject(py, inst))
170                    .collect::<PyResult<Vec<Py<PyAny>>>>()?;
171
172                let list = PyList::new(py, &objs)?;
173                Ok(list.into_py_any_unwrap(py))
174            })
175        })
176    }
177
178    /// Fetches quote ticks for the given parameters.
179    ///
180    /// # Errors
181    ///
182    /// Returns an error if the API request or data processing fails.
183    #[pyo3(name = "get_range_quotes")]
184    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None, schema=None))]
185    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
186    fn py_get_range_quotes<'py>(
187        &self,
188        py: Python<'py>,
189        dataset: String,
190        instrument_ids: Vec<InstrumentId>,
191        start: u64,
192        end: Option<u64>,
193        limit: Option<u64>,
194        price_precision: Option<u8>,
195        schema: Option<String>,
196    ) -> PyResult<Bound<'py, PyAny>> {
197        let inner = self.inner.clone();
198        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
199
200        let params = RangeQueryParams {
201            dataset,
202            symbols,
203            start: start.into(),
204            end: end.map(Into::into),
205            limit,
206            price_precision,
207        };
208
209        pyo3_async_runtimes::tokio::future_into_py(py, async move {
210            let quotes = inner
211                .get_range_quotes(params, schema)
212                .await
213                .map_err(to_pyvalue_err)?;
214            Python::attach(|py| quotes.into_py_any(py))
215        })
216    }
217
218    /// Fetches trade ticks for the given parameters.
219    ///
220    /// # Errors
221    ///
222    /// Returns an error if the API request or data processing fails.
223    #[pyo3(name = "get_range_trades")]
224    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None, schema=None))]
225    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
226    fn py_get_range_trades<'py>(
227        &self,
228        py: Python<'py>,
229        dataset: String,
230        instrument_ids: Vec<InstrumentId>,
231        start: u64,
232        end: Option<u64>,
233        limit: Option<u64>,
234        price_precision: Option<u8>,
235        schema: Option<String>,
236    ) -> PyResult<Bound<'py, PyAny>> {
237        let inner = self.inner.clone();
238        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
239
240        let params = RangeQueryParams {
241            dataset,
242            symbols,
243            start: start.into(),
244            end: end.map(Into::into),
245            limit,
246            price_precision,
247        };
248
249        pyo3_async_runtimes::tokio::future_into_py(py, async move {
250            let trades = inner
251                .get_range_trades(params, schema)
252                .await
253                .map_err(to_pyvalue_err)?;
254            Python::attach(|py| trades.into_py_any(py))
255        })
256    }
257
258    /// Fetches bars for the given parameters.
259    ///
260    /// # Errors
261    ///
262    /// Returns an error if the API request or data processing fails.
263    #[pyo3(name = "get_range_bars")]
264    #[pyo3(signature = (dataset, instrument_ids, aggregation, start, end=None, limit=None, price_precision=None, timestamp_on_close=true))]
265    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
266    fn py_get_range_bars<'py>(
267        &self,
268        py: Python<'py>,
269        dataset: String,
270        instrument_ids: Vec<InstrumentId>,
271        aggregation: BarAggregation,
272        start: u64,
273        end: Option<u64>,
274        limit: Option<u64>,
275        price_precision: Option<u8>,
276        timestamp_on_close: bool,
277    ) -> PyResult<Bound<'py, PyAny>> {
278        let inner = self.inner.clone();
279        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
280
281        let params = RangeQueryParams {
282            dataset,
283            symbols,
284            start: start.into(),
285            end: end.map(Into::into),
286            limit,
287            price_precision,
288        };
289
290        pyo3_async_runtimes::tokio::future_into_py(py, async move {
291            let bars = inner
292                .get_range_bars(params, aggregation, timestamp_on_close)
293                .await
294                .map_err(to_pyvalue_err)?;
295            Python::attach(|py| bars.into_py_any(py))
296        })
297    }
298
299    #[pyo3(name = "get_order_book_depth10")]
300    #[pyo3(signature = (dataset, instrument_ids, start, end=None, depth=None))]
301    #[expect(clippy::needless_pass_by_value)]
302    fn py_get_order_book_depth10<'py>(
303        &self,
304        py: Python<'py>,
305        dataset: String,
306        instrument_ids: Vec<InstrumentId>,
307        start: u64,
308        end: Option<u64>,
309        depth: Option<usize>,
310    ) -> PyResult<Bound<'py, PyAny>> {
311        let inner = self.inner.clone();
312        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
313
314        let params = RangeQueryParams {
315            dataset,
316            symbols,
317            start: start.into(),
318            end: end.map(Into::into),
319            limit: None,
320            price_precision: None,
321        };
322
323        pyo3_async_runtimes::tokio::future_into_py(py, async move {
324            let depths = inner
325                .get_range_order_book_depth10(params, depth)
326                .await
327                .map_err(to_pyvalue_err)?;
328            Python::attach(|py| depths.into_py_any(py))
329        })
330    }
331
332    /// Fetches order book deltas for the given parameters.
333    ///
334    /// # Errors
335    ///
336    /// Returns an error if the API request or data processing fails.
337    #[pyo3(name = "get_range_order_book_deltas")]
338    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
339    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
340    fn py_get_range_order_book_deltas<'py>(
341        &self,
342        py: Python<'py>,
343        dataset: String,
344        instrument_ids: Vec<InstrumentId>,
345        start: u64,
346        end: Option<u64>,
347        limit: Option<u64>,
348        price_precision: Option<u8>,
349    ) -> PyResult<Bound<'py, PyAny>> {
350        let inner = self.inner.clone();
351        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
352
353        let params = RangeQueryParams {
354            dataset,
355            symbols,
356            start: start.into(),
357            end: end.map(Into::into),
358            limit,
359            price_precision,
360        };
361
362        pyo3_async_runtimes::tokio::future_into_py(py, async move {
363            let deltas = inner
364                .get_range_order_book_deltas(params)
365                .await
366                .map_err(to_pyvalue_err)?;
367            Python::attach(|py| deltas.into_py_any(py))
368        })
369    }
370
371    /// Fetches imbalance data for the given parameters.
372    ///
373    /// # Errors
374    ///
375    /// Returns an error if the API request or data processing fails.
376    #[pyo3(name = "get_range_imbalance")]
377    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
378    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
379    fn py_get_range_imbalance<'py>(
380        &self,
381        py: Python<'py>,
382        dataset: String,
383        instrument_ids: Vec<InstrumentId>,
384        start: u64,
385        end: Option<u64>,
386        limit: Option<u64>,
387        price_precision: Option<u8>,
388    ) -> PyResult<Bound<'py, PyAny>> {
389        let inner = self.inner.clone();
390        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
391
392        let params = RangeQueryParams {
393            dataset,
394            symbols,
395            start: start.into(),
396            end: end.map(Into::into),
397            limit,
398            price_precision,
399        };
400
401        pyo3_async_runtimes::tokio::future_into_py(py, async move {
402            let imbalances = inner
403                .get_range_imbalance(params)
404                .await
405                .map_err(to_pyvalue_err)?;
406            Python::attach(|py| imbalances.into_py_any(py))
407        })
408    }
409
410    /// Fetches statistics data for the given parameters.
411    ///
412    /// # Errors
413    ///
414    /// Returns an error if the API request or data processing fails.
415    #[pyo3(name = "get_range_statistics")]
416    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
417    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
418    fn py_get_range_statistics<'py>(
419        &self,
420        py: Python<'py>,
421        dataset: String,
422        instrument_ids: Vec<InstrumentId>,
423        start: u64,
424        end: Option<u64>,
425        limit: Option<u64>,
426        price_precision: Option<u8>,
427    ) -> PyResult<Bound<'py, PyAny>> {
428        let inner = self.inner.clone();
429        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
430
431        let params = RangeQueryParams {
432            dataset,
433            symbols,
434            start: start.into(),
435            end: end.map(Into::into),
436            limit,
437            price_precision,
438        };
439
440        pyo3_async_runtimes::tokio::future_into_py(py, async move {
441            let statistics = inner
442                .get_range_statistics(params)
443                .await
444                .map_err(to_pyvalue_err)?;
445            Python::attach(|py| statistics.into_py_any(py))
446        })
447    }
448
449    /// Fetches status data for the given parameters.
450    ///
451    /// # Errors
452    ///
453    /// Returns an error if the API request or data processing fails.
454    #[pyo3(name = "get_range_status")]
455    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
456    #[expect(clippy::needless_pass_by_value)]
457    fn py_get_range_status<'py>(
458        &self,
459        py: Python<'py>,
460        dataset: String,
461        instrument_ids: Vec<InstrumentId>,
462        start: u64,
463        end: Option<u64>,
464        limit: Option<u64>,
465    ) -> PyResult<Bound<'py, PyAny>> {
466        let inner = self.inner.clone();
467        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
468
469        let params = RangeQueryParams {
470            dataset,
471            symbols,
472            start: start.into(),
473            end: end.map(Into::into),
474            limit,
475            price_precision: None,
476        };
477
478        pyo3_async_runtimes::tokio::future_into_py(py, async move {
479            let statuses = inner
480                .get_range_status(params)
481                .await
482                .map_err(to_pyvalue_err)?;
483            Python::attach(|py| statuses.into_py_any(py))
484        })
485    }
486}
nautilus_databento/python/historical.rs

nautilus_databento/python/
historical.rs
