nautilus_persistence/python/
config.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 persistence configuration types.
17
18use nautilus_core::{
19    DurationNanos, UnixNanos,
20    datetime::NANOSECONDS_IN_DAY,
21    from_pydict,
22    python::{params::params_to_pydict, to_pytype_err, to_pyvalue_err},
23};
24use nautilus_model::{
25    data::{NautilusDataType, NautilusRecordType},
26    instruments::NautilusInstrumentType,
27    python::{
28        data::{PyNautilusDataType, PyNautilusRecordType},
29        instruments::PyNautilusInstrumentType,
30    },
31};
32use pyo3::{
33    Bound, Py, PyAny, PyRef, PyResult, Python,
34    types::{PyAnyMethods, PyDict},
35};
36
37use crate::{
38    config::{
39        CatalogBackendType, DataCatalogConfig, RotationConfig, StreamingConfig,
40        StreamingRecordFilterConfig, default_fs_protocol,
41    },
42    writer::factory::WriterBackendType,
43};
44
45#[derive(Clone, Debug, Eq, Hash, PartialEq)]
46#[pyo3::pyclass(
47    frozen,
48    name = "CatalogBackend",
49    module = "nautilus_trader.persistence",
50    skip_from_py_object
51)]
52#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
53pub struct PyCatalogBackend {
54    inner: CatalogBackendType,
55}
56
57impl PyCatalogBackend {
58    #[must_use]
59    pub const fn new(inner: CatalogBackendType) -> Self {
60        Self { inner }
61    }
62
63    #[must_use]
64    pub fn inner(&self) -> CatalogBackendType {
65        self.inner.clone()
66    }
67}
68
69fn parse_catalog_backend(value: &str) -> PyResult<CatalogBackendType> {
70    value.parse::<CatalogBackendType>().map_err(to_pyvalue_err)
71}
72
73#[pyo3_stub_gen::derive::gen_stub_pymethods]
74#[pyo3::pymethods]
75impl PyCatalogBackend {
76    #[classattr]
77    #[expect(
78        non_snake_case,
79        clippy::use_self,
80        reason = "PyO3 stub generation needs the concrete Python enum type"
81    )]
82    fn Parquet() -> PyCatalogBackend {
83        Self::new(CatalogBackendType::Parquet)
84    }
85
86    #[staticmethod]
87    fn from_str(value: &str) -> PyResult<Self> {
88        parse_catalog_backend(value).map(Self::new)
89    }
90
91    #[staticmethod]
92    #[pyo3(name = "External")]
93    fn py_external(name: &str) -> PyResult<Self> {
94        let backend = name.parse::<CatalogBackendType>().map_err(to_pyvalue_err)?;
95        match backend {
96            CatalogBackendType::External(_) => Ok(Self::new(backend)),
97            CatalogBackendType::Parquet => Err(to_pyvalue_err(format!(
98                "Catalog backend name '{name}' is reserved for a built-in backend"
99            ))),
100        }
101    }
102
103    #[getter]
104    fn name(&self) -> &str {
105        match self.inner {
106            CatalogBackendType::Parquet => "Parquet",
107            CatalogBackendType::External(_) => "External",
108        }
109    }
110
111    #[getter]
112    fn value(&self) -> String {
113        self.inner.to_string()
114    }
115
116    #[getter]
117    fn external_name(&self) -> Option<&str> {
118        match &self.inner {
119            CatalogBackendType::External(name) => Some(name),
120            CatalogBackendType::Parquet => None,
121        }
122    }
123
124    fn __str__(&self) -> String {
125        self.inner.to_string()
126    }
127
128    fn __repr__(&self) -> String {
129        match &self.inner {
130            CatalogBackendType::Parquet => "CatalogBackend.Parquet".to_string(),
131            CatalogBackendType::External(name) => {
132                format!("CatalogBackend.External({name:?})")
133            }
134        }
135    }
136
137    fn __richcmp__(&self, other: &Self, op: pyo3::pyclass::CompareOp, py: Python<'_>) -> Py<PyAny> {
138        use nautilus_core::python::IntoPyObjectNautilusExt;
139
140        match op {
141            pyo3::pyclass::CompareOp::Eq => (self.inner == other.inner).into_py_any_unwrap(py),
142            pyo3::pyclass::CompareOp::Ne => (self.inner != other.inner).into_py_any_unwrap(py),
143            _ => py.NotImplemented(),
144        }
145    }
146
147    #[expect(
148        clippy::cast_possible_truncation,
149        clippy::cast_possible_wrap,
150        reason = "Python hashes use the platform signed integer width"
151    )]
152    fn __hash__(&self) -> isize {
153        use std::{
154            collections::hash_map::DefaultHasher,
155            hash::{Hash, Hasher},
156        };
157
158        let mut hasher = DefaultHasher::new();
159        self.inner.hash(&mut hasher);
160        hasher.finish() as isize
161    }
162}
163
164#[derive(Clone, Debug)]
165#[pyo3::pyclass(
166    frozen,
167    name = "RotationConfig",
168    module = "nautilus_trader.persistence",
169    from_py_object
170)]
171#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
172pub struct PyRotationConfig {
173    inner: RotationConfig,
174}
175
176impl From<PyRotationConfig> for RotationConfig {
177    fn from(config: PyRotationConfig) -> Self {
178        config.inner
179    }
180}
181
182impl From<RotationConfig> for PyRotationConfig {
183    fn from(config: RotationConfig) -> Self {
184        Self { inner: config }
185    }
186}
187
188#[pyo3_stub_gen::derive::gen_stub_pymethods]
189#[pyo3::pymethods]
190impl PyRotationConfig {
191    #[staticmethod]
192    fn no_rotation() -> Self {
193        Self {
194            inner: RotationConfig::NoRotation,
195        }
196    }
197
198    #[staticmethod]
199    fn size(max_size: u64) -> Self {
200        Self {
201            inner: RotationConfig::Size { max_size },
202        }
203    }
204
205    #[staticmethod]
206    fn interval(interval_ns: u64) -> Self {
207        Self {
208            inner: RotationConfig::Interval {
209                interval_ns: nautilus_core::DurationNanos::new(interval_ns),
210            },
211        }
212    }
213
214    #[staticmethod]
215    fn scheduled_dates(interval_ns: u64, schedule_ns: u64) -> Self {
216        Self {
217            inner: RotationConfig::ScheduledDates {
218                interval_ns: nautilus_core::DurationNanos::new(interval_ns),
219                schedule_ns: UnixNanos::from(schedule_ns),
220            },
221        }
222    }
223
224    #[getter]
225    fn mode(&self) -> &'static str {
226        match self.inner {
227            RotationConfig::Size { .. } => "size",
228            RotationConfig::Interval { .. } => "interval",
229            RotationConfig::ScheduledDates { .. } => "scheduled_dates",
230            RotationConfig::NoRotation => "no_rotation",
231        }
232    }
233
234    #[getter]
235    fn max_size(&self) -> Option<u64> {
236        match self.inner {
237            RotationConfig::Size { max_size } => Some(max_size),
238            _ => None,
239        }
240    }
241
242    #[getter]
243    fn interval_ns(&self) -> Option<u64> {
244        match self.inner {
245            RotationConfig::Interval { interval_ns }
246            | RotationConfig::ScheduledDates { interval_ns, .. } => Some(interval_ns.as_u64()),
247            _ => None,
248        }
249    }
250
251    #[getter]
252    fn schedule_ns(&self) -> Option<u64> {
253        match self.inner {
254            RotationConfig::ScheduledDates { schedule_ns, .. } => Some(schedule_ns.as_u64()),
255            _ => None,
256        }
257    }
258
259    fn __repr__(&self) -> String {
260        format!("{:?}", self.inner)
261    }
262}
263
264#[pyo3_stub_gen::derive::gen_stub_pymethods]
265#[pyo3::pymethods]
266impl StreamingConfig {
267    /// Configuration streaming live or backtest runs to a persistence writer.
268    #[new]
269    #[expect(
270        clippy::too_many_arguments,
271        reason = "the PyO3 constructor mirrors the public Python configuration signature"
272    )]
273    #[pyo3(signature = (
274        catalog_path,
275        fs_protocol = None,
276        flush_interval_ms = 1000,
277        replace_existing = false,
278        rotation_config = None,
279        writer_backend = None,
280        data_types = None,
281        record_types = None,
282        instrument_types = None,
283        record_filters = None,
284        params = None,
285        rotation_mode = None,
286        max_file_size = None,
287        rotation_interval_ns = None,
288        schedule_ns = None,
289    ))]
290    fn py_new(
291        catalog_path: String,
292        fs_protocol: Option<String>,
293        flush_interval_ms: u64,
294        replace_existing: bool,
295        rotation_config: Option<PyRotationConfig>,
296        writer_backend: Option<String>,
297        data_types: Option<&Bound<'_, PyAny>>,
298        record_types: Option<&Bound<'_, PyAny>>,
299        instrument_types: Option<&Bound<'_, PyAny>>,
300        record_filters: Option<&Bound<'_, PyAny>>,
301        params: Option<Py<PyDict>>,
302        rotation_mode: Option<&str>,
303        max_file_size: Option<u64>,
304        rotation_interval_ns: Option<u64>,
305        schedule_ns: Option<u64>,
306    ) -> pyo3::PyResult<Self> {
307        let rotation_config = if let Some(config) = rotation_config {
308            if rotation_mode.is_some()
309                || max_file_size.is_some()
310                || rotation_interval_ns.is_some()
311                || schedule_ns.is_some()
312            {
313                return Err(to_pyvalue_err(
314                    "rotation_config cannot be combined with legacy rotation options",
315                ));
316            }
317            config.into()
318        } else {
319            match rotation_mode
320                .unwrap_or("NO_ROTATION")
321                .to_ascii_uppercase()
322                .as_str()
323            {
324                "SIZE" => {
325                    let max_size = max_file_size.unwrap_or(1_073_741_824);
326                    if max_size == 0 {
327                        return Err(to_pyvalue_err("max_file_size must be positive"));
328                    }
329                    RotationConfig::Size { max_size }
330                }
331                "INTERVAL" => RotationConfig::Interval {
332                    interval_ns: positive_interval(rotation_interval_ns)?,
333                },
334                "SCHEDULED_DATES" => RotationConfig::ScheduledDates {
335                    interval_ns: positive_interval(rotation_interval_ns)?,
336                    schedule_ns: UnixNanos::from(schedule_ns.unwrap_or(0)),
337                },
338                "NO_ROTATION" => RotationConfig::NoRotation,
339                mode => return Err(to_pyvalue_err(format!("Invalid rotation_mode: '{mode}'"))),
340            }
341        };
342        let mut config = Self::new(
343            catalog_path,
344            fs_protocol.unwrap_or_else(default_fs_protocol),
345            flush_interval_ms,
346            replace_existing,
347            rotation_config,
348        );
349        config.writer_backend = writer_backend
350            .map(|backend| backend.parse::<WriterBackendType>())
351            .transpose()
352            .map_err(to_pyvalue_err)?
353            .unwrap_or_default();
354        let mut parsed_types = py_streaming_types_from_any(data_types)?;
355        parsed_types
356            .records
357            .extend(py_record_types_from_any(record_types)?.unwrap_or_default());
358        parsed_types
359            .instruments
360            .extend(py_instrument_types_from_any(instrument_types)?.unwrap_or_default());
361        config.data_types = (!parsed_types.data.is_empty()).then_some(parsed_types.data);
362        config.record_types = (!parsed_types.records.is_empty()).then_some(parsed_types.records);
363        config.instrument_types =
364            (!parsed_types.instruments.is_empty()).then_some(parsed_types.instruments);
365        config.record_filters = py_record_filters_from_any(record_filters)?;
366        config.params = Python::attach(|py| match params {
367            Some(params) => from_pydict(py, &params),
368            None => Ok(None),
369        })?;
370
371        Ok(config)
372    }
373
374    #[getter]
375    fn catalog_path(&self) -> &str {
376        &self.catalog_path
377    }
378
379    #[getter]
380    fn fs_protocol(&self) -> &str {
381        &self.fs_protocol
382    }
383
384    #[getter]
385    const fn flush_interval_ms(&self) -> u64 {
386        self.flush_interval_ms
387    }
388
389    #[getter]
390    const fn replace_existing(&self) -> bool {
391        self.replace_existing
392    }
393
394    #[getter]
395    fn rotation_config(&self) -> PyRotationConfig {
396        self.rotation_config.clone().into()
397    }
398
399    #[getter]
400    fn rotation_mode(&self) -> String {
401        self.rotation_config().mode().to_ascii_uppercase()
402    }
403
404    #[getter]
405    fn max_file_size(&self) -> Option<u64> {
406        self.rotation_config().max_size()
407    }
408
409    #[getter]
410    fn rotation_interval_ns(&self) -> Option<u64> {
411        self.rotation_config().interval_ns()
412    }
413
414    #[getter]
415    fn schedule_ns(&self) -> Option<u64> {
416        self.rotation_config().schedule_ns()
417    }
418
419    #[getter]
420    fn writer_backend(&self) -> String {
421        self.writer_backend.to_string()
422    }
423
424    #[getter]
425    #[pyo3(name = "params")]
426    fn py_params(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
427        self.params
428            .as_ref()
429            .map(|params| params_to_pydict(py, params))
430            .transpose()
431    }
432
433    #[getter]
434    fn data_types(&self) -> Option<Vec<String>> {
435        self.data_types
436            .clone()
437            .map(|values| values.into_iter().map(|value| value.to_string()).collect())
438    }
439
440    #[getter]
441    fn record_types(&self) -> Option<Vec<String>> {
442        self.record_types
443            .clone()
444            .map(|values| values.into_iter().map(|value| value.to_string()).collect())
445    }
446
447    #[getter]
448    fn instrument_types(&self) -> Option<Vec<String>> {
449        self.instrument_types
450            .clone()
451            .map(|values| values.into_iter().map(|value| value.to_string()).collect())
452    }
453
454    #[getter]
455    fn record_filters(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
456        let Some(filters) = &self.record_filters else {
457            return Ok(None);
458        };
459        let result = PyDict::new(py);
460        for filter in filters {
461            result.set_item(filter.record_type.to_string(), filter.identifiers.clone())?;
462        }
463        Ok(Some(result.unbind()))
464    }
465
466    fn __repr__(&self) -> String {
467        format!("{self:?}")
468    }
469}
470
471#[pyo3_stub_gen::derive::gen_stub_pymethods]
472#[pyo3::pymethods]
473impl DataCatalogConfig {
474    /// Configuration for a catalog available to request-time historical data loading.
475    #[new]
476    #[pyo3(signature = (path, fs_protocol = None, catalog_backend = None, params = None, name = None, read_only = false, fs_rust_storage_options = None))]
477    fn py_new(
478        path: String,
479        fs_protocol: Option<String>,
480        catalog_backend: Option<PyRef<'_, PyCatalogBackend>>,
481        params: Option<Py<PyDict>>,
482        name: Option<String>,
483        read_only: bool,
484        fs_rust_storage_options: Option<std::collections::HashMap<String, String>>,
485    ) -> pyo3::PyResult<Self> {
486        let catalog_backend = catalog_backend.map(|backend| backend.inner());
487        let params = Python::attach(|py| match params {
488            Some(params) => from_pydict(py, &params),
489            None => Ok(None),
490        })?;
491        Ok(Self::new(path, fs_protocol, catalog_backend)
492            .with_params(params)
493            .with_name(name)
494            .with_read_only(read_only)
495            .with_storage_options(
496                fs_rust_storage_options.map(|options| options.into_iter().collect()),
497            ))
498    }
499
500    /// Returns the path to the data catalog.
501    #[getter]
502    #[pyo3(name = "path")]
503    fn py_path(&self) -> &str {
504        self.path()
505    }
506
507    /// Returns the catalog registration name.
508    #[getter]
509    #[pyo3(name = "name")]
510    fn py_name(&self) -> Option<&str> {
511        self.name()
512    }
513
514    /// Returns whether the catalog rejects response write-back.
515    #[getter]
516    #[pyo3(name = "read_only")]
517    fn py_read_only(&self) -> bool {
518        self.read_only()
519    }
520
521    /// Returns the fsspec file system protocol for the data catalog.
522    #[getter]
523    #[pyo3(name = "fs_protocol")]
524    fn py_fs_protocol(&self) -> &str {
525        self.fs_protocol()
526    }
527
528    /// Returns the catalog backend implementation to use.
529    #[getter]
530    #[pyo3(name = "catalog_backend")]
531    fn py_catalog_backend(&self) -> PyCatalogBackend {
532        PyCatalogBackend::new(self.catalog_backend().clone())
533    }
534
535    /// Returns backend-specific catalog parameters.
536    #[getter]
537    #[pyo3(name = "params")]
538    fn py_params(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
539        self.params()
540            .map(|params| params_to_pydict(py, params))
541            .transpose()
542    }
543
544    #[getter]
545    fn fs_rust_storage_option_keys(&self) -> Option<Vec<String>> {
546        self.fs_rust_storage_options().map(|options| {
547            let mut keys = options.keys().cloned().collect::<Vec<_>>();
548            keys.sort();
549            keys
550        })
551    }
552
553    fn __repr__(&self) -> String {
554        format!("{self:?}")
555    }
556}
557
558#[derive(Default)]
559struct ParsedStreamingTypes {
560    data: Vec<NautilusDataType>,
561    records: Vec<NautilusRecordType>,
562    instruments: Vec<NautilusInstrumentType>,
563}
564
565fn py_streaming_type_from_any(
566    value: &Bound<'_, PyAny>,
567    parsed: &mut ParsedStreamingTypes,
568) -> pyo3::PyResult<()> {
569    if let Ok(data_type) = value.extract::<PyRef<'_, PyNautilusDataType>>() {
570        parsed.data.push(data_type.inner());
571        return Ok(());
572    }
573
574    if let Ok(record_type) = value.extract::<PyRef<'_, PyNautilusRecordType>>() {
575        parsed.records.push(record_type.inner());
576        return Ok(());
577    }
578
579    if let Ok(instrument_type) = value.extract::<PyRef<'_, PyNautilusInstrumentType>>() {
580        parsed.instruments.push(instrument_type.inner());
581        return Ok(());
582    }
583
584    if let Ok(value) = value.extract::<String>() {
585        if let Ok(data_type) = value.parse::<NautilusDataType>() {
586            parsed.data.push(data_type);
587            return Ok(());
588        }
589
590        if let Ok(record_type) = value.parse::<NautilusRecordType>() {
591            parsed.records.push(record_type);
592            return Ok(());
593        }
594
595        if let Ok(instrument_type) = value.parse::<NautilusInstrumentType>() {
596            parsed.instruments.push(instrument_type);
597            return Ok(());
598        }
599    }
600
601    Err(to_pytype_err(
602        "streaming type must be NautilusDataType, NautilusRecordType, NautilusInstrumentType, or str",
603    ))
604}
605
606fn py_streaming_types_from_any(
607    values: Option<&Bound<'_, PyAny>>,
608) -> pyo3::PyResult<ParsedStreamingTypes> {
609    let mut parsed = ParsedStreamingTypes::default();
610
611    if let Some(values) = values {
612        for item in values.try_iter()? {
613            py_streaming_type_from_any(&item?, &mut parsed)?;
614        }
615    }
616    Ok(parsed)
617}
618
619fn py_record_type_from_any(record_type: &Bound<'_, PyAny>) -> pyo3::PyResult<NautilusRecordType> {
620    if let Ok(record_type) = record_type.extract::<PyRef<'_, PyNautilusRecordType>>() {
621        return Ok(record_type.inner());
622    }
623
624    if let Ok(record_type) = record_type.extract::<String>() {
625        return record_type
626            .parse::<NautilusRecordType>()
627            .map_err(to_pytype_err);
628    }
629
630    Err(to_pytype_err(
631        "record_type must be NautilusRecordType or str",
632    ))
633}
634
635fn py_record_types_from_any(
636    record_types: Option<&Bound<'_, PyAny>>,
637) -> pyo3::PyResult<Option<Vec<NautilusRecordType>>> {
638    record_types
639        .map(|record_types| {
640            record_types
641                .try_iter()?
642                .map(|item| py_record_type_from_any(&item?))
643                .collect::<pyo3::PyResult<Vec<_>>>()
644        })
645        .transpose()
646}
647
648pub(crate) fn py_instrument_type_from_any(
649    instrument_type: &Bound<'_, PyAny>,
650) -> pyo3::PyResult<NautilusInstrumentType> {
651    if let Ok(instrument_type) = instrument_type.extract::<PyRef<'_, PyNautilusInstrumentType>>() {
652        return Ok(instrument_type.inner());
653    }
654
655    if let Ok(instrument_type) = instrument_type.extract::<String>() {
656        return instrument_type
657            .parse::<NautilusInstrumentType>()
658            .map_err(to_pytype_err);
659    }
660
661    Err(to_pytype_err(
662        "instrument_type must be NautilusInstrumentType or str",
663    ))
664}
665
666fn py_instrument_types_from_any(
667    instrument_types: Option<&Bound<'_, PyAny>>,
668) -> pyo3::PyResult<Option<Vec<NautilusInstrumentType>>> {
669    instrument_types
670        .map(|instrument_types| {
671            instrument_types
672                .try_iter()?
673                .map(|item| py_instrument_type_from_any(&item?))
674                .collect::<pyo3::PyResult<Vec<_>>>()
675        })
676        .transpose()
677}
678
679fn py_record_filters_from_any(
680    record_filters: Option<&Bound<'_, PyAny>>,
681) -> pyo3::PyResult<Option<Vec<StreamingRecordFilterConfig>>> {
682    let Some(record_filters) = record_filters else {
683        return Ok(None);
684    };
685    let record_filters = record_filters.cast::<PyDict>()?;
686    let mut filters = Vec::with_capacity(record_filters.len()?);
687    for (record_type, identifiers) in record_filters {
688        let record_type = py_record_type_from_any(&record_type)?;
689        let identifiers = if identifiers.is_none() {
690            None
691        } else if let Ok(identifier) = identifiers.extract::<String>() {
692            Some(vec![identifier])
693        } else {
694            Some(identifiers.extract::<Vec<String>>()?)
695        };
696        filters.push(StreamingRecordFilterConfig {
697            record_type,
698            identifiers,
699        });
700    }
701
702    Ok((!filters.is_empty()).then_some(filters))
703}
704
705fn positive_interval(interval_ns: Option<u64>) -> PyResult<DurationNanos> {
706    let interval_ns = interval_ns.unwrap_or(NANOSECONDS_IN_DAY);
707    if interval_ns == 0 {
708        return Err(to_pyvalue_err("rotation_interval_ns must be positive"));
709    }
710    Ok(DurationNanos::new(interval_ns))
711}
nautilus_persistence/python/config.rs

nautilus_persistence/python/
config.rs
