nautilus_tardis/csv/

load.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.
// -------------------------------------------------------------------------------------------------

use std::{error::Error, path::Path};

use ahash::AHashMap;
use csv::StringRecord;
use nautilus_core::UnixNanos;
use nautilus_model::{
    data::{
        DEPTH10_LEN, Data, FundingRateUpdate, NULL_ORDER, OrderBookDelta, OrderBookDepth,
        QuoteTick, TradeTick,
    },
    enums::{OrderSide, RecordFlag},
    identifiers::InstrumentId,
    types::{Quantity, fixed::FIXED_PRECISION},
};

use crate::{
    common::parse::{parse_instrument_id, parse_timestamp},
    csv::{
        create_book_order, create_csv_reader, infer_precision, matches_underlying_filter,
        normalize_underlying_filters, parse_delta_record, parse_derivative_ticker_record,
        parse_options_chain_record, parse_options_chain_record_as_quote, parse_quote_record,
        parse_trade_record,
        record::{
            TardisBookUpdateRecord, TardisDerivativeTickerRecord, TardisOptionsChainRecord,
            TardisOrderBookSnapshot5Record, TardisOrderBookSnapshot25Record, TardisQuoteRecord,
            TardisTradeRecord,
        },
    },
};

#[derive(Debug, Clone, Copy)]
pub(in crate::csv) struct OptionsChainPrecision {
    pub(in crate::csv) price: u8,
    pub(in crate::csv) size: u8,
}

impl OptionsChainPrecision {
    pub(in crate::csv) const fn new(
        price_precision: Option<u8>,
        size_precision: Option<u8>,
    ) -> Self {
        Self {
            price: match price_precision {
                Some(precision) => precision,
                None => 0,
            },
            size: match size_precision {
                Some(precision) => precision,
                None => 0,
            },
        }
    }

    pub(in crate::csv) fn update(
        &mut self,
        record: &TardisOptionsChainRecord,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
    ) {
        if price_precision.is_none() {
            for value in [record.last_price, record.bid_price, record.ask_price]
                .into_iter()
                .flatten()
            {
                update_precision_if_needed(&mut self.price, value, price_precision);
            }
        }

        if size_precision.is_none() {
            for value in [record.bid_amount, record.ask_amount].into_iter().flatten() {
                update_precision_if_needed(&mut self.size, value, size_precision);
            }
        }
    }
}

fn update_precision_if_needed(current: &mut u8, value: f64, explicit: Option<u8>) -> bool {
    if explicit.is_some() {
        return false;
    }

    let inferred = infer_precision(value).min(FIXED_PRECISION);
    if inferred > *current {
        *current = inferred;
        true
    } else {
        false
    }
}

fn update_deltas_precision(
    deltas: &mut [OrderBookDelta],
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    current_price_precision: u8,
    current_size_precision: u8,
) {
    for delta in deltas {
        if price_precision.is_none() {
            delta.order.price.precision = current_price_precision;
        }

        if size_precision.is_none() {
            delta.order.size.precision = current_size_precision;
        }
    }
}

fn update_quotes_precision(
    quotes: &mut [QuoteTick],
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    current_price_precision: u8,
    current_size_precision: u8,
) {
    for quote in quotes {
        if price_precision.is_none() {
            quote.bid_price.precision = current_price_precision;
            quote.ask_price.precision = current_price_precision;
        }

        if size_precision.is_none() {
            quote.bid_size.precision = current_size_precision;
            quote.ask_size.precision = current_size_precision;
        }
    }
}

fn update_trades_precision(
    trades: &mut [TradeTick],
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    current_price_precision: u8,
    current_size_precision: u8,
) {
    for trade in trades {
        if price_precision.is_none() {
            trade.price.precision = current_price_precision;
        }

        if size_precision.is_none() {
            trade.size.precision = current_size_precision;
        }
    }
}

/// Loads [`OrderBookDelta`]s from a Tardis format CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
/// Load order book delta records from a CSV or gzipped CSV file.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn load_deltas<P: AsRef<Path>>(
    filepath: P,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<OrderBookDelta>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut deltas: Vec<OrderBookDelta> = Vec::with_capacity(estimated_capacity);

    let mut current_price_precision = price_precision.unwrap_or(0);
    let mut current_size_precision = size_precision.unwrap_or(0);
    let mut last_ts_init: Option<UnixNanos> = None;
    let mut last_is_snapshot = false;

    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        if let Some(limit) = limit
            && deltas.len() >= limit
        {
            break;
        }

        let data: TardisBookUpdateRecord = record.deserialize(None)?;

        update_precision_if_needed(&mut current_price_precision, data.price, price_precision);
        update_precision_if_needed(&mut current_size_precision, data.amount, size_precision);

        let ts_event = parse_timestamp(data.timestamp);
        let ts_init = parse_timestamp(data.local_timestamp);

        // Insert CLEAR on snapshot boundary to reset order book state.
        // Some venues emit every book event as a full snapshot, so a new
        // snapshot message must also reset the previous snapshot state.
        let starts_new_snapshot =
            data.is_snapshot && (!last_is_snapshot || last_ts_init != Some(ts_init));

        if starts_new_snapshot {
            let clear_instrument_id =
                instrument_id.unwrap_or_else(|| parse_instrument_id(&data.exchange, data.symbol));

            if last_ts_init != Some(ts_init)
                && let Some(last_delta) = deltas.last_mut()
            {
                last_delta.flags = RecordFlag::F_LAST as u8;
            }
            last_ts_init = Some(ts_init);

            let clear_delta = OrderBookDelta::clear(clear_instrument_id, 0, ts_event, ts_init);
            deltas.push(clear_delta);

            if let Some(limit) = limit
                && deltas.len() >= limit
            {
                break;
            }
        }
        last_is_snapshot = data.is_snapshot;

        let delta = match parse_delta_record(
            &data,
            current_price_precision,
            current_size_precision,
            instrument_id,
        ) {
            Ok(d) => d,
            Err(e) => {
                log::warn!("Skipping invalid delta record: {e}");
                continue;
            }
        };

        let ts_init = delta.ts_init;
        if last_ts_init != Some(ts_init)
            && let Some(last_delta) = deltas.last_mut()
        {
            last_delta.flags = RecordFlag::F_LAST as u8;
        }

        last_ts_init = Some(ts_init);

        deltas.push(delta);
    }

    // Set F_LAST flag for final delta
    if let Some(last_delta) = deltas.last_mut() {
        last_delta.flags = RecordFlag::F_LAST as u8;
    }

    // Update all deltas to use the final (maximum) precision discovered
    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
    update_deltas_precision(
        &mut deltas,
        price_precision,
        size_precision,
        current_price_precision,
        current_size_precision,
    );

    Ok(deltas)
}

/// Loads [`OrderBookDepth`]s from a Tardis format CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
/// Load order book depth snapshots (5-level) from a CSV or gzipped CSV file.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
///
/// # Panics
///
/// Panics if a record level cannot be parsed to depth.
pub fn load_depth_from_snapshot5<P: AsRef<Path>>(
    filepath: P,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<OrderBookDepth>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut depths: Vec<OrderBookDepth> = Vec::with_capacity(estimated_capacity);

    let mut current_price_precision = price_precision.unwrap_or(0);
    let mut current_size_precision = size_precision.unwrap_or(0);

    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        let data: TardisOrderBookSnapshot5Record = record.deserialize(None)?;

        // Update precisions dynamically if not explicitly set
        let mut precision_updated = false;

        if price_precision.is_none()
            && let Some(bid_price) = data.bids_0_price
        {
            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
            if inferred_price_precision > current_price_precision {
                current_price_precision = inferred_price_precision;
                precision_updated = true;
            }
        }

        if size_precision.is_none()
            && let Some(bid_amount) = data.bids_0_amount
        {
            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
            if inferred_size_precision > current_size_precision {
                current_size_precision = inferred_size_precision;
                precision_updated = true;
            }
        }

        // If precision increased, update all previous depths
        if precision_updated {
            for depth in &mut depths {
                for order in depth.bids.iter_mut().chain(depth.asks.iter_mut()) {
                    if price_precision.is_none() {
                        order.price.precision = current_price_precision;
                    }

                    if size_precision.is_none() {
                        order.size.precision = current_size_precision;
                    }
                }
            }
        }

        let instrument_id = match &instrument_id {
            Some(id) => *id,
            None => parse_instrument_id(&data.exchange, data.symbol),
        };
        // Mark as both snapshot and last (consistent with streaming implementation)
        let flags = RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8;
        let sequence = 0; // Sequence not available
        let ts_event = parse_timestamp(data.timestamp);
        let ts_init = parse_timestamp(data.local_timestamp);

        // Initialize empty arrays
        let mut bids = [NULL_ORDER; DEPTH10_LEN];
        let mut asks = [NULL_ORDER; DEPTH10_LEN];
        let mut bid_counts = [0u32; DEPTH10_LEN];
        let mut ask_counts = [0u32; DEPTH10_LEN];

        for i in 0..=4 {
            // Create bids
            let (bid_order, bid_count) = create_book_order(
                OrderSide::Buy,
                match i {
                    0 => data.bids_0_price,
                    1 => data.bids_1_price,
                    2 => data.bids_2_price,
                    3 => data.bids_3_price,
                    4 => data.bids_4_price,
                    _ => unreachable!("i is constrained to 0..=4 by loop"),
                },
                match i {
                    0 => data.bids_0_amount,
                    1 => data.bids_1_amount,
                    2 => data.bids_2_amount,
                    3 => data.bids_3_amount,
                    4 => data.bids_4_amount,
                    _ => unreachable!("i is constrained to 0..=4 by loop"),
                },
                current_price_precision,
                current_size_precision,
            );
            bids[i] = bid_order;
            bid_counts[i] = bid_count;

            // Create asks
            let (ask_order, ask_count) = create_book_order(
                OrderSide::Sell,
                match i {
                    0 => data.asks_0_price,
                    1 => data.asks_1_price,
                    2 => data.asks_2_price,
                    3 => data.asks_3_price,
                    4 => data.asks_4_price,
                    _ => None, // Unreachable, but for safety
                },
                match i {
                    0 => data.asks_0_amount,
                    1 => data.asks_1_amount,
                    2 => data.asks_2_amount,
                    3 => data.asks_3_amount,
                    4 => data.asks_4_amount,
                    _ => None, // Unreachable, but for safety
                },
                current_price_precision,
                current_size_precision,
            );
            asks[i] = ask_order;
            ask_counts[i] = ask_count;
        }

        let depth = OrderBookDepth::new(
            instrument_id,
            bids,
            asks,
            bid_counts,
            ask_counts,
            flags,
            sequence,
            ts_event,
            ts_init,
        );

        depths.push(depth);

        if let Some(limit) = limit
            && depths.len() >= limit
        {
            break;
        }
    }

    Ok(depths)
}

/// Loads [`OrderBookDepth`]s from a Tardis format CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
/// Load order book depth snapshots (25-level) from a CSV or gzipped CSV file.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn load_depth_from_snapshot25<P: AsRef<Path>>(
    filepath: P,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<OrderBookDepth>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut depths: Vec<OrderBookDepth> = Vec::with_capacity(estimated_capacity);

    let mut current_price_precision = price_precision.unwrap_or(0);
    let mut current_size_precision = size_precision.unwrap_or(0);
    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        let data: TardisOrderBookSnapshot25Record = record.deserialize(None)?;

        // Update precisions dynamically if not explicitly set
        let mut precision_updated = false;

        if price_precision.is_none()
            && let Some(bid_price) = data.bids_0_price
        {
            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
            if inferred_price_precision > current_price_precision {
                current_price_precision = inferred_price_precision;
                precision_updated = true;
            }
        }

        if size_precision.is_none()
            && let Some(bid_amount) = data.bids_0_amount
        {
            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
            if inferred_size_precision > current_size_precision {
                current_size_precision = inferred_size_precision;
                precision_updated = true;
            }
        }

        // If precision increased, update all previous depths
        if precision_updated {
            for depth in &mut depths {
                for order in depth.bids.iter_mut().chain(depth.asks.iter_mut()) {
                    if price_precision.is_none() {
                        order.price.precision = current_price_precision;
                    }

                    if size_precision.is_none() {
                        order.size.precision = current_size_precision;
                    }
                }
            }
        }

        let instrument_id = match &instrument_id {
            Some(id) => *id,
            None => parse_instrument_id(&data.exchange, data.symbol),
        };
        // Mark as both snapshot and last (consistent with streaming implementation)
        let flags = RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8;
        let sequence = 0; // Sequence not available
        let ts_event = parse_timestamp(data.timestamp);
        let ts_init = parse_timestamp(data.local_timestamp);

        // Initialize empty arrays for all 25 levels
        let mut bids = [NULL_ORDER; TardisOrderBookSnapshot25Record::LEVELS];
        let mut asks = [NULL_ORDER; TardisOrderBookSnapshot25Record::LEVELS];
        let mut bid_counts = [0u32; TardisOrderBookSnapshot25Record::LEVELS];
        let mut ask_counts = [0u32; TardisOrderBookSnapshot25Record::LEVELS];

        // Fill all 25 levels from the 25-level record
        for i in 0..TardisOrderBookSnapshot25Record::LEVELS {
            // Create bids
            let (bid_price, bid_amount) = data.bid_level(i);
            let (bid_order, bid_count) = create_book_order(
                OrderSide::Buy,
                bid_price,
                bid_amount,
                current_price_precision,
                current_size_precision,
            );
            bids[i] = bid_order;
            bid_counts[i] = bid_count;

            // Create asks
            let (ask_price, ask_amount) = data.ask_level(i);
            let (ask_order, ask_count) = create_book_order(
                OrderSide::Sell,
                ask_price,
                ask_amount,
                current_price_precision,
                current_size_precision,
            );
            asks[i] = ask_order;
            ask_counts[i] = ask_count;
        }

        let depth = OrderBookDepth::new(
            instrument_id,
            bids,
            asks,
            bid_counts,
            ask_counts,
            flags,
            sequence,
            ts_event,
            ts_init,
        );

        depths.push(depth);

        if let Some(limit) = limit
            && depths.len() >= limit
        {
            break;
        }
    }

    Ok(depths)
}

/// Loads [`QuoteTick`]s from a Tardis format CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
/// Load quote ticks from a CSV or gzipped CSV file.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn load_quotes<P: AsRef<Path>>(
    filepath: P,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<QuoteTick>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut quotes: Vec<QuoteTick> = Vec::with_capacity(estimated_capacity);

    let mut current_price_precision = price_precision.unwrap_or(0);
    let mut current_size_precision = size_precision.unwrap_or(0);
    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        let data: TardisQuoteRecord = record.deserialize(None)?;

        if price_precision.is_none()
            && let Some(bid_price) = data.bid_price
        {
            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
            if inferred_price_precision > current_price_precision {
                current_price_precision = inferred_price_precision;
            }
        }

        if size_precision.is_none()
            && let Some(bid_amount) = data.bid_amount
        {
            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
            if inferred_size_precision > current_size_precision {
                current_size_precision = inferred_size_precision;
            }
        }

        let quote = parse_quote_record(
            &data,
            current_price_precision,
            current_size_precision,
            instrument_id,
        );

        quotes.push(quote);

        if let Some(limit) = limit
            && quotes.len() >= limit
        {
            break;
        }
    }

    // Update all quotes to use the final (maximum) precision discovered
    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
    update_quotes_precision(
        &mut quotes,
        price_precision,
        size_precision,
        current_price_precision,
        current_size_precision,
    );

    Ok(quotes)
}

/// Loads [`TradeTick`]s from a Tardis format CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
/// Load trade ticks from a CSV or gzipped CSV file.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn load_trades<P: AsRef<Path>>(
    filepath: P,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<TradeTick>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut trades: Vec<TradeTick> = Vec::with_capacity(estimated_capacity);

    let mut current_price_precision = price_precision.unwrap_or(0);
    let mut current_size_precision = size_precision.unwrap_or(0);
    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        let data: TardisTradeRecord = record.deserialize(None)?;

        if price_precision.is_none() {
            let inferred_price_precision = infer_precision(data.price).min(FIXED_PRECISION);
            if inferred_price_precision > current_price_precision {
                current_price_precision = inferred_price_precision;
            }
        }

        if size_precision.is_none() {
            let inferred_size_precision = infer_precision(data.amount).min(FIXED_PRECISION);
            if inferred_size_precision > current_size_precision {
                current_size_precision = inferred_size_precision;
            }
        }

        let size = Quantity::new_checked(data.amount, current_size_precision)?;

        if size.is_positive() {
            let trade = parse_trade_record(&data, size, current_price_precision, instrument_id);

            trades.push(trade);

            if let Some(limit) = limit
                && trades.len() >= limit
            {
                break;
            }
        } else {
            log::warn!("Skipping zero-sized trade: {data:?}");
        }
    }

    // Update all trades to use the final (maximum) precision discovered
    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
    update_trades_precision(
        &mut trades,
        price_precision,
        size_precision,
        current_price_precision,
        current_size_precision,
    );

    Ok(trades)
}

/// Loads [`FundingRateUpdate`]s from a Tardis format derivative ticker CSV at the given `filepath`,
/// automatically applying `GZip` decompression for files ending in ".gz".
///
/// This function parses the `funding_rate` and `funding_timestamp` fields from derivative ticker
/// data to create funding rate updates.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn load_funding_rates<P: AsRef<Path>>(
    filepath: P,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> Result<Vec<FundingRateUpdate>, Box<dyn Error>> {
    // Estimate capacity for Vec pre-allocation
    let estimated_capacity = limit.unwrap_or(100_000).min(1_000_000);
    let mut funding_rates: Vec<FundingRateUpdate> = Vec::with_capacity(estimated_capacity);

    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        let data: TardisDerivativeTickerRecord = record.deserialize(None)?;

        // Parse to funding rate update (returns None if no funding data)
        if let Some(funding_rate) = parse_derivative_ticker_record(&data, instrument_id) {
            funding_rates.push(funding_rate);

            if let Some(limit) = limit
                && funding_rates.len() >= limit
            {
                break;
            }
        }
    }

    Ok(funding_rates)
}

/// Loads option chain rows from a Tardis `options_chain` CSV file.
///
/// Returns quote ticks before option greeks for rows with a complete best bid/offer. Rows missing
/// any best bid/offer field still return option greeks.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV, or if a complete best
/// bid/offer row contains invalid price or size values.
pub fn load_options_chain<P: AsRef<Path>>(
    filepath: P,
    underlyings: Option<Vec<String>>,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    limit: Option<usize>,
) -> Result<Vec<Data>, Box<dyn Error>> {
    let underlyings = normalize_underlying_filters(underlyings);
    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
    let mut records: Vec<TardisOptionsChainRecord> = Vec::with_capacity(estimated_capacity);
    let mut precision_by_instrument: AHashMap<InstrumentId, OptionsChainPrecision> =
        AHashMap::new();

    let mut reader = create_csv_reader(filepath)?;
    let mut record = StringRecord::new();

    while reader.read_record(&mut record)? {
        if let Some(underlyings) = underlyings.as_deref() {
            let Some(symbol) = record.get(1) else {
                continue;
            };
            let symbol = symbol.to_uppercase();
            if !matches_underlying_filter(&symbol, Some(underlyings)) {
                continue;
            }
        }

        let data: TardisOptionsChainRecord = record.deserialize(None)?;
        let instrument_id = parse_instrument_id(&data.exchange, data.symbol);
        precision_by_instrument
            .entry(instrument_id)
            .or_insert_with(|| OptionsChainPrecision::new(price_precision, size_precision))
            .update(&data, price_precision, size_precision);
        records.push(data);

        if let Some(limit) = limit
            && records.len() >= limit
        {
            break;
        }
    }

    let mut output = Vec::with_capacity(records.len() * 2);
    for record in records {
        let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
        let precision = precision_by_instrument
            .get(&instrument_id)
            .copied()
            .unwrap_or_else(|| OptionsChainPrecision::new(price_precision, size_precision));

        if let Some(quote) = parse_options_chain_record_as_quote(
            &record,
            precision.price,
            precision.size,
            instrument_id,
        )? {
            output.push(Data::Quote(quote));
        }

        output.push(Data::OptionGreeks(parse_options_chain_record(
            &record,
            instrument_id,
        )));
    }

    Ok(output)
}

#[cfg(test)]
mod tests {
    use std::{fs, fs::File, sync::Arc};

    use nautilus_core::paths::get_test_data_path as get_test_data_root;
    use nautilus_model::{
        enums::{AggressorSide, BookAction, OrderSide},
        identifiers::{InstrumentId, TradeId},
        types::Price,
    };
    use nautilus_serialization::arrow::{ArrowSchemaProvider, EncodeToRecordBatch};
    use nautilus_testkit::common::{
        get_tardis_binance_snapshot5_path, get_tardis_binance_snapshot25_path,
        get_tardis_bitmex_trades_path, get_tardis_deribit_book_l2_path,
        get_tardis_huobi_quotes_path,
    };
    use parquet::{arrow::ArrowWriter, file::properties::WriterProperties};
    use rstest::*;
    use rust_decimal_macros::dec;

    use super::*;
    use crate::common::{parse::parse_price, testing::get_test_data_path};

    #[rstest]
    #[case(0.0, 0)]
    #[case(42.0, 0)]
    #[case(0.1, 1)]
    #[case(0.25, 2)]
    #[case(123.0001, 4)]
    #[case(-42.987654321,       9)]
    #[case(1.234_567_890_123, 12)]
    fn test_infer_precision(#[case] input: f64, #[case] expected: u8) {
        assert_eq!(infer_precision(input), expected);
    }

    #[rstest]
    pub fn test_dynamic_precision_inference() {
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50000.0,1.0
binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.5,2.0
binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50000.12,1.5
binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49999.123,3.0
binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50000.1234,0.5";

        let temp_file = std::env::temp_dir().join("test_dynamic_precision.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let deltas = load_deltas(&temp_file, None, None, None, None).unwrap();

        // 5 data rows + 1 CLEAR delta at start (first row is snapshot)
        assert_eq!(deltas.len(), 6);

        // Skip the CLEAR delta at index 0
        for (i, delta) in deltas.iter().skip(1).enumerate() {
            assert_eq!(
                delta.order.price.precision, 4,
                "Price precision should be 4 for delta {i}",
            );
            assert_eq!(
                delta.order.size.precision, 1,
                "Size precision should be 1 for delta {i}",
            );
        }

        // Test exact values to ensure retroactive precision updates work correctly
        // Index 0 is CLEAR, data starts at index 1
        assert_eq!(deltas[0].action, BookAction::Clear);

        assert_eq!(deltas[1].order.price, parse_price(50000.0, 4));
        assert_eq!(deltas[1].order.size, Quantity::new(1.0, 1));

        assert_eq!(deltas[2].order.price, parse_price(49999.5, 4));
        assert_eq!(deltas[2].order.size, Quantity::new(2.0, 1));

        assert_eq!(deltas[3].order.price, parse_price(50000.12, 4));
        assert_eq!(deltas[3].order.size, Quantity::new(1.5, 1));

        assert_eq!(deltas[4].order.price, parse_price(49999.123, 4));
        assert_eq!(deltas[4].order.size, Quantity::new(3.0, 1));

        assert_eq!(deltas[5].order.price, parse_price(50000.1234, 4));
        assert_eq!(deltas[5].order.size, Quantity::new(0.5, 1));

        assert_eq!(
            deltas[1].order.price.precision,
            deltas[5].order.price.precision
        );
        assert_eq!(
            deltas[1].order.size.precision,
            deltas[3].order.size.precision
        );

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    #[case(Some(1), Some(0))] // Explicit precisions
    #[case(None, None)] // Inferred precisions
    pub fn test_read_deltas(
        #[case] price_precision: Option<u8>,
        #[case] size_precision: Option<u8>,
    ) {
        let filepath = get_tardis_deribit_book_l2_path();
        let deltas =
            load_deltas(filepath, price_precision, size_precision, None, Some(100)).unwrap();

        // 15 data rows + 1 CLEAR delta at start (first row is snapshot)
        assert_eq!(deltas.len(), 16);

        // Index 0 is CLEAR delta
        assert_eq!(deltas[0].action, BookAction::Clear);

        // Index 1 is first data delta
        assert_eq!(
            deltas[1].instrument_id,
            InstrumentId::from("BTC-PERPETUAL.DERIBIT")
        );
        assert_eq!(deltas[1].action, BookAction::Add);
        assert_eq!(deltas[1].order.side, OrderSide::Sell.into());
        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
        assert_eq!(deltas[1].order.size, Quantity::from("18640"));
        assert_eq!(deltas[1].flags, 0);
        assert_eq!(deltas[1].sequence, 0);
        assert_eq!(deltas[1].ts_event, 1585699200245000000);
        assert_eq!(deltas[1].ts_init, 1585699200355684000);
    }

    #[rstest]
    #[case(Some(2), Some(3))] // Explicit precisions
    #[case(None, None)] // Inferred precisions
    pub fn test_read_depths_from_snapshot5(
        #[case] price_precision: Option<u8>,
        #[case] size_precision: Option<u8>,
    ) {
        let filepath = get_tardis_binance_snapshot5_path();
        let depths =
            load_depth_from_snapshot5(filepath, price_precision, size_precision, None, Some(100))
                .unwrap();

        assert_eq!(depths.len(), 10);
        assert_eq!(
            depths[0].instrument_id,
            InstrumentId::from("BTCUSDT.BINANCE")
        );
        assert_eq!(depths[0].bids.len(), 5);
        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
        assert_eq!(depths[0].bids[0].side, OrderSide::Buy.into());
        assert_eq!(depths[0].bids[0].order_id, 0);
        assert_eq!(depths[0].asks.len(), 5);
        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
        assert_eq!(depths[0].asks[0].side, OrderSide::Sell.into());
        assert_eq!(depths[0].asks[0].order_id, 0);
        assert_eq!(depths[0].bid_counts.as_slice(), &[1; 5]);
        assert_eq!(depths[0].ask_counts.as_slice(), &[1; 5]);
        // F_SNAPSHOT (32) | F_LAST (128) = 160
        assert_eq!(
            depths[0].flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(depths[0].ts_event, 1598918403696000000);
        assert_eq!(depths[0].ts_init, 1598918403810979000);
        assert_eq!(depths[0].sequence, 0);
    }

    #[rstest]
    #[case(Some(2), Some(3))] // Explicit precisions
    #[case(None, None)] // Inferred precisions
    pub fn test_read_depths_from_snapshot25(
        #[case] price_precision: Option<u8>,
        #[case] size_precision: Option<u8>,
    ) {
        let filepath = get_tardis_binance_snapshot25_path();
        let depths =
            load_depth_from_snapshot25(filepath, price_precision, size_precision, None, Some(100))
                .unwrap();

        assert_eq!(depths.len(), 10);
        assert_eq!(
            depths[0].instrument_id,
            InstrumentId::from("BTCUSDT.BINANCE")
        );
        assert_eq!(depths[0].bids.len(), 25);
        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
        assert_eq!(depths[0].bids[0].side, OrderSide::Buy.into());
        assert_eq!(depths[0].bids[0].order_id, 0);
        assert_eq!(depths[0].asks.len(), 25);
        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
        assert_eq!(depths[0].asks[0].side, OrderSide::Sell.into());
        assert_eq!(depths[0].asks[0].order_id, 0);
        assert_eq!(depths[0].bid_counts[0], 1);
        assert_eq!(depths[0].ask_counts[0], 1);
        // F_SNAPSHOT (32) | F_LAST (128) = 160
        assert_eq!(
            depths[0].flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(depths[0].ts_event, 1598918403696000000);
        assert_eq!(depths[0].ts_init, 1598918403810979000);
        assert_eq!(depths[0].sequence, 0);
    }

    #[rstest]
    #[case(Some(1), Some(0))] // Explicit precisions
    #[case(None, None)] // Inferred precisions
    pub fn test_read_quotes(
        #[case] price_precision: Option<u8>,
        #[case] size_precision: Option<u8>,
    ) {
        let filepath = get_tardis_huobi_quotes_path();
        let quotes =
            load_quotes(filepath, price_precision, size_precision, None, Some(100)).unwrap();

        assert_eq!(quotes.len(), 10);
        assert_eq!(
            quotes[0].instrument_id,
            InstrumentId::from("BTC-USD.HUOBI_DELIVERY")
        );
        assert_eq!(quotes[0].bid_price, Price::from("8629.2"));
        assert_eq!(quotes[0].bid_size, Quantity::from("806"));
        assert_eq!(quotes[0].ask_price, Price::from("8629.3"));
        assert_eq!(quotes[0].ask_size, Quantity::from("5494"));
        assert_eq!(quotes[0].ts_event, 1588291201099000000);
        assert_eq!(quotes[0].ts_init, 1588291201234268000);
    }

    #[rstest]
    fn test_load_options_chain_filters_underlying_and_emits_quote_then_greeks() {
        let filepath = get_test_data_path("options_chain.csv");
        let data =
            load_options_chain(filepath, Some(vec!["btc-".to_string()]), None, None, None).unwrap();

        assert_eq!(data.len(), 9);

        let Data::Quote(quote) = &data[0] else {
            panic!("Expected first data item to be Quote");
        };
        let Data::OptionGreeks(greeks) = &data[1] else {
            panic!("Expected second data item to be OptionGreeks");
        };

        assert_eq!(
            quote.instrument_id,
            InstrumentId::from("BTC-9JUN20-9875-P.DERIBIT")
        );
        assert_eq!(quote.bid_price, Price::from("0.0205"));
        assert_eq!(quote.ask_price, Price::from("0.0235"));
        assert_eq!(quote.bid_size, Quantity::from("15.1"));
        assert_eq!(quote.ask_size, Quantity::from("15.2"));
        assert_eq!(quote.bid_price.precision, 4);
        assert_eq!(quote.bid_size.precision, 1);

        assert_eq!(greeks.instrument_id, quote.instrument_id);
        assert_eq!(greeks.greeks.delta, -0.61752);
        assert_eq!(greeks.mark_iv, Some(62.89));
        assert_eq!(greeks.underlying_price, Some(9756.36));
    }

    #[rstest]
    fn test_load_options_chain_missing_bbo_emits_greeks_only_with_default_greeks() {
        let filepath = get_test_data_path("options_chain.csv");
        let data = load_options_chain(
            filepath,
            Some(vec!["BTC-10JUN20".to_string()]),
            None,
            None,
            None,
        )
        .unwrap();

        assert_eq!(data.len(), 1);

        let Data::OptionGreeks(greeks) = &data[0] else {
            panic!("Expected OptionGreeks, was {:?}", data[0]);
        };

        assert_eq!(
            greeks.instrument_id,
            InstrumentId::from("BTC-10JUN20-10000-C.DERIBIT")
        );
        assert_eq!(greeks.open_interest, None);
        assert_eq!(greeks.bid_iv, None);
        assert_eq!(greeks.ask_iv, None);
        assert_eq!(greeks.greeks.delta, 0.0);
        assert_eq!(greeks.greeks.gamma, 0.0);
        assert_eq!(greeks.greeks.vega, 0.0);
        assert_eq!(greeks.greeks.theta, 0.0);
        assert_eq!(greeks.greeks.rho, 0.0);
    }

    #[rstest]
    fn test_load_options_chain_rejects_zero_bbo_size() {
        let temp_file = tempfile::NamedTempFile::new().unwrap();
        let csv_data = "exchange,symbol,timestamp,local_timestamp,type,strike_price,expiration,open_interest,last_price,bid_price,bid_amount,bid_iv,ask_price,ask_amount,ask_iv,mark_price,mark_iv,underlying_index,underlying_price,delta,gamma,vega,theta,rho
deribit,BTC-9JUN20-9875-P,1591574399413000,1591574400196008,put,9875,1591689600000000,0.1,0.0295,0.0205,0,55.91,0.0235,15.2,68.94,0.02210436,62.89,SYN.BTC-9JUN20,9756.36,-0.61752,0.00103,2.24964,-53.05655,-0.22796";
        fs::write(temp_file.path(), csv_data).unwrap();

        let error = load_options_chain(temp_file.path(), None, None, None, None).unwrap_err();

        assert_eq!(error.to_string(), "value was zero");
    }

    #[rstest]
    fn test_load_options_chain_infers_precision_per_instrument() {
        let filepath = get_test_data_path("options_chain.csv");
        let data =
            load_options_chain(filepath, Some(vec!["ETH-".to_string()]), None, None, None).unwrap();

        assert_eq!(data.len(), 3);

        let Data::Quote(quote) = &data[0] else {
            panic!("Expected first data item to be Quote");
        };

        assert_eq!(
            quote.instrument_id,
            InstrumentId::from("ETH-9JUN20-250-P.DERIBIT")
        );
        assert_eq!(quote.bid_price, Price::from("0.12345"));
        assert_eq!(quote.ask_price, Price::from("0.12456"));
        assert_eq!(quote.bid_size, Quantity::from("0.123456"));
        assert_eq!(quote.ask_size, Quantity::from("0.223456"));
        assert_eq!(quote.bid_price.precision, 5);
        assert_eq!(quote.bid_size.precision, 6);

        assert!(matches!(data[1], Data::OptionGreeks(_)));
        assert!(matches!(data[2], Data::OptionGreeks(_)));
    }

    #[rstest]
    #[case(Some(1), Some(0))] // Explicit precisions
    #[case(None, None)] // Inferred precisions
    pub fn test_read_trades(
        #[case] price_precision: Option<u8>,
        #[case] size_precision: Option<u8>,
    ) {
        let filepath = get_tardis_bitmex_trades_path();
        let trades =
            load_trades(filepath, price_precision, size_precision, None, Some(100)).unwrap();

        assert_eq!(trades.len(), 10);
        assert_eq!(trades[0].instrument_id, InstrumentId::from("XBTUSD.BITMEX"));
        assert_eq!(trades[0].price, Price::from("8531.5"));
        assert_eq!(trades[0].size, Quantity::from("2152"));
        assert_eq!(trades[0].aggressor_side, AggressorSide::Sell);
        assert_eq!(
            trades[0].trade_id,
            TradeId::new("ccc3c1fa-212c-e8b0-1706-9b9c4f3d5ecf")
        );
        assert_eq!(trades[0].ts_event, 1583020803145000000);
        assert_eq!(trades[0].ts_init, 1583020803307160000);
    }

    #[rstest]
    pub fn test_load_trades_derives_id_when_csv_id_empty() {
        // Two rows with empty `id` column must both hash deterministically
        // to the same TradeId, and a row with differing price must hash differently.
        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50001.0,1.0";

        let temp_file = std::env::temp_dir().join("test_load_trades_empty_id.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let trades = load_trades(&temp_file, Some(2), Some(1), None, None).unwrap();
        assert_eq!(trades.len(), 3);

        assert_eq!(trades[0].trade_id, trades[1].trade_id);
        assert_eq!(trades[0].trade_id.as_str().len(), 16);
        assert_ne!(trades[0].trade_id, trades[2].trade_id);

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    pub fn test_load_trades_with_zero_sized_trade() {
        // Create test CSV data with one zero-sized trade that should be skipped
        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,trade1,buy,50000.0,1.0
binance,BTCUSDT,1640995201000000,1640995201100000,trade2,sell,49999.5,0.0
binance,BTCUSDT,1640995202000000,1640995202100000,trade3,buy,50000.12,1.5
binance,BTCUSDT,1640995203000000,1640995203100000,trade4,sell,49999.123,3.0";

        let temp_file = std::env::temp_dir().join("test_load_trades_zero_size.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let trades = load_trades(
            &temp_file,
            Some(4),
            Some(1),
            None,
            None, // No limit, load all
        )
        .unwrap();

        // Should have 3 trades (zero-sized trade skipped)
        assert_eq!(trades.len(), 3);

        // Verify the correct trades were loaded (not the zero-sized one)
        assert_eq!(trades[0].size, Quantity::from("1.0"));
        assert_eq!(trades[1].size, Quantity::from("1.5"));
        assert_eq!(trades[2].size, Quantity::from("3.0"));

        // Verify trade IDs to confirm correct trades were loaded
        assert_eq!(trades[0].trade_id, TradeId::new("trade1"));
        assert_eq!(trades[1].trade_id, TradeId::new("trade3"));
        assert_eq!(trades[2].trade_id, TradeId::new("trade4"));

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    pub fn test_load_trades_from_local_file() {
        let filepath = get_test_data_path("csv/trades_1.csv");
        let trades = load_trades(filepath, Some(1), Some(0), None, None).unwrap();
        assert_eq!(trades.len(), 2);
        assert_eq!(trades[0].price, Price::from("8531.5"));
        assert_eq!(trades[1].size, Quantity::from("1000"));
    }

    #[rstest]
    pub fn test_load_deltas_from_local_file() {
        let filepath = get_test_data_path("csv/deltas_1.csv");
        let deltas = load_deltas(filepath, Some(1), Some(0), None, None).unwrap();

        // 2 data rows + 1 CLEAR delta at start (first row is snapshot)
        assert_eq!(deltas.len(), 3);
        assert_eq!(deltas[0].action, BookAction::Clear);
        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
        assert_eq!(deltas[2].order.size, Quantity::from("10000"));
    }

    #[rstest]
    fn test_load_deltas_groups_messages_by_local_timestamp() {
        let filepath = get_test_data_path("csv/deltas_message_boundaries.csv");
        let deltas = load_deltas(filepath, Some(1), Some(1), None, None).unwrap();

        assert_eq!(deltas.len(), 4);
        assert_eq!(
            deltas.iter().map(|delta| delta.flags).collect::<Vec<_>>(),
            vec![0, RecordFlag::F_LAST as u8, 0, RecordFlag::F_LAST as u8]
        );
        assert_eq!(
            deltas
                .iter()
                .map(|delta| delta.ts_event)
                .collect::<Vec<_>>(),
            vec![
                UnixNanos::from(1_000_000),
                UnixNanos::from(1_000_000),
                UnixNanos::from(1_000_000),
                UnixNanos::from(1_010_000),
            ]
        );
        assert_eq!(
            deltas.iter().map(|delta| delta.ts_init).collect::<Vec<_>>(),
            vec![
                UnixNanos::from(2_000_000),
                UnixNanos::from(2_000_000),
                UnixNanos::from(2_010_000),
                UnixNanos::from(2_010_000),
            ]
        );
        assert_eq!(deltas[0].order.side, Some(OrderSide::Buy));
        assert_eq!(deltas[0].order.price, Price::from("100.0"));
        assert_eq!(deltas[0].order.size, Quantity::from("1.0"));
        assert_eq!(deltas[1].order.side, Some(OrderSide::Sell));
        assert_eq!(deltas[1].order.price, Price::from("101.0"));
        assert_eq!(deltas[1].order.size, Quantity::from("2.0"));
        assert_eq!(deltas[2].order.side, Some(OrderSide::Buy));
        assert_eq!(deltas[2].order.price, Price::from("99.0"));
        assert_eq!(deltas[2].order.size, Quantity::from("3.0"));
        assert_eq!(deltas[3].order.side, Some(OrderSide::Sell));
        assert_eq!(deltas[3].order.price, Price::from("102.0"));
        assert_eq!(deltas[3].order.size, Quantity::from("4.0"));
    }

    #[rstest]
    fn test_load_funding_rates_okex_xperp() {
        let filepath = get_test_data_path("csv/okex_futures_xperp_derivative_ticker.csv");
        let funding_rates = load_funding_rates(filepath, None, None).unwrap();

        let instrument_id = InstrumentId::from("BTC-USD_UM_XPERP-310404.OKEX");

        assert_eq!(funding_rates.len(), 8);
        assert!(
            funding_rates
                .iter()
                .all(|f| f.instrument_id == instrument_id)
        );

        // OKX X-Perps publish no predicted rate, so the funding timestamp is the only forward
        // reference Tardis carries, and the interval is not representable in a derivative ticker
        assert!(
            funding_rates
                .iter()
                .all(|f| f.next_funding_ns.is_some() && f.interval.is_none())
        );

        let first = &funding_rates[0];
        let rolled = &funding_rates[3];

        assert_eq!(first.rate, dec!(-0.0003972900658902));
        assert_eq!(
            first.next_funding_ns,
            Some(UnixNanos::from(1_786_320_000_000_000_000))
        );
        assert_eq!(first.ts_event, UnixNanos::from(1_786_320_006_952_000_000));
        assert_eq!(first.ts_init, UnixNanos::from(1_786_320_006_971_532_000));

        assert_eq!(rolled.rate, dec!(-0.0003962534258591));
        assert_eq!(
            rolled.next_funding_ns,
            Some(UnixNanos::from(1_786_348_800_000_000_000))
        );
        assert_eq!(rolled.ts_event, UnixNanos::from(1_786_320_007_369_000_000));
        assert_eq!(rolled.ts_init, UnixNanos::from(1_786_320_007_402_423_000));
    }

    #[rstest]
    fn test_load_funding_rates_without_funding_timestamp() {
        let filepath = get_test_data_path("csv/deribit_derivative_ticker.csv");
        let funding_rates = load_funding_rates(filepath, None, None).unwrap();

        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");

        assert_eq!(funding_rates.len(), 3);
        assert!(
            funding_rates
                .iter()
                .all(|f| f.instrument_id == instrument_id)
        );

        // Deribit publishes no funding timestamp, so there is no forward reference to carry
        assert!(
            funding_rates
                .iter()
                .all(|f| f.next_funding_ns.is_none() && f.interval.is_none())
        );

        let first = &funding_rates[0];
        let changed = &funding_rates[2];

        assert_eq!(first.rate, dec!(0.00000459));
        assert_eq!(first.ts_event, UnixNanos::from(1_786_320_665_523_000_000));
        assert_eq!(first.ts_init, UnixNanos::from(1_786_320_665_533_324_000));

        assert_eq!(changed.rate, dec!(0.00000452));
        assert_eq!(changed.ts_event, UnixNanos::from(1_786_320_665_645_000_000));
        assert_eq!(changed.ts_init, UnixNanos::from(1_786_320_665_661_106_000));
    }

    #[rstest]
    fn test_load_funding_rates_okex_usdc_across_index_migration() {
        let filepath = get_test_data_path("csv/okex_swap_usdc_index_migration.csv");
        let funding_rates = load_funding_rates(filepath, None, None).unwrap();

        let instrument_id = InstrumentId::from("BTC-USDC-SWAP.OKEX");

        assert_eq!(funding_rates.len(), 6);

        // Tardis remaps USDC-margined contracts to the USDC index after 2023-04-10T08:40Z, which
        // changes the index price feed but never the contract symbol, so replaying across the
        // migration must resolve to one instrument
        assert!(
            funding_rates
                .iter()
                .all(|f| f.instrument_id == instrument_id)
        );

        let pre = &funding_rates[0];
        let post = &funding_rates[3];

        assert_eq!(pre.rate, dec!(0.0001035718476117));
        assert_eq!(
            pre.next_funding_ns,
            Some(UnixNanos::from(1_680_336_000_000_000_000))
        );
        assert_eq!(pre.ts_event, UnixNanos::from(1_680_309_048_427_000_000));
        assert_eq!(pre.ts_init, UnixNanos::from(1_680_309_048_450_728_000));

        assert_eq!(post.rate, dec!(-0.000055804472025));
        assert_eq!(
            post.next_funding_ns,
            Some(UnixNanos::from(1_682_928_000_000_000_000))
        );
        assert_eq!(post.ts_event, UnixNanos::from(1_682_900_658_676_000_000));
        assert_eq!(post.ts_init, UnixNanos::from(1_682_900_658_698_855_000));
    }

    #[rstest]
    fn test_load_depth_from_snapshot5_comprehensive() {
        let filepath = get_tardis_binance_snapshot5_path();
        let depths = load_depth_from_snapshot5(&filepath, None, None, None, Some(100)).unwrap();

        assert_eq!(depths.len(), 10);

        let first = &depths[0];
        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
        assert_eq!(first.bids.len(), 5);
        assert_eq!(first.asks.len(), 5);

        // Check all bid levels (5 from data)
        assert_eq!(first.bids[0].price, Price::from("11657.07"));
        assert_eq!(first.bids[0].size, Quantity::from("10.896"));
        assert_eq!(first.bids[0].side, OrderSide::Buy.into());

        assert_eq!(first.bids[1].price, Price::from("11656.97"));
        assert_eq!(first.bids[1].size, Quantity::from("0.2"));
        assert_eq!(first.bids[1].side, OrderSide::Buy.into());

        assert_eq!(first.bids[2].price, Price::from("11655.78"));
        assert_eq!(first.bids[2].size, Quantity::from("0.2"));
        assert_eq!(first.bids[2].side, OrderSide::Buy.into());

        assert_eq!(first.bids[3].price, Price::from("11655.77"));
        assert_eq!(first.bids[3].size, Quantity::from("0.98"));
        assert_eq!(first.bids[3].side, OrderSide::Buy.into());

        assert_eq!(first.bids[4].price, Price::from("11655.68"));
        assert_eq!(first.bids[4].size, Quantity::from("0.111"));
        assert_eq!(first.bids[4].side, OrderSide::Buy.into());

        // Check all ask levels (5 from data)
        assert_eq!(first.asks[0].price, Price::from("11657.08"));
        assert_eq!(first.asks[0].size, Quantity::from("1.714"));
        assert_eq!(first.asks[0].side, OrderSide::Sell.into());

        assert_eq!(first.asks[1].price, Price::from("11657.54"));
        assert_eq!(first.asks[1].size, Quantity::from("5.4"));
        assert_eq!(first.asks[1].side, OrderSide::Sell.into());

        assert_eq!(first.asks[2].price, Price::from("11657.56"));
        assert_eq!(first.asks[2].size, Quantity::from("0.238"));
        assert_eq!(first.asks[2].side, OrderSide::Sell.into());

        assert_eq!(first.asks[3].price, Price::from("11657.61"));
        assert_eq!(first.asks[3].size, Quantity::from("0.077"));
        assert_eq!(first.asks[3].side, OrderSide::Sell.into());

        assert_eq!(first.asks[4].price, Price::from("11657.92"));
        assert_eq!(first.asks[4].size, Quantity::from("0.918"));
        assert_eq!(first.asks[4].side, OrderSide::Sell.into());

        // Logical checks: bid prices should decrease
        for i in 1..5 {
            assert!(
                first.bids[i].price < first.bids[i - 1].price,
                "Bid price at level {} should be less than level {}",
                i,
                i - 1
            );
        }

        // Logical checks: ask prices should increase
        for i in 1..5 {
            assert!(
                first.asks[i].price > first.asks[i - 1].price,
                "Ask price at level {} should be greater than level {}",
                i,
                i - 1
            );
        }

        // Logical check: spread should be positive
        assert!(
            first.asks[0].price > first.bids[0].price,
            "Best ask should be greater than best bid"
        );

        assert_eq!(first.bid_counts.as_slice(), &[1; 5]);
        assert_eq!(first.ask_counts.as_slice(), &[1; 5]);
        for order in first.bids.iter().chain(&first.asks) {
            assert_eq!(order.order_id, 0);
        }

        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
        assert_eq!(
            first.flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
        assert_eq!(first.sequence, 0);
    }

    #[rstest]
    fn test_load_depth_from_snapshot25_comprehensive() {
        let filepath = get_tardis_binance_snapshot25_path();
        let depths = load_depth_from_snapshot25(&filepath, None, None, None, Some(100)).unwrap();

        assert_eq!(depths.len(), 10);

        let first = &depths[0];
        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
        assert_eq!(first.bids.len(), 25);
        assert_eq!(first.asks.len(), 25);

        // Check all 25 bid levels from snapshot25
        let expected_bids = vec![
            ("11657.07", "10.896"),
            ("11656.97", "0.2"),
            ("11655.78", "0.2"),
            ("11655.77", "0.98"),
            ("11655.68", "0.111"),
            ("11655.66", "0.077"),
            ("11655.57", "0.34"),
            ("11655.48", "0.4"),
            ("11655.26", "1.185"),
            ("11654.86", "0.195"),
            ("11654.85", "0.275"),
            ("11654.7", "0.175"),
            ("11654.69", "0.194"),
            ("11654.67", "1"),
            ("11654.65", "0.05"),
            ("11654.58", "0.05"),
            ("11654.41", "0.11"),
            ("11654.28", "0.618"),
            ("11653.84", "0.135"),
            ("11653.4", "0.17"),
            ("11653.39", "1.008"),
            ("11653.35", "4"),
            ("11653.34", "2"),
            ("11653.32", "0.5"),
            ("11653.25", "1.003"),
        ];

        for (i, (price, size)) in expected_bids.iter().enumerate() {
            assert_eq!(first.bids[i].price, Price::from(*price));
            assert_eq!(first.bids[i].size, Quantity::from(*size));
            assert_eq!(first.bids[i].side, OrderSide::Buy.into());
        }

        // Check all 25 ask levels from snapshot25
        let expected_asks = vec![
            ("11657.08", "1.714"),
            ("11657.54", "5.4"),
            ("11657.56", "0.238"),
            ("11657.61", "0.077"),
            ("11657.92", "0.918"),
            ("11658.09", "1.015"),
            ("11658.12", "0.665"),
            ("11658.19", "0.583"),
            ("11658.28", "0.255"),
            ("11658.29", "0.656"),
            ("11658.64", "1.463"),
            ("11658.71", "0.155"),
            ("11658.75", "0.155"),
            ("11658.88", "0.625"),
            ("11658.94", "0.155"),
            ("11658.98", "1.005"),
            ("11658.99", "0.155"),
            ("11659", "1.922"),
            ("11659.02", "0.001"),
            ("11659.13", "0.11"),
            ("11659.17", "0.144"),
            ("11659.18", "0.665"),
            ("11659.22", "0.22"),
            ("11659.28", "0.06"),
            ("11659.34", "0.618"),
        ];

        for (i, (price, size)) in expected_asks.iter().enumerate() {
            assert_eq!(first.asks[i].price, Price::from(*price));
            assert_eq!(first.asks[i].size, Quantity::from(*size));
            assert_eq!(first.asks[i].side, OrderSide::Sell.into());
        }

        // Logical checks: bid prices should strictly decrease
        for i in 1..25 {
            assert!(
                first.bids[i].price < first.bids[i - 1].price,
                "Bid price at level {} ({}) should be less than level {} ({})",
                i,
                first.bids[i].price,
                i - 1,
                first.bids[i - 1].price
            );
        }

        // Logical checks: ask prices should strictly increase
        for i in 1..25 {
            assert!(
                first.asks[i].price > first.asks[i - 1].price,
                "Ask price at level {} ({}) should be greater than level {} ({})",
                i,
                first.asks[i].price,
                i - 1,
                first.asks[i - 1].price
            );
        }

        // Logical check: spread should be positive
        assert!(
            first.asks[0].price > first.bids[0].price,
            "Best ask ({}) should be greater than best bid ({})",
            first.asks[0].price,
            first.bids[0].price
        );

        // Check counts (all should be 1 for snapshot data)
        for i in 0..25 {
            assert_eq!(first.bid_counts[i], 1);
            assert_eq!(first.ask_counts[i], 1);
        }

        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
        assert_eq!(
            first.flags,
            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
        );
        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
        assert_eq!(first.sequence, 0);
    }

    /// Writes a two-row depth snapshot CSV with `levels` levels per side where the
    /// second row's best bid price carries higher precision, forcing a retroactive
    /// precision rewrite of the first row's depth.
    fn write_precision_rewrite_csv(path: &std::path::Path, levels: usize) {
        let mut headers = vec![
            "exchange".to_string(),
            "symbol".to_string(),
            "timestamp".to_string(),
            "local_timestamp".to_string(),
        ];

        for i in 0..levels {
            headers.extend([
                format!("asks[{i}].price"),
                format!("asks[{i}].amount"),
                format!("bids[{i}].price"),
                format!("bids[{i}].amount"),
            ]);
        }

        let row = |timestamp: &str, best_bid: &str| -> String {
            let mut fields = vec![
                "binance-futures".to_string(),
                "BTCUSDT".to_string(),
                timestamp.to_string(),
                timestamp.to_string(),
            ];

            for i in 0..levels {
                let bid = if i == 0 {
                    best_bid.to_string()
                } else {
                    format!("{}", 49_990 - i)
                };
                fields.extend([
                    format!("{}", 50_001 + i),
                    "1.5".to_string(),
                    bid,
                    "1.5".to_string(),
                ]);
            }
            fields.join(",")
        };

        let csv_data = format!(
            "{}\n{}\n{}",
            headers.join(","),
            row("1640995200000000", "49999"),
            row("1640995201000000", "49998.12"),
        );
        std::fs::write(path, csv_data).unwrap();
    }

    #[rstest]
    fn test_load_depth_from_snapshot25_precision_rewrite_updates_all_levels() {
        let temp_file = std::env::temp_dir().join("test_depth_snapshot25_precision_rewrite.csv");
        write_precision_rewrite_csv(&temp_file, 25);

        let depths = load_depth_from_snapshot25(&temp_file, None, None, None, None).unwrap();
        assert_eq!(depths.len(), 2);

        // Both rows end at the maximum inferred precision on every level,
        // including levels past the first 10
        for (r, depth) in depths.iter().enumerate() {
            assert_eq!(depth.bids.len(), 25, "row {r}");
            assert_eq!(depth.asks.len(), 25, "row {r}");
            for (i, order) in depth.bids.iter().chain(depth.asks.iter()).enumerate() {
                assert_eq!(order.price.precision, 2, "row {r} level {i}");
                assert_eq!(order.size.precision, 1, "row {r} level {i}");
            }
        }
        // Values are unchanged; only the precision display is rewritten
        assert_eq!(depths[0].bids[0].price, Price::new(49999.0, 2));
        assert_eq!(depths[0].bids[24].price, Price::new(49966.0, 2));
        assert_eq!(depths[1].bids[0].price, Price::new(49998.12, 2));

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    fn test_load_depth_from_snapshot5_precision_rewrite_updates_all_levels() {
        let temp_file = std::env::temp_dir().join("test_depth_snapshot5_precision_rewrite.csv");
        write_precision_rewrite_csv(&temp_file, 5);

        let depths = load_depth_from_snapshot5(&temp_file, None, None, None, None).unwrap();
        assert_eq!(depths.len(), 2);

        // The rewrite must cover the 5 retained levels without indexing past them
        for (r, depth) in depths.iter().enumerate() {
            assert_eq!(depth.bids.len(), 5, "row {r}");
            assert_eq!(depth.asks.len(), 5, "row {r}");
            for (i, order) in depth.bids.iter().chain(depth.asks.iter()).enumerate() {
                assert_eq!(order.price.precision, 2, "row {r} level {i}");
                assert_eq!(order.size.precision, 1, "row {r} level {i}");
            }
        }
        assert_eq!(depths[0].bids[0].price, Price::new(49999.0, 2));
        assert_eq!(depths[1].bids[0].price, Price::new(49998.12, 2));

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    fn test_snapshot_csv_field_order_interleaved() {
        // This test verifies that the CSV structs correctly handle the interleaved
        // asks/bids field ordering from Tardis CSV files

        let csv_data = "exchange,symbol,timestamp,local_timestamp,\
asks[0].price,asks[0].amount,bids[0].price,bids[0].amount,\
asks[1].price,asks[1].amount,bids[1].price,bids[1].amount,\
asks[2].price,asks[2].amount,bids[2].price,bids[2].amount,\
asks[3].price,asks[3].amount,bids[3].price,bids[3].amount,\
asks[4].price,asks[4].amount,bids[4].price,bids[4].amount
binance-futures,BTCUSDT,1000000,2000000,\
100.5,1.0,100.4,2.0,\
100.6,1.1,100.3,2.1,\
100.7,1.2,100.2,2.2,\
100.8,1.3,100.1,2.3,\
100.9,1.4,100.0,2.4";

        let temp_file = std::env::temp_dir().join("test_interleaved_snapshot5.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let depths = load_depth_from_snapshot5(&temp_file, None, None, None, Some(1)).unwrap();
        assert_eq!(depths.len(), 1);

        let depth = &depths[0];

        // Verify bids are correctly parsed (should be decreasing)
        assert_eq!(depth.bids[0].price, Price::from("100.4"));
        assert_eq!(depth.bids[1].price, Price::from("100.3"));
        assert_eq!(depth.bids[2].price, Price::from("100.2"));
        assert_eq!(depth.bids[3].price, Price::from("100.1"));
        assert_eq!(depth.bids[4].price, Price::from("100.0"));

        // Verify asks are correctly parsed (should be increasing)
        assert_eq!(depth.asks[0].price, Price::from("100.5"));
        assert_eq!(depth.asks[1].price, Price::from("100.6"));
        assert_eq!(depth.asks[2].price, Price::from("100.7"));
        assert_eq!(depth.asks[3].price, Price::from("100.8"));
        assert_eq!(depth.asks[4].price, Price::from("100.9"));

        // Verify sizes
        assert_eq!(depth.bids[0].size, Quantity::from("2.0"));
        assert_eq!(depth.asks[0].size, Quantity::from("1.0"));

        std::fs::remove_file(temp_file).unwrap();
    }

    #[rstest]
    fn test_load_deltas_limit_includes_clear_deltas() {
        // Test that limit counts total emitted deltas (including CLEARs)
        // When limit=5, we should get exactly 5 deltas: 1 CLEAR + 4 data deltas
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0
binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.0,0.5
binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50002.0,1.5
binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49998.0,0.5
binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50003.0,2.0
binance-futures,BTCUSDT,1640995205000000,1640995205100000,false,bid,49997.0,0.5";

        let temp_file = std::env::temp_dir().join("test_load_deltas_limit.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        // Load with limit=5 (should emit exactly 5 deltas including CLEAR)
        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(5)).unwrap();

        // Should have exactly 5 deltas: 1 CLEAR + 4 data deltas
        assert_eq!(deltas.len(), 5);
        assert_eq!(deltas[0].action, BookAction::Clear);
        assert_eq!(deltas[1].action, BookAction::Add);
        assert_eq!(deltas[2].action, BookAction::Add);
        assert_eq!(deltas[3].action, BookAction::Update);
        assert_eq!(deltas[4].action, BookAction::Update);

        // Verify the last delta is from the 4th CSV record (49999.0 bid)
        assert_eq!(deltas[3].order.price, parse_price(49999.0, 1));

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    fn test_load_deltas_limit_stops_at_clear() {
        // Test that limit=1 with snapshot data returns only the CLEAR delta
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0";

        let temp_file = std::env::temp_dir().join("test_load_deltas_limit_stops_at_clear.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        // Load with limit=1 should only get the CLEAR delta
        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(1)).unwrap();

        assert_eq!(deltas.len(), 1);
        assert_eq!(deltas[0].action, BookAction::Clear);

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    fn test_load_deltas_with_consecutive_snapshots_inserts_clear() {
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
hyperliquid,BTC,1640995200000000,1640995200100000,true,bid,50000.0,1.0
hyperliquid,BTC,1640995200000001,1640995200100000,true,ask,50001.0,2.0
hyperliquid,BTC,1640995201000000,1640995201100000,true,bid,49990.0,3.0
hyperliquid,BTC,1640995201000001,1640995201100000,true,ask,49991.0,4.0";

        let temp_file = std::env::temp_dir().join("test_load_deltas_consecutive_snapshots.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, None).unwrap();
        let clear_count = deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        assert_eq!(clear_count, 2);
        assert_eq!(deltas[0].action, BookAction::Clear);
        assert_eq!(deltas[3].action, BookAction::Clear);
        assert_eq!(
            deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
        assert_eq!(deltas[3].flags & RecordFlag::F_LAST as u8, 0);
        assert_eq!(
            deltas
                .iter()
                .map(|delta| (delta.action, delta.flags, delta.ts_event, delta.ts_init))
                .collect::<Vec<_>>(),
            vec![
                (
                    BookAction::Clear,
                    RecordFlag::F_SNAPSHOT as u8,
                    UnixNanos::from(1_640_995_200_000_000_000),
                    UnixNanos::from(1_640_995_200_100_000_000),
                ),
                (
                    BookAction::Add,
                    0,
                    UnixNanos::from(1_640_995_200_000_000_000),
                    UnixNanos::from(1_640_995_200_100_000_000),
                ),
                (
                    BookAction::Add,
                    RecordFlag::F_LAST as u8,
                    UnixNanos::from(1_640_995_200_000_001_000),
                    UnixNanos::from(1_640_995_200_100_000_000),
                ),
                (
                    BookAction::Clear,
                    RecordFlag::F_SNAPSHOT as u8,
                    UnixNanos::from(1_640_995_201_000_000_000),
                    UnixNanos::from(1_640_995_201_100_000_000),
                ),
                (
                    BookAction::Add,
                    0,
                    UnixNanos::from(1_640_995_201_000_000_000),
                    UnixNanos::from(1_640_995_201_100_000_000),
                ),
                (
                    BookAction::Add,
                    RecordFlag::F_LAST as u8,
                    UnixNanos::from(1_640_995_201_000_001_000),
                    UnixNanos::from(1_640_995_201_100_000_000),
                ),
            ]
        );

        std::fs::remove_file(&temp_file).ok();
    }

    #[rstest]
    fn test_load_deltas_limit_with_mid_day_snapshot() {
        // Test limit behavior when there's a mid-day snapshot
        // The limit counts total emitted deltas including CLEARs
        let filepath = get_test_data_path("csv/deltas_with_snapshot.csv");
        let deltas = load_deltas(filepath, Some(1), Some(1), None, Some(5)).unwrap();

        // With limit=5, we get exactly 5 deltas
        // First snapshot inserts CLEAR, then we get 4 more data deltas
        assert_eq!(deltas.len(), 5);
        assert_eq!(deltas[0].action, BookAction::Clear);
    }

    // Curates the large Tardis Deribit CSV.gz into NautilusTrader Parquet format.
    // Run manually: `cargo test -p nautilus-tardis test_curate_deribit_deltas -- --ignored --nocapture`
    #[rstest]
    #[ignore = "one-time dataset curation, not for routine CI"]
    fn test_curate_deribit_deltas() {
        let csv_path = get_test_data_root()
            .join("large")
            .join("tardis_deribit_incremental_book_L2_2020-04-01_BTC-PERPETUAL.csv.gz");

        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
        let parquet_path = "/tmp/tardis_BTC-PERPETUAL.DERIBIT_2020-04-01_deltas.parquet";

        println!("Loading deltas from {}", csv_path.display());
        let deltas = load_deltas(&csv_path, None, None, Some(instrument_id), None).unwrap();
        let count = deltas.len();
        println!("Loaded {count} deltas");

        let sample = deltas
            .iter()
            .find(|d| d.order.price.precision > 0)
            .expect("Should have at least one non-CLEAR delta");
        let price_precision = sample.order.price.precision;
        let size_precision = sample.order.size.precision;
        println!("Precision: price={price_precision}, size={size_precision}");

        // Write in chunks to avoid stack overflow on large batches
        let metadata =
            OrderBookDelta::get_metadata(&instrument_id, price_precision, size_precision);
        let schema = OrderBookDelta::get_schema(Some(metadata.clone()));

        println!("Writing Parquet to {parquet_path}");
        let file = File::create(parquet_path).unwrap();
        let zstd_level = parquet::basic::ZstdLevel::try_new(3).unwrap();
        let props = WriterProperties::builder()
            .set_compression(parquet::basic::Compression::ZSTD(zstd_level))
            .set_max_row_group_row_count(Some(1_000_000))
            .build();
        let mut writer = ArrowWriter::try_new(file, Arc::new(schema), Some(props)).unwrap();

        let chunk_size = 1_000_000;
        for (i, chunk) in deltas.chunks(chunk_size).enumerate() {
            println!("  Encoding chunk {} ({} records)...", i + 1, chunk.len());
            let batch = OrderBookDelta::encode_batch(&metadata, chunk).unwrap();
            writer.write(&batch).unwrap();
        }
        writer.close().unwrap();

        let file_size = fs::metadata(parquet_path).unwrap().len();
        println!("\n=== CURATION COMPLETE ===");
        println!("Records: {count}");
        println!("Price precision: {price_precision}");
        println!("Size precision: {size_precision}");
        println!(
            "File size: {} bytes ({:.1} MB)",
            file_size,
            file_size as f64 / 1_048_576.0
        );
        println!("Output: {parquet_path}");
        println!("\nNext steps:");
        println!("  sha256sum {parquet_path}");
    }
}