nautilus_persistence/python/backend/parquet/

catalog.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

#![expect(
    clippy::match_same_arms,
    clippy::too_many_arguments,
    reason = "PyO3 catalog wrapper mirrors Python API dispatch surface"
)]

use std::collections::HashMap;

use nautilus_core::{
    UnixNanos,
    python::{to_pytype_err, to_pyvalue_err},
};
use nautilus_model::{
    data::{
        Bar, Data, FundingRateUpdate, HasTsInit, IndexPriceUpdate, InstrumentStatus,
        MarkPriceUpdate, NautilusDataType, OptionGreeks, OrderBookDelta, OrderBookDepth, QuoteTick,
        TradeTick, close::InstrumentClose,
    },
    python::{
        data::data_to_pyobject,
        instruments::{
            PyNautilusInstrumentType, instrument_any_to_pyobject, pyobject_to_instrument_any,
        },
    },
};
use nautilus_serialization::{
    arrow::{
        DecodeTypedFromRecordBatch, EncodeToRecordBatch, display::instrument::encode_instruments,
    },
    python::arrow::{arrow_record_batches_to_pyarrow_stream, arrow_record_batches_to_pybytes},
};
use pyo3::{
    exceptions::PyIOError,
    prelude::*,
    types::{PyBytes, PyDict, PyList},
};

use crate::{
    backend::{migration::build_catalog_migration_plan, parquet::catalog::ParquetDataCatalog},
    catalog::{
        traits::{CatalogQuery, CatalogReader, CatalogRecordQuery, CatalogWriter},
        types::{
            HasCatalogDataType, custom_data_read_prefixes, custom_type_name,
            parquet_catalog_data_type_path_prefixes,
        },
    },
    python::backend::{
        PyCatalogDataType, arrow_ipc_batches, arrow_ipc_data_schema, arrow_ipc_record_schema,
        arrow_record_batches_from_pybytes, catalog_metadata_to_pydict, catalog_record_type_from_py,
        nautilus_data_type_from_py, to_pyio_err, write_record_params_from_py,
    },
};

#[expect(
    clippy::needless_pass_by_value,
    reason = "PyO3 supplies owned Python data at the catalog write boundary"
)]
fn write_parquet_data<T>(
    catalog: &ParquetDataCatalog,
    data: Vec<T>,
    start: Option<u64>,
    end: Option<u64>,
    skip_disjoint_check: bool,
    label: &str,
) -> PyResult<String>
where
    T: HasTsInit + EncodeToRecordBatch + HasCatalogDataType,
{
    catalog
        .write_to_parquet(
            &data,
            start.map(UnixNanos::from),
            end.map(UnixNanos::from),
            Some(skip_disjoint_check),
        )
        .map(|path| path.to_string_lossy().to_string())
        .map_err(|e| PyIOError::new_err(format!("Failed to write {label}: {e}")))
}

fn query_parquet_data<T>(
    catalog: &mut ParquetDataCatalog,
    identifiers: Option<Vec<String>>,
    start: Option<u64>,
    end: Option<u64>,
    where_clause: Option<&str>,
    files: Option<Vec<String>>,
    optimize_file_loading: bool,
    error_context: &str,
) -> PyResult<Vec<T>>
where
    T: DecodeTypedFromRecordBatch + HasCatalogDataType + HasTsInit,
{
    catalog
        .query_typed_data::<T>(
            identifiers,
            start.map(UnixNanos::from),
            end.map(UnixNanos::from),
            where_clause,
            files,
            optimize_file_loading,
        )
        .map_err(|e| PyIOError::new_err(format!("{error_context}: {e}")))
}

fn reject_parquet_as_of(as_of: Option<&Bound<'_, PyAny>>) -> PyResult<()> {
    if as_of.is_some() {
        return Err(to_pyvalue_err("ParquetDataCatalog does not support as_of"));
    }
    Ok(())
}

/// A catalog for writing data to Parquet files.
#[pyclass(name = "ParquetDataCatalog", module = "nautilus_trader.persistence")]
#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
pub struct PyParquetDataCatalog {
    inner: ParquetDataCatalog,
}

#[pymethods]
#[pyo3_stub_gen::derive::gen_stub_pymethods]
impl PyParquetDataCatalog {
    /// Create a new `ParquetCatalog` with the given base path and optional parameters.
    ///
    /// # Parameters
    ///
    /// - `base_path`: The base path for the catalog
    /// - `storage_options`: Optional storage configuration for cloud backends
    /// - `batch_size`: Optional batch size for processing (default: 5000)
    /// - `compression`: Optional compression type (0=UNCOMPRESSED, 1=SNAPPY, 2=GZIP, 3=LZO, 4=BROTLI, 5=LZ4, 6=ZSTD)
    /// - `max_row_group_size`: Optional maximum row group size (default: 131,072)
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying [`ParquetDataCatalog`] cannot be created.
    #[new]
    #[pyo3(signature = (base_path, storage_options=None, batch_size=None, compression=None, max_row_group_size=None))]
    pub fn py_new(
        base_path: &str,
        storage_options: Option<HashMap<String, String>>,
        batch_size: Option<usize>,
        compression: Option<u8>,
        max_row_group_size: Option<usize>,
    ) -> PyResult<Self> {
        let compression = compression.map(|c| match c {
            0 => parquet::basic::Compression::UNCOMPRESSED,
            1 => parquet::basic::Compression::SNAPPY,
            // For GZIP, LZO, BROTLI, LZ4, ZSTD we need to use the default level
            // since we can't pass the level parameter through PyO3
            2 => {
                let level = parquet::basic::GzipLevel::default();
                parquet::basic::Compression::GZIP(level)
            }
            3 => parquet::basic::Compression::LZO,
            4 => {
                let level = parquet::basic::BrotliLevel::default();
                parquet::basic::Compression::BROTLI(level)
            }
            5 => parquet::basic::Compression::LZ4,
            6 => {
                let level = parquet::basic::ZstdLevel::default();
                parquet::basic::Compression::ZSTD(level)
            }
            _ => parquet::basic::Compression::SNAPPY,
        });

        // Convert HashMap to AHashMap for internal use
        let storage_options = storage_options.map(|m| m.into_iter().collect());

        let inner = ParquetDataCatalog::from_uri(
            base_path,
            storage_options,
            batch_size,
            compression,
            max_row_group_size,
        )
        .map_err(|e| PyIOError::new_err(format!("Failed to create ParquetDataCatalog: {e}")))?;

        Ok(Self { inner })
    }

    /// Rewrites a legacy Parquet catalog into this current Parquet catalog.
    #[pyo3(signature = (parquet_path, storage_options=None, dry_run=false))]
    pub fn migrate_from_legacy_parquet_path(
        mut slf: PyRefMut<'_, Self>,
        parquet_path: &str,
        storage_options: Option<HashMap<String, String>>,
        dry_run: bool,
    ) -> PyResult<usize> {
        let storage_options = storage_options.map(|m| m.into_iter().collect());
        let source = ParquetDataCatalog::from_uri(parquet_path, storage_options, None, None, None)
            .map_err(to_pyio_err)?;
        let py = slf.py();
        if dry_run {
            return py
                .detach(|| {
                    let plan = build_catalog_migration_plan(&source)?;
                    plan.ensure_ready()?;
                    Ok::<usize, anyhow::Error>(0)
                })
                .map_err(to_pyio_err);
        }
        let inner = &mut slf.inner;
        py.detach(|| {
            inner
                .migrate_from_legacy_parquet_catalog(&source)
                .map(|report| report.migrated_rows)
        })
        .map_err(to_pyio_err)
    }

    /// Write quote tick data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of quote ticks to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_quote_ticks(
        &self,
        data: Vec<QuoteTick>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "quote ticks",
        )
    }

    /// Write trade tick data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of trade ticks to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_trade_ticks(
        &self,
        data: Vec<TradeTick>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "trade ticks",
        )
    }

    /// Write order book delta data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of order book deltas to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_order_book_deltas(
        &self,
        data: Vec<OrderBookDelta>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "order book deltas",
        )
    }

    /// Write bar data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of bars to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_bars(
        &self,
        data: Vec<Bar>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(&self.inner, data, start, end, skip_disjoint_check, "bars")
    }

    /// Write order book depth data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of order book depths to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_order_book_depths(
        &self,
        data: Vec<OrderBookDepth>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "order book depths",
        )
    }

    /// Write mark price update data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of mark price updates to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_mark_price_updates(
        &self,
        data: Vec<MarkPriceUpdate>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "mark price updates",
        )
    }

    /// Write index price update data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of index price updates to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_index_price_updates(
        &self,
        data: Vec<IndexPriceUpdate>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "index price updates",
        )
    }

    /// Write option greeks data to Parquet files.
    ///
    /// # Parameters
    ///
    /// - `data`: Vector of option greeks to write
    /// - `start`: Optional start timestamp override (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp override (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns the path of the created file as a string.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_option_greeks(
        &self,
        data: Vec<OptionGreeks>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "option greeks",
        )
    }

    /// Write instrument status data to Parquet files.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_instrument_statuses(
        &self,
        data: Vec<InstrumentStatus>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "instrument statuses",
        )
    }

    /// Write instrument close data to Parquet files.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_instrument_closes(
        &self,
        data: Vec<InstrumentClose>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        write_parquet_data(
            &self.inner,
            data,
            start,
            end,
            skip_disjoint_check,
            "instrument closes",
        )
    }

    /// Write instruments to Parquet files in the catalog.
    ///
    /// Instruments are stored under `data/instruments/{instrument_id}/` using timestamp-ranged
    /// parquet file names, allowing multiple historical versions of the same instrument to be
    /// written across separate calls.
    ///
    /// # Parameters
    ///
    /// - `instruments`: A Python list of instrument objects (e.g. `CurrencyPair`, Equity).
    ///
    /// # Returns
    ///
    /// Returns a list of written file paths.
    pub fn write_instruments(&self, instruments: &Bound<'_, PyAny>) -> PyResult<Vec<String>> {
        let data = instruments;
        let py = data.py();
        let list = data.cast::<PyList>()?;
        let mut instruments = Vec::with_capacity(list.len());
        for item in list.iter() {
            let py_item: Py<PyAny> = item.unbind();
            let instrument = pyobject_to_instrument_any(py, py_item)?;
            instruments.push(instrument);
        }
        self.inner
            .write_instruments(instruments)
            .map(|paths| {
                paths
                    .into_iter()
                    .map(|p| p.to_string_lossy().to_string())
                    .collect()
            })
            .map_err(|e| PyIOError::new_err(format!("Failed to write instruments: {e}")))
    }

    /// Query instruments from the catalog.
    ///
    /// # Parameters
    ///
    /// - `instrument_ids`: Optional list of instrument IDs to filter by. If `None`, returns all instruments.
    /// - `start`: Optional inclusive lower bound for `ts_init` filtering.
    /// - `end`: Optional inclusive upper bound for `ts_init` filtering.
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering.
    ///
    /// # Returns
    ///
    /// Returns a list of instrument objects (e.g. `CurrencyPair`, Equity).
    #[pyo3(signature = (
        instrument_ids=None,
        start=None,
        end=None,
        where_clause=None,
        instrument_type=None,
    ))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn instruments(
        &mut self,
        instrument_ids: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        instrument_type: Option<PyRef<'_, PyNautilusInstrumentType>>,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let instrument_type = instrument_type.map(|instrument_type| instrument_type.inner());
        let rust_instruments = self
            .inner
            .query_instruments_filtered_with_where_and_type(
                instrument_ids.as_deref(),
                start.map(UnixNanos::from),
                end.map(UnixNanos::from),
                where_clause,
                instrument_type.as_ref(),
            )
            .map_err(|e| PyIOError::new_err(format!("Failed to query instruments: {e}")))?;
        Python::attach(|py| {
            rust_instruments
                .into_iter()
                .map(|inst| instrument_any_to_pyobject(py, inst))
                .collect()
        })
    }

    /// Query instruments as display-friendly Arrow IPC stream bytes.
    #[pyo3(signature = (
        instrument_ids=None,
        start=None,
        end=None,
        where_clause=None,
        instrument_type=None,
    ))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn query_instrument_arrow_bytes(
        &mut self,
        py: Python<'_>,
        instrument_ids: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        instrument_type: Option<PyRef<'_, PyNautilusInstrumentType>>,
    ) -> PyResult<Py<PyBytes>> {
        let instrument_type = instrument_type.map(|instrument_type| instrument_type.inner());
        let instruments = py
            .detach(|| {
                self.inner.query_instruments_filtered_with_where_and_type(
                    instrument_ids.as_deref(),
                    start.map(UnixNanos::from),
                    end.map(UnixNanos::from),
                    where_clause,
                    instrument_type.as_ref(),
                )
            })
            .map_err(|e| PyIOError::new_err(format!("Failed query instruments: {e}")))?;
        let batch = encode_instruments(&instruments)
            .map_err(|e| PyIOError::new_err(format!("Failed encode instruments: {e}")))?;
        let schema = batch.schema().as_ref().clone();
        arrow_record_batches_to_pybytes(py, &schema, &[batch])
    }

    /// Query instruments as an Arrow C stream `PyCapsule`.
    #[pyo3(signature = (
        instrument_ids=None,
        start=None,
        end=None,
        where_clause=None,
        instrument_type=None,
    ))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn query_instrument_arrow_stream(
        &mut self,
        py: Python<'_>,
        instrument_ids: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        instrument_type: Option<PyRef<'_, PyNautilusInstrumentType>>,
    ) -> PyResult<Py<PyAny>> {
        let instrument_type = instrument_type.map(|instrument_type| instrument_type.inner());
        let instruments = py
            .detach(|| {
                self.inner.query_instruments_filtered_with_where_and_type(
                    instrument_ids.as_deref(),
                    start.map(UnixNanos::from),
                    end.map(UnixNanos::from),
                    where_clause,
                    instrument_type.as_ref(),
                )
            })
            .map_err(|e| PyIOError::new_err(format!("Failed query instruments: {e}")))?;
        let batch = encode_instruments(&instruments)
            .map_err(|e| PyIOError::new_err(format!("Failed encode instruments: {e}")))?;
        let schema = batch.schema().as_ref().clone();
        arrow_record_batches_to_pyarrow_stream(py, &schema, vec![batch])
    }

    /// Extend file names in the catalog with additional timestamp information.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    /// - `start`: Start timestamp (nanoseconds since Unix epoch)
    /// - `end`: End timestamp (nanoseconds since Unix epoch)
    #[pyo3(signature = (data_type, instrument_id=None, *, start, end))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn extend_file_name(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
        start: u64,
        end: u64,
    ) -> PyResult<()> {
        let data_type = data_type.into_inner();
        let start_nanos = UnixNanos::from(start);
        let end_nanos = UnixNanos::from(end);

        self.inner
            .extend_file_name(&data_type, instrument_id.as_deref(), start_nanos, end_nanos)
            .map_err(|e| PyIOError::new_err(format!("Failed to extend file name: {e}")))
    }

    /// Consolidate all data files in the catalog within the specified time range.
    ///
    /// # Parameters
    ///
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `ensure_contiguous_files`: Optional flag to ensure files are contiguous
    /// - `deduplicate`: Optional flag to deduplicate rows when combining files
    #[pyo3(signature = (start=None, end=None, ensure_contiguous_files=None, deduplicate=None))]
    pub fn consolidate_catalog(
        &self,
        start: Option<u64>,
        end: Option<u64>,
        ensure_contiguous_files: Option<bool>,
        deduplicate: Option<bool>,
    ) -> PyResult<()> {
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .consolidate_catalog(start_nanos, end_nanos, ensure_contiguous_files, deduplicate)
            .map_err(|e| PyIOError::new_err(format!("Failed to consolidate catalog: {e}")))
    }

    /// Consolidate data files for a specific data type within the specified time range.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `ensure_contiguous_files`: Optional flag to ensure files are contiguous
    /// - `deduplicate`: Optional flag to deduplicate rows when combining files
    #[pyo3(signature = (data_type, instrument_id=None, start=None, end=None, ensure_contiguous_files=None, deduplicate=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn consolidate_data(
        &mut self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
        start: Option<u64>,
        end: Option<u64>,
        ensure_contiguous_files: Option<bool>,
        deduplicate: Option<bool>,
    ) -> PyResult<()> {
        let data_type = data_type.into_inner();
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .consolidate_data(
                &data_type,
                instrument_id.as_deref(),
                start_nanos,
                end_nanos,
                ensure_contiguous_files,
                deduplicate,
            )
            .map_err(|e| PyIOError::new_err(format!("Failed to consolidate data: {e}")))
    }

    /// Consolidate all data files in the catalog by splitting them into fixed time periods.
    ///
    /// This method identifies all leaf directories in the catalog that contain parquet files
    /// and consolidates them by period. A leaf directory is one that contains files but no subdirectories.
    /// This is a convenience method that effectively calls `consolidate_data_by_period` for all data types
    /// and instrument IDs in the catalog.
    ///
    /// # Parameters
    ///
    /// - `period_nanos`: Optional period duration for consolidation in nanoseconds. Default is 1 day (86400000000000).
    ///   Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
    /// - `start`: Optional start timestamp for the consolidation range (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp for the consolidation range (nanoseconds since Unix epoch)
    /// - `ensure_contiguous_files`: Optional flag to control file naming strategy
    #[pyo3(signature = (period_nanos=None, start=None, end=None, ensure_contiguous_files=None))]
    pub fn consolidate_catalog_by_period(
        &mut self,
        period_nanos: Option<u64>,
        start: Option<u64>,
        end: Option<u64>,
        ensure_contiguous_files: Option<bool>,
    ) -> PyResult<()> {
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .consolidate_catalog_by_period(
                period_nanos,
                start_nanos,
                end_nanos,
                ensure_contiguous_files,
            )
            .map_err(|e| {
                PyIOError::new_err(format!("Failed to consolidate catalog by period: {e}"))
            })
    }

    /// Consolidate data files by splitting them into fixed time periods.
    ///
    /// This method queries data by period and writes consolidated files immediately,
    /// using efficient period-based consolidation logic. When start/end boundaries intersect existing files,
    /// the function automatically splits those files to preserve all data.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `identifier`: Optional instrument ID to consolidate. If None, consolidates all instruments
    /// - `period_nanos`: Optional period duration for consolidation in nanoseconds. Default is 1 day (86400000000000).
    ///   Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
    /// - `start`: Optional start timestamp for consolidation range (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp for consolidation range (nanoseconds since Unix epoch)
    /// - `ensure_contiguous_files`: Optional flag to control file naming strategy
    #[pyo3(signature = (data_type, identifier=None, period_nanos=None, start=None, end=None, ensure_contiguous_files=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn consolidate_data_by_period(
        &mut self,
        data_type: PyCatalogDataType,
        identifier: Option<String>,
        period_nanos: Option<u64>,
        start: Option<u64>,
        end: Option<u64>,
        ensure_contiguous_files: Option<bool>,
    ) -> PyResult<()> {
        let data_type = data_type.into_inner();
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .consolidate_data_by_period(
                &data_type,
                identifier.as_deref(),
                period_nanos,
                start_nanos,
                end_nanos,
                ensure_contiguous_files,
            )
            .map_err(|e| PyIOError::new_err(format!("Failed to consolidate data by period: {e}")))
    }

    /// Reset all catalog file names to their canonical form.
    pub fn reset_all_file_names(&self) -> PyResult<()> {
        self.inner
            .reset_all_file_names()
            .map_err(|e| PyIOError::new_err(format!("Failed to reset catalog file names: {e}")))
    }

    /// Reset data file names for a specific data class to their canonical form.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    #[pyo3(signature = (data_type, instrument_id=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn reset_data_file_names(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
    ) -> PyResult<()> {
        let data_type = data_type.into_inner();
        self.inner
            .reset_data_file_names(&data_type, instrument_id.as_deref())
            .map_err(|e| PyIOError::new_err(format!("Failed to reset data file names: {e}")))
    }

    /// Delete data within a specified time range across the entire catalog.
    ///
    /// This method identifies all leaf directories in the catalog that contain parquet files
    /// and deletes data within the specified time range from each directory. A leaf directory
    /// is one that contains files but no subdirectories. This is a convenience method that
    /// effectively calls `delete_data_range` for all data types and instrument IDs in the catalog.
    ///
    /// # Parameters
    ///
    /// - `start`: Optional start timestamp for the deletion range (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp for the deletion range (nanoseconds since Unix epoch)
    ///
    /// # Notes
    ///
    /// - This operation permanently removes data and cannot be undone
    /// - The deletion process handles file intersections intelligently by splitting files
    ///   when they partially overlap with the deletion range
    /// - Files completely within the deletion range are removed entirely
    /// - Files partially overlapping the deletion range are split to preserve data outside the range
    /// - This method is useful for bulk data cleanup operations across the entire catalog
    /// - Empty directories are not automatically removed after deletion
    #[pyo3(signature = (start=None, end=None))]
    pub fn delete_catalog_range(&mut self, start: Option<u64>, end: Option<u64>) -> PyResult<()> {
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .delete_catalog_range(start_nanos, end_nanos)
            .map_err(|e| PyIOError::new_err(format!("Failed to delete catalog range: {e}")))
    }

    /// Delete data within a specified time range for a specific data type and instrument.
    ///
    /// This method identifies all parquet files that intersect with the specified time range
    /// and handles them appropriately:
    /// - Files completely within the range are deleted
    /// - Files partially overlapping the range are split to preserve data outside the range
    /// - The original intersecting files are removed after processing
    ///
    /// # Parameters
    ///
    /// - `data_type`: The data type to delete from.
    /// - `identifier`: Optional identifier to delete data for. If None, deletes data across all identifiers
    /// - `start`: Optional start timestamp for the deletion range (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp for the deletion range (nanoseconds since Unix epoch)
    ///
    /// # Notes
    ///
    /// - This operation permanently removes data and cannot be undone
    /// - Files that partially overlap the deletion range are split to preserve data outside the range
    /// - The method ensures data integrity by using atomic operations where possible
    /// - Empty directories are not automatically removed after deletion
    #[pyo3(signature = (data_type, identifier=None, start=None, end=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn delete_data_range(
        &mut self,
        #[gen_stub(override_type(type_repr = "model.NautilusDataType"))] data_type: &Bound<
            '_,
            PyAny,
        >,
        identifier: Option<String>,
        start: Option<u64>,
        end: Option<u64>,
    ) -> PyResult<()> {
        let data_type = nautilus_data_type_from_py(data_type)?;
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .delete_data_range(&data_type, identifier.as_deref(), start_nanos, end_nanos)
            .map_err(|e| PyIOError::new_err(format!("Failed to delete data range: {e}")))
    }

    /// Writes custom data to Parquet files.
    ///
    /// Requires `CustomData` wrappers. Callers must wrap raw custom objects in
    /// `CustomData(data_type=DataType(cls, metadata=...), data=...)` before writing.
    ///
    /// The registered Arrow schema must contain `ts_init`. Any `ts_event` or `ts_init` fields
    /// must use `timestamp("ns", tz="UTC")`; incompatible schemas fail before writing.
    #[pyo3(signature = (data, start=None, end=None, skip_disjoint_check=false))]
    pub fn write_custom_data(
        &self,
        _py: Python<'_>,
        data: Vec<Bound<'_, PyAny>>,
        start: Option<u64>,
        end: Option<u64>,
        skip_disjoint_check: bool,
    ) -> PyResult<String> {
        use nautilus_model::data::CustomData;

        let mut custom_items: Vec<CustomData> = Vec::with_capacity(data.len());
        for obj in data {
            let custom = obj.extract::<CustomData>().map_err(|_| {
                to_pytype_err(
                    "write_custom_data requires CustomData wrappers; wrap with CustomData(data_type=DataType(cls, metadata=...), data=...)",
                )
            })?;
            custom_items.push(custom);
        }

        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .write_custom_data_batch(
                &custom_items,
                start_nanos,
                end_nanos,
                Some(skip_disjoint_check),
            )
            .map(|path| path.to_string_lossy().to_string())
            .map_err(|e| PyIOError::new_err(format!("Failed to write custom data: {e}")))
    }

    /// List all instrument IDs available in the catalog for a given catalog type.
    pub fn list_instruments(&self, data_type: PyCatalogDataType) -> PyResult<Vec<String>> {
        let data_type = data_type.into_inner();
        self.inner
            .list_instruments(&data_type)
            .map_err(|e| PyIOError::new_err(format!("Failed to list instruments: {e}")))
    }

    /// List all Parquet files in the catalog for a given data type and instrument.
    pub fn list_parquet_files(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: &str,
    ) -> PyResult<Vec<String>> {
        let data_type = data_type.into_inner();
        let mut files = Vec::new();

        if let Some(type_name) = custom_type_name(&data_type) {
            for prefix in custom_data_read_prefixes(type_name) {
                let directory = format!("data/{prefix}/{instrument_id}");
                files.extend(self.inner.list_parquet_files(&directory).map_err(|e| {
                    PyIOError::new_err(format!("Failed to list parquet files: {e}"))
                })?);
            }

            files.sort();
            files.dedup();
            return Ok(files);
        }

        for prefix in parquet_catalog_data_type_path_prefixes(&data_type) {
            let prefix = prefix.as_ref();
            let directory = format!("data/{prefix}/{instrument_id}");
            files.extend(
                self.inner.list_parquet_files(&directory).map_err(|e| {
                    PyIOError::new_err(format!("Failed to list parquet files: {e}"))
                })?,
            );
        }

        Ok(files)
    }

    /// Query files in the catalog matching the specified criteria.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `identifiers`: Optional list of identifiers to filter by. Can be `instrument_id` strings
    ///   (e.g., "EUR/USD.SIM") or `bar_type` strings (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL").
    ///   For bars, partial matching is supported.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    ///
    /// # Returns
    ///
    /// Returns a list of file paths matching the criteria.
    #[pyo3(signature = (data_type, identifiers=None, start=None, end=None))]
    pub fn query_files(
        &self,
        data_type: PyCatalogDataType,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
    ) -> PyResult<Vec<String>> {
        let data_type = data_type.into_inner();
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        self.inner
            .query_files(&data_type, identifiers, start_nanos, end_nanos)
            .map_err(|e| PyIOError::new_err(format!("Failed to query files list: {e}")))
    }

    /// Query catalog metadata keyed by the first timestamp where each metadata is used.
    #[pyo3(signature = (data_type, identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_metadata(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusDataType"))] data_type: &Bound<
            '_,
            PyAny,
        >,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Py<PyDict>> {
        let data_type = nautilus_data_type_from_py(data_type)?;
        let metadata = py
            .detach(|| {
                CatalogReader::query_metadata(
                    &mut self.inner,
                    &CatalogQuery::new(data_type)
                        .with_identifiers(identifiers)
                        .with_range(start.map(UnixNanos::from), end.map(UnixNanos::from))
                        .with_where_clause(where_clause.map(str::to_string)),
                )
            })
            .map_err(|e| PyIOError::new_err(format!("Metadata query failed: {e}")))?;

        catalog_metadata_to_pydict(py, metadata)
    }

    /// Query catalog data as display-friendly Arrow IPC stream bytes.
    #[pyo3(signature = (
        data_type,
        identifiers=None,
        start=None,
        end=None,
        where_clause=None,
        display=true,
        as_of=None,
    ))]
    pub fn query_data_arrow_bytes(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusDataType"))] data_type: &Bound<
            '_,
            PyAny,
        >,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        display: bool,
        as_of: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Py<PyBytes>> {
        let data_type = nautilus_data_type_from_py(data_type)?;
        reject_parquet_as_of(as_of)?;
        let query = CatalogQuery::new(data_type.clone())
            .with_identifiers(identifiers)
            .with_range(start.map(UnixNanos::from), end.map(UnixNanos::from))
            .with_where_clause(where_clause.map(str::to_string));
        let batches = py
            .detach(|| {
                if display {
                    CatalogReader::query_display_record_batches(&mut self.inner, &query)
                } else {
                    let data = CatalogReader::query_batch(&mut self.inner, &query)?
                        .to_data_vec_for_compat();
                    crate::common::arrow::data_to_arrow_batches(&data_type, data)
                }
            })
            .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?;
        let schema = arrow_ipc_data_schema(&data_type, &batches, display)?;
        let batches = arrow_ipc_batches(&schema, batches)?;
        arrow_record_batches_to_pybytes(py, &schema, &batches)
    }

    /// Query catalog data as an Arrow C stream `PyCapsule`.
    #[pyo3(signature = (
        data_type,
        identifiers=None,
        start=None,
        end=None,
        where_clause=None,
        display=true,
        as_of=None,
    ))]
    pub fn query_data_arrow_stream(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusDataType"))] data_type: &Bound<
            '_,
            PyAny,
        >,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        display: bool,
        as_of: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Py<PyAny>> {
        let data_type = nautilus_data_type_from_py(data_type)?;
        reject_parquet_as_of(as_of)?;
        let query = CatalogQuery::new(data_type.clone())
            .with_identifiers(identifiers)
            .with_range(start.map(UnixNanos::from), end.map(UnixNanos::from))
            .with_where_clause(where_clause.map(str::to_string));
        let batches = py
            .detach(|| {
                if display {
                    CatalogReader::query_display_record_batches(&mut self.inner, &query)
                } else {
                    let data = CatalogReader::query_batch(&mut self.inner, &query)?
                        .to_data_vec_for_compat();
                    crate::common::arrow::data_to_arrow_batches(&data_type, data)
                }
            })
            .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?;
        let schema = arrow_ipc_data_schema(&data_type, &batches, display)?;
        let batches = arrow_ipc_batches(&schema, batches)?;
        arrow_record_batches_to_pyarrow_stream(py, &schema, batches)
    }

    /// Write catalog records from Arrow IPC stream bytes.
    #[pyo3(signature = (record_type, data, identifier=None, params=None))]
    pub fn write_record_arrow_bytes(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusRecordType"))] record_type: &Bound<
            '_,
            PyAny,
        >,
        data: Vec<u8>,
        identifier: Option<String>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        let record_type = catalog_record_type_from_py(record_type)?;
        let batches = arrow_record_batches_from_pybytes(data)?;
        let params = write_record_params_from_py(py, identifier, params)?;

        py.detach(|| CatalogWriter::write_records(&mut self.inner, record_type, &batches, params))
            .map_err(|e| PyIOError::new_err(format!("Failed write records: {e}")))
    }

    /// Query catalog records and return Arrow IPC stream bytes.
    #[pyo3(signature = (
        record_type,
        identifier=None,
        start=None,
        end=None,
        where_clause=None,
        display=true,
        as_of=None,
    ))]
    pub fn query_record_arrow_bytes(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusRecordType"))] record_type: &Bound<
            '_,
            PyAny,
        >,
        identifier: Option<String>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        display: bool,
        as_of: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Py<PyBytes>> {
        let record_type = catalog_record_type_from_py(record_type)?;
        reject_parquet_as_of(as_of)?;
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);
        let batches = py
            .detach(|| {
                if display {
                    CatalogReader::query_record_display_batches(
                        &mut self.inner,
                        &CatalogRecordQuery::new(record_type)
                            .with_identifier(identifier)
                            .with_range(start_nanos, end_nanos)
                            .with_where_clause(where_clause.map(str::to_string)),
                    )
                } else {
                    CatalogReader::query_record_batches(
                        &mut self.inner,
                        &CatalogRecordQuery::new(record_type)
                            .with_identifier(identifier)
                            .with_range(start_nanos, end_nanos)
                            .with_where_clause(where_clause.map(str::to_string)),
                    )
                }
            })
            .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?;
        let schema = arrow_ipc_record_schema(record_type, &batches)?;
        let batches = arrow_ipc_batches(&schema, batches)?;

        arrow_record_batches_to_pybytes(py, &schema, &batches)
    }

    /// Query catalog records as an Arrow C stream `PyCapsule`.
    #[pyo3(signature = (
        record_type,
        identifier=None,
        start=None,
        end=None,
        where_clause=None,
        display=true,
        as_of=None,
    ))]
    pub fn query_record_arrow_stream(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusRecordType"))] record_type: &Bound<
            '_,
            PyAny,
        >,
        identifier: Option<String>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        display: bool,
        as_of: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Py<PyAny>> {
        let record_type = catalog_record_type_from_py(record_type)?;
        reject_parquet_as_of(as_of)?;
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);
        let batches = py
            .detach(|| {
                if display {
                    CatalogReader::query_record_display_batches(
                        &mut self.inner,
                        &CatalogRecordQuery::new(record_type)
                            .with_identifier(identifier)
                            .with_range(start_nanos, end_nanos)
                            .with_where_clause(where_clause.map(str::to_string)),
                    )
                } else {
                    CatalogReader::query_record_batches(
                        &mut self.inner,
                        &CatalogRecordQuery::new(record_type)
                            .with_identifier(identifier)
                            .with_range(start_nanos, end_nanos)
                            .with_where_clause(where_clause.map(str::to_string)),
                    )
                }
            })
            .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?;
        let schema = arrow_ipc_record_schema(record_type, &batches)?;
        let batches = arrow_ipc_batches(&schema, batches)?;

        arrow_record_batches_to_pyarrow_stream(py, &schema, batches)
    }

    /// Get missing time intervals for a data request.
    ///
    /// # Parameters
    ///
    /// - `start`: Start timestamp (nanoseconds since Unix epoch)
    /// - `end`: End timestamp (nanoseconds since Unix epoch)
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    ///
    /// # Returns
    ///
    /// Returns a list of (start, end) timestamp tuples representing missing intervals.
    #[pyo3(signature = (start, end, data_type, instrument_id=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn get_missing_intervals_for_request(
        &self,
        start: u64,
        end: u64,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
    ) -> PyResult<Vec<(u64, u64)>> {
        let data_type = data_type.into_inner();
        self.inner
            .get_missing_intervals_for_request(start, end, &data_type, instrument_id.as_deref())
            .map_err(|e| PyIOError::new_err(format!("Failed to get missing intervals: {e}")))
    }

    /// Query the first timestamp for a specific data class and instrument.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    ///
    /// # Returns
    ///
    /// Returns the first timestamp as nanoseconds since Unix epoch, or None if no data exists.
    #[pyo3(signature = (data_type, instrument_id=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn query_first_timestamp(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
    ) -> PyResult<Option<u64>> {
        let data_type = data_type.into_inner();
        self.inner
            .query_first_timestamp(&data_type, instrument_id.as_deref())
            .map_err(|e| PyIOError::new_err(format!("Failed to query first timestamp: {e}")))
    }

    /// Query the last timestamp for a specific data class and instrument.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    ///
    /// # Returns
    ///
    /// Returns the last timestamp as nanoseconds since Unix epoch, or None if no data exists.
    #[pyo3(signature = (data_type, instrument_id=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn query_last_timestamp(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
    ) -> PyResult<Option<u64>> {
        let data_type = data_type.into_inner();
        self.inner
            .query_last_timestamp(&data_type, instrument_id.as_deref())
            .map_err(|e| PyIOError::new_err(format!("Failed to query last timestamp: {e}")))
    }

    /// Get time intervals covered by data for a specific data class and instrument.
    ///
    /// # Parameters
    ///
    /// - `data_type`: The stored family to target (data type, record type, or instrument type).
    /// - `instrument_id`: Optional instrument ID filter
    ///
    /// # Returns
    ///
    /// Returns a list of (start, end) timestamp tuples representing covered intervals.
    #[pyo3(signature = (data_type, instrument_id=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn get_intervals(
        &self,
        data_type: PyCatalogDataType,
        instrument_id: Option<String>,
    ) -> PyResult<Vec<(u64, u64)>> {
        let data_type = data_type.into_inner();
        self.inner
            .get_intervals(&data_type, instrument_id.as_deref())
            .map_err(|e| PyIOError::new_err(format!("Failed to get intervals: {e}")))
    }

    /// Queries one data family and returns the decoded Python objects.
    #[pyo3(signature = (data_type, identifiers=None, start=None, end=None, where_clause=None, files=None, optimize_file_loading=true))]
    #[expect(
        clippy::too_many_arguments,
        reason = "PyO3 signature mirrors the catalog query filters"
    )]
    pub fn query(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.NautilusDataType"))] data_type: &Bound<
            '_,
            PyAny,
        >,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
        files: Option<Vec<String>>,
        optimize_file_loading: bool,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let data_type = nautilus_data_type_from_py(data_type)?;
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        macro_rules! typed {
            ($type:ty) => {
                query_parquet_data::<$type>(
                    &mut self.inner,
                    identifiers,
                    start,
                    end,
                    where_clause,
                    files,
                    optimize_file_loading,
                    "Query failed",
                )?
                .into_iter()
                .map(Data::from)
                .collect::<Vec<Data>>()
            };
        }

        let data = match data_type {
            NautilusDataType::QuoteTick => typed!(QuoteTick),
            NautilusDataType::TradeTick => typed!(TradeTick),
            NautilusDataType::Bar => typed!(Bar),
            NautilusDataType::OrderBookDelta => typed!(OrderBookDelta),
            NautilusDataType::OrderBookDepth => typed!(OrderBookDepth),
            NautilusDataType::IndexPriceUpdate => typed!(IndexPriceUpdate),
            NautilusDataType::MarkPriceUpdate => typed!(MarkPriceUpdate),
            NautilusDataType::FundingRateUpdate => typed!(FundingRateUpdate),
            NautilusDataType::OptionGreeks => typed!(OptionGreeks),
            NautilusDataType::InstrumentStatus => typed!(InstrumentStatus),
            NautilusDataType::InstrumentClose => typed!(InstrumentClose),
            NautilusDataType::Custom { type_name } => py
                .detach(|| {
                    self.inner.query_custom_data_dynamic(
                        &type_name,
                        identifiers.as_deref(),
                        start_nanos,
                        end_nanos,
                        where_clause,
                        files.clone(),
                        optimize_file_loading,
                    )
                })
                .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?,
            NautilusDataType::Instrument => {
                let instruments = py
                    .detach(|| {
                        self.inner.query_instruments_filtered_with_where(
                            identifiers.as_deref(),
                            start_nanos,
                            end_nanos,
                            where_clause,
                        )
                    })
                    .map_err(|e| PyIOError::new_err(format!("Query failed: {e}")))?;
                return instruments
                    .into_iter()
                    .map(|instrument| instrument_any_to_pyobject(py, instrument))
                    .collect();
            }
            #[cfg(feature = "defi")]
            NautilusDataType::Defi => {
                return Err(to_pytype_err("Defi data is not supported by query"));
            }
        };

        let mut python_objects = Vec::new();
        for item in data {
            python_objects.push(data_to_pyobject(py, item)?);
        }
        Ok(python_objects)
    }

    /// Query quote tick data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by. Can be `instrument_id` strings
    ///   (e.g., "EUR/USD.SIM") or `bar_type` strings (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL").
    ///   For bars, partial matching is supported.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `QuoteTick` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_quote_ticks(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<QuoteTick>> {
        query_parquet_data::<QuoteTick>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query trade tick data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by. Can be `instrument_id` strings
    ///   (e.g., "EUR/USD.SIM") or `bar_type` strings (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL").
    ///   For bars, partial matching is supported.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `TradeTick` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_trade_ticks(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<TradeTick>> {
        query_parquet_data::<TradeTick>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query order book delta data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by. Can be `instrument_id` strings
    ///   (e.g., "EUR/USD.SIM") or `bar_type` strings (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL").
    ///   For bars, partial matching is supported.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `OrderBookDelta` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_order_book_deltas(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<OrderBookDelta>> {
        query_parquet_data::<OrderBookDelta>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query bar data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by. Can be `instrument_id` strings
    ///   (e.g., "EUR/USD.SIM") or `bar_type` strings (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL").
    ///   For bars, partial matching is supported (e.g., "EUR/USD.SIM" will match all bar types for that instrument).
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of Bar objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_bars(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<Bar>> {
        query_parquet_data::<Bar>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query order book depth data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `OrderBookDepth` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_order_book_depths(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<OrderBookDepth>> {
        query_parquet_data::<OrderBookDepth>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query mark price update data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `MarkPriceUpdate` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_mark_price_updates(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<MarkPriceUpdate>> {
        query_parquet_data::<MarkPriceUpdate>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query index price update data from Parquet files.
    ///
    /// # Parameters
    ///
    /// - `identifiers`: Optional list of identifiers to filter by.
    /// - `start`: Optional start timestamp (nanoseconds since Unix epoch)
    /// - `end`: Optional end timestamp (nanoseconds since Unix epoch)
    /// - `where_clause`: Optional SQL WHERE clause for additional filtering
    ///
    /// # Returns
    ///
    /// Returns a vector of `IndexPriceUpdate` objects matching the query criteria.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_index_price_updates(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<IndexPriceUpdate>> {
        query_parquet_data::<IndexPriceUpdate>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// Query option greeks data from Parquet files.
    #[pyo3(signature = (identifiers=None, start=None, end=None, where_clause=None))]
    pub fn query_option_greeks(
        &mut self,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<OptionGreeks>> {
        query_parquet_data::<OptionGreeks>(
            &mut self.inner,
            identifiers,
            start,
            end,
            where_clause,
            None,
            true,
            "Failed to query data",
        )
    }

    /// List all data types available in the catalog.
    ///
    /// # Returns
    ///
    /// Returns a list of data type names (as directory stems) in the catalog.
    pub fn list_data_types(&self) -> PyResult<Vec<String>> {
        self.inner
            .list_data_types()
            .map_err(|e| PyIOError::new_err(format!("Failed to list data types: {e}")))
    }

    /// List all live run IDs available in the catalog.
    ///
    /// # Returns
    ///
    /// Returns a list of live run IDs (as directory stems) in the catalog.
    pub fn list_live_runs(&self) -> PyResult<Vec<String>> {
        self.inner
            .list_live_runs()
            .map_err(|e| PyIOError::new_err(format!("Failed to list live runs: {e}")))
    }

    /// List all backtest run IDs available in the catalog.
    ///
    /// # Returns
    ///
    /// Returns a list of backtest run IDs (as directory stems) in the catalog.
    pub fn list_backtest_runs(&self) -> PyResult<Vec<String>> {
        self.inner
            .list_backtest_runs()
            .map_err(|e| PyIOError::new_err(format!("Failed to list backtest runs: {e}")))
    }

    /// List all backtest run instances available in the catalog.
    pub fn list_backtests(&self) -> PyResult<Vec<String>> {
        self.inner
            .list_backtest_runs()
            .map_err(|e| PyIOError::new_err(format!("Failed to list backtests: {e}")))
    }

    /// Read data from a live run instance.
    ///
    /// # Parameters
    ///
    /// - `instance_id`: The ID of the live run instance
    ///
    /// # Returns
    ///
    /// Returns a list of data objects from the live run, sorted by timestamp.
    #[pyo3(signature = (instance_id))]
    pub fn read_live_run(&self, py: Python<'_>, instance_id: &str) -> PyResult<Vec<Py<PyAny>>> {
        let data = self
            .inner
            .read_live_run(instance_id)
            .map_err(|e| PyIOError::new_err(format!("Failed to read live run: {e}")))?;

        let mut python_objects = Vec::new();
        for item in data {
            python_objects.push(data_to_pyobject(py, item)?);
        }
        Ok(python_objects)
    }

    /// Read data from a backtest run instance.
    ///
    /// # Parameters
    ///
    /// - `instance_id`: The ID of the backtest run instance
    ///
    /// # Returns
    ///
    /// Returns a list of data objects from the backtest run, sorted by timestamp.
    #[pyo3(signature = (instance_id))]
    pub fn read_backtest(&self, py: Python<'_>, instance_id: &str) -> PyResult<Vec<Py<PyAny>>> {
        let data = self
            .inner
            .read_backtest(instance_id)
            .map_err(|e| PyIOError::new_err(format!("Failed to read backtest: {e}")))?;

        let mut python_objects = Vec::new();
        for item in data {
            python_objects.push(data_to_pyobject(py, item)?);
        }
        Ok(python_objects)
    }

    /// Convert stream data from feather files to parquet files.
    ///
    /// This method reads data from feather files generated during a backtest or live run
    /// and writes it to the catalog in parquet format. It's useful for converting temporary
    /// stream data into a more permanent and queryable format.
    ///
    /// # Parameters
    ///
    /// - `instance_id`: The ID of the backtest or live run instance
    /// - `data_type`: The stored family to convert (data type or record type).
    /// - `subdirectory`: Optional subdirectory containing the feather files. Either "backtest" or "live" (default: "backtest")
    /// - `identifiers`: Optional list of identifiers to filter by (instrument IDs or bar types)
    /// - `use_ts_event_for_ts_init`: If true, replaces the `ts_init` column with `ts_event` column values before deserializing
    ///
    /// # Returns
    ///
    /// Returns nothing on success.
    ///
    /// # Examples
    ///
    /// ```python
    /// # Convert backtest stream data to parquet
    /// catalog.convert_stream_to_data(
    ///     "instance-123",
    ///     NautilusDataType.QuoteTick,
    ///     subdirectory="backtest"
    /// )
    ///
    /// # Convert live run data with identifier filtering
    /// catalog.convert_stream_to_data(
    ///     "instance-456",
    ///     NautilusDataType.TradeTick,
    ///     subdirectory="live",
    ///     identifiers=["EUR/USD.SIM"]
    /// )
    /// ```
    #[pyo3(signature = (instance_id, data_type, subdirectory=None, identifiers=None, use_ts_event_for_ts_init=false))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn convert_stream_to_data(
        &mut self,
        instance_id: &str,
        data_type: PyCatalogDataType,
        subdirectory: Option<&str>,
        identifiers: Option<Vec<String>>,
        use_ts_event_for_ts_init: bool,
    ) -> PyResult<()> {
        let data_type = data_type.into_inner();
        let subdir = subdirectory.unwrap_or("backtest");

        self.inner
            .convert_stream_to_data(
                instance_id,
                &data_type,
                Some(subdir),
                identifiers.as_deref(),
                use_ts_event_for_ts_init,
            )
            .map_err(|e| PyIOError::new_err(format!("Failed to convert stream to data: {e}")))
    }

    /// Query custom data from Parquet files.
    #[pyo3(signature = (type_name, identifiers=None, start=None, end=None, where_clause=None))]
    #[expect(clippy::needless_pass_by_value)]
    pub fn query_custom_data(
        &mut self,
        py: Python<'_>,
        type_name: &str,
        identifiers: Option<Vec<String>>,
        start: Option<u64>,
        end: Option<u64>,
        where_clause: Option<&str>,
    ) -> PyResult<Vec<Py<PyAny>>> {
        let start_nanos = start.map(UnixNanos::from);
        let end_nanos = end.map(UnixNanos::from);

        let data = py
            .detach(|| {
                self.inner.query_custom_data_dynamic(
                    type_name,
                    identifiers.as_deref(),
                    start_nanos,
                    end_nanos,
                    where_clause,
                    None,
                    true,
                )
            })
            .map_err(|e| PyIOError::new_err(format!("Failed to query custom data: {e}")))?;

        let mut python_objects = Vec::new();

        for item in data {
            let py_obj: Py<PyAny> = match item {
                Data::Custom(custom) => Py::new(py, custom)?.into_any(),
                _ => return Err(PyIOError::new_err("Expected custom data")),
            };
            python_objects.push(py_obj);
        }
        Ok(python_objects)
    }
}