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.core.nautilus_pyo3.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    #[pyo3(name = "get_dataset_range")]
106    fn py_get_dataset_range<'py>(
107        &self,
108        py: Python<'py>,
109        dataset: String,
110    ) -> PyResult<Bound<'py, PyAny>> {
111        let inner = self.inner.clone();
112
113        pyo3_async_runtimes::tokio::future_into_py(py, async move {
114            let response = inner.get_dataset_range(&dataset).await;
115            match response {
116                Ok(res) => Python::attach(|py| {
117                    let dict = PyDict::new(py);
118                    dict.set_item("start", res.start)?;
119                    dict.set_item("end", res.end)?;
120                    dict.into_py_any(py)
121                }),
122                Err(e) => Err(to_pyexception(format!("Error handling response: {e}"))),
123            }
124        })
125    }
126
127    /// Fetches instrument definitions for the given parameters.
128    #[pyo3(name = "get_range_instruments")]
129    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
130    #[expect(clippy::needless_pass_by_value)]
131    fn py_get_range_instruments<'py>(
132        &self,
133        py: Python<'py>,
134        dataset: String,
135        instrument_ids: Vec<InstrumentId>,
136        start: u64,
137        end: Option<u64>,
138        limit: Option<u64>,
139    ) -> PyResult<Bound<'py, PyAny>> {
140        let inner = self.inner.clone();
141        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
142
143        let params = RangeQueryParams {
144            dataset,
145            symbols,
146            start: start.into(),
147            end: end.map(Into::into),
148            limit,
149            price_precision: None,
150        };
151
152        pyo3_async_runtimes::tokio::future_into_py(py, async move {
153            let instruments = inner
154                .get_range_instruments(params)
155                .await
156                .map_err(to_pyvalue_err)?;
157
158            Python::attach(|py| -> PyResult<Py<PyAny>> {
159                let objs: Vec<Py<PyAny>> = instruments
160                    .into_iter()
161                    .map(|inst| instrument_any_to_pyobject(py, inst))
162                    .collect::<PyResult<Vec<Py<PyAny>>>>()?;
163
164                let list = PyList::new(py, &objs).expect("Invalid `ExactSizeIterator`");
165                Ok(list.into_py_any_unwrap(py))
166            })
167        })
168    }
169
170    /// Fetches quote ticks for the given parameters.
171    #[pyo3(name = "get_range_quotes")]
172    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None, schema=None))]
173    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
174    fn py_get_range_quotes<'py>(
175        &self,
176        py: Python<'py>,
177        dataset: String,
178        instrument_ids: Vec<InstrumentId>,
179        start: u64,
180        end: Option<u64>,
181        limit: Option<u64>,
182        price_precision: Option<u8>,
183        schema: Option<String>,
184    ) -> PyResult<Bound<'py, PyAny>> {
185        let inner = self.inner.clone();
186        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
187
188        let params = RangeQueryParams {
189            dataset,
190            symbols,
191            start: start.into(),
192            end: end.map(Into::into),
193            limit,
194            price_precision,
195        };
196
197        pyo3_async_runtimes::tokio::future_into_py(py, async move {
198            let quotes = inner
199                .get_range_quotes(params, schema)
200                .await
201                .map_err(to_pyvalue_err)?;
202            Python::attach(|py| quotes.into_py_any(py))
203        })
204    }
205
206    /// Fetches trade ticks for the given parameters.
207    #[pyo3(name = "get_range_trades")]
208    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None, schema=None))]
209    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
210    fn py_get_range_trades<'py>(
211        &self,
212        py: Python<'py>,
213        dataset: String,
214        instrument_ids: Vec<InstrumentId>,
215        start: u64,
216        end: Option<u64>,
217        limit: Option<u64>,
218        price_precision: Option<u8>,
219        schema: Option<String>,
220    ) -> PyResult<Bound<'py, PyAny>> {
221        let inner = self.inner.clone();
222        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
223
224        let params = RangeQueryParams {
225            dataset,
226            symbols,
227            start: start.into(),
228            end: end.map(Into::into),
229            limit,
230            price_precision,
231        };
232
233        pyo3_async_runtimes::tokio::future_into_py(py, async move {
234            let trades = inner
235                .get_range_trades(params, schema)
236                .await
237                .map_err(to_pyvalue_err)?;
238            Python::attach(|py| trades.into_py_any(py))
239        })
240    }
241
242    /// Fetches bars for the given parameters.
243    #[pyo3(name = "get_range_bars")]
244    #[pyo3(signature = (dataset, instrument_ids, aggregation, start, end=None, limit=None, price_precision=None, timestamp_on_close=true))]
245    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
246    fn py_get_range_bars<'py>(
247        &self,
248        py: Python<'py>,
249        dataset: String,
250        instrument_ids: Vec<InstrumentId>,
251        aggregation: BarAggregation,
252        start: u64,
253        end: Option<u64>,
254        limit: Option<u64>,
255        price_precision: Option<u8>,
256        timestamp_on_close: bool,
257    ) -> PyResult<Bound<'py, PyAny>> {
258        let inner = self.inner.clone();
259        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
260
261        let params = RangeQueryParams {
262            dataset,
263            symbols,
264            start: start.into(),
265            end: end.map(Into::into),
266            limit,
267            price_precision,
268        };
269
270        pyo3_async_runtimes::tokio::future_into_py(py, async move {
271            let bars = inner
272                .get_range_bars(params, aggregation, timestamp_on_close)
273                .await
274                .map_err(to_pyvalue_err)?;
275            Python::attach(|py| bars.into_py_any(py))
276        })
277    }
278
279    #[pyo3(name = "get_order_book_depth10")]
280    #[pyo3(signature = (dataset, instrument_ids, start, end=None, depth=None))]
281    #[expect(clippy::needless_pass_by_value)]
282    fn py_get_order_book_depth10<'py>(
283        &self,
284        py: Python<'py>,
285        dataset: String,
286        instrument_ids: Vec<InstrumentId>,
287        start: u64,
288        end: Option<u64>,
289        depth: Option<usize>,
290    ) -> PyResult<Bound<'py, PyAny>> {
291        let inner = self.inner.clone();
292        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
293
294        let params = RangeQueryParams {
295            dataset,
296            symbols,
297            start: start.into(),
298            end: end.map(Into::into),
299            limit: None,
300            price_precision: None,
301        };
302
303        pyo3_async_runtimes::tokio::future_into_py(py, async move {
304            let depths = inner
305                .get_range_order_book_depth10(params, depth)
306                .await
307                .map_err(to_pyvalue_err)?;
308            Python::attach(|py| depths.into_py_any(py))
309        })
310    }
311
312    /// Fetches order book deltas for the given parameters.
313    #[pyo3(name = "get_range_order_book_deltas")]
314    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
315    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
316    fn py_get_range_order_book_deltas<'py>(
317        &self,
318        py: Python<'py>,
319        dataset: String,
320        instrument_ids: Vec<InstrumentId>,
321        start: u64,
322        end: Option<u64>,
323        limit: Option<u64>,
324        price_precision: Option<u8>,
325    ) -> PyResult<Bound<'py, PyAny>> {
326        let inner = self.inner.clone();
327        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
328
329        let params = RangeQueryParams {
330            dataset,
331            symbols,
332            start: start.into(),
333            end: end.map(Into::into),
334            limit,
335            price_precision,
336        };
337
338        pyo3_async_runtimes::tokio::future_into_py(py, async move {
339            let deltas = inner
340                .get_range_order_book_deltas(params)
341                .await
342                .map_err(to_pyvalue_err)?;
343            Python::attach(|py| deltas.into_py_any(py))
344        })
345    }
346
347    /// Fetches imbalance data for the given parameters.
348    #[pyo3(name = "get_range_imbalance")]
349    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
350    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
351    fn py_get_range_imbalance<'py>(
352        &self,
353        py: Python<'py>,
354        dataset: String,
355        instrument_ids: Vec<InstrumentId>,
356        start: u64,
357        end: Option<u64>,
358        limit: Option<u64>,
359        price_precision: Option<u8>,
360    ) -> PyResult<Bound<'py, PyAny>> {
361        let inner = self.inner.clone();
362        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
363
364        let params = RangeQueryParams {
365            dataset,
366            symbols,
367            start: start.into(),
368            end: end.map(Into::into),
369            limit,
370            price_precision,
371        };
372
373        pyo3_async_runtimes::tokio::future_into_py(py, async move {
374            let imbalances = inner
375                .get_range_imbalance(params)
376                .await
377                .map_err(to_pyvalue_err)?;
378            Python::attach(|py| imbalances.into_py_any(py))
379        })
380    }
381
382    /// Fetches statistics data for the given parameters.
383    #[pyo3(name = "get_range_statistics")]
384    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, price_precision=None))]
385    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
386    fn py_get_range_statistics<'py>(
387        &self,
388        py: Python<'py>,
389        dataset: String,
390        instrument_ids: Vec<InstrumentId>,
391        start: u64,
392        end: Option<u64>,
393        limit: Option<u64>,
394        price_precision: Option<u8>,
395    ) -> PyResult<Bound<'py, PyAny>> {
396        let inner = self.inner.clone();
397        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
398
399        let params = RangeQueryParams {
400            dataset,
401            symbols,
402            start: start.into(),
403            end: end.map(Into::into),
404            limit,
405            price_precision,
406        };
407
408        pyo3_async_runtimes::tokio::future_into_py(py, async move {
409            let statistics = inner
410                .get_range_statistics(params)
411                .await
412                .map_err(to_pyvalue_err)?;
413            Python::attach(|py| statistics.into_py_any(py))
414        })
415    }
416
417    /// Fetches status data for the given parameters.
418    #[pyo3(name = "get_range_status")]
419    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
420    #[expect(clippy::needless_pass_by_value)]
421    fn py_get_range_status<'py>(
422        &self,
423        py: Python<'py>,
424        dataset: String,
425        instrument_ids: Vec<InstrumentId>,
426        start: u64,
427        end: Option<u64>,
428        limit: Option<u64>,
429    ) -> PyResult<Bound<'py, PyAny>> {
430        let inner = self.inner.clone();
431        let symbols = inner.prepare_symbols_from_instrument_ids(&instrument_ids);
432
433        let params = RangeQueryParams {
434            dataset,
435            symbols,
436            start: start.into(),
437            end: end.map(Into::into),
438            limit,
439            price_precision: None,
440        };
441
442        pyo3_async_runtimes::tokio::future_into_py(py, async move {
443            let statuses = inner
444                .get_range_status(params)
445                .await
446                .map_err(to_pyvalue_err)?;
447            Python::attach(|py| statuses.into_py_any(py))
448        })
449    }
450}
nautilus_databento/python/historical.rs

nautilus_databento/python/
historical.rs
