nautilus_tardis/csv/

stream.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::{io::Read, path::Path};

use ahash::AHashMap;
use csv::{Reader, StringRecord};
use nautilus_core::UnixNanos;
#[cfg(feature = "python")]
use nautilus_model::{data::OrderBookDeltas, python::data::data_to_pyobject};
use nautilus_model::{
    data::{DEPTH10_LEN, Data, NULL_ORDER, OrderBookDelta, OrderBookDepth10, QuoteTick, TradeTick},
    enums::{OrderSide, RecordFlag},
    identifiers::InstrumentId,
    types::Quantity,
};
#[cfg(feature = "python")]
use pyo3::{Py, PyAny, PyResult, Python};

use crate::{
    common::parse::{parse_instrument_id, parse_timestamp},
    csv::{
        create_book_order, create_csv_reader, infer_precision,
        load::OptionsChainPrecision,
        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, TardisOptionsChainRecord, TardisOrderBookSnapshot5Record,
            TardisOrderBookSnapshot25Record, TardisQuoteRecord, TardisTradeRecord,
        },
    },
};

////////////////////////////////////////////////////////////////////////////////
// OrderBookDelta Streaming
////////////////////////////////////////////////////////////////////////////////

/// Streaming iterator over CSV records that yields chunks of parsed data.
struct DeltaStreamIterator {
    reader: Reader<Box<dyn std::io::Read>>,
    record: StringRecord,
    buffer: Vec<OrderBookDelta>,
    chunk_size: usize,
    instrument_id: Option<InstrumentId>,
    price_precision: u8,
    size_precision: u8,
    last_ts_init: Option<UnixNanos>,
    last_is_snapshot: bool,
    limit: Option<usize>,
    deltas_emitted: usize,

    pending: Option<anyhow::Result<TardisBookUpdateRecord>>,
}

impl DeltaStreamIterator {
    /// Creates a new [`DeltaStreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read.
    fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let (final_price_precision, final_size_precision) =
            if let (Some(price_prec), Some(size_prec)) = (price_precision, size_precision) {
                // Both precisions provided, use them directly
                (price_prec, size_prec)
            } else {
                // One or both precisions missing, detect only the missing ones
                let mut reader = create_csv_reader(&filepath)?;
                let mut record = StringRecord::new();
                let (detected_price, detected_size) =
                    Self::detect_precision_from_sample(&mut reader, &mut record, 10_000);
                (
                    price_precision.unwrap_or(detected_price),
                    size_precision.unwrap_or(detected_size),
                )
            };

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size),
            chunk_size,
            instrument_id,
            price_precision: final_price_precision,
            size_precision: final_size_precision,
            last_ts_init: None,
            last_is_snapshot: false,
            limit,
            deltas_emitted: 0,
            pending: None,
        })
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn std::io::Read>>,
        record: &mut StringRecord,
        sample_size: usize,
    ) -> (u8, u8) {
        let mut max_price_precision = 0u8;
        let mut max_size_precision = 0u8;
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    if let Ok(data) = record.deserialize::<TardisBookUpdateRecord>(None) {
                        max_price_precision = max_price_precision.max(infer_precision(data.price));
                        max_size_precision = max_size_precision.max(infer_precision(data.amount));
                        records_scanned += 1;
                    }
                }
                Ok(false) => break,             // End of file
                Err(_) => records_scanned += 1, // Skip malformed records
            }
        }

        (max_price_precision, max_size_precision)
    }
}

impl Iterator for DeltaStreamIterator {
    type Item = anyhow::Result<Vec<OrderBookDelta>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.deltas_emitted >= limit
        {
            return None;
        }

        self.buffer.clear();

        loop {
            if self.buffer.len() >= self.chunk_size {
                break;
            }

            if let Some(limit) = self.limit
                && self.deltas_emitted >= limit
            {
                break;
            }

            let data = match self.pending.take() {
                Some(Ok(data)) => data,
                Some(Err(e)) => return Some(Err(e)),
                None => match self.read_record() {
                    Ok(Some(data)) => data,
                    Ok(None) => {
                        if self.buffer.is_empty() {
                            return None;
                        }

                        if let Some(last_delta) = self.buffer.last_mut() {
                            last_delta.flags = RecordFlag::F_LAST as u8;
                        }
                        return Some(Ok(self.buffer.clone()));
                    }
                    Err(e) => return Some(Err(e)),
                },
            };

            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 && (!self.last_is_snapshot || self.last_ts_init != Some(ts_init));

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

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

                let clear_delta = OrderBookDelta::clear(clear_instrument_id, 0, ts_event, ts_init);
                self.buffer.push(clear_delta);
                self.deltas_emitted += 1;

                // Defer real delta to next chunk if constraints reached
                if self.buffer.len() >= self.chunk_size
                    || self.limit.is_some_and(|l| self.deltas_emitted >= l)
                {
                    self.last_is_snapshot = data.is_snapshot;
                    self.pending = Some(Ok(data));
                    break;
                }
            }
            self.last_is_snapshot = data.is_snapshot;

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

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

            self.last_ts_init = Some(delta.ts_init);

            self.buffer.push(delta);
            self.deltas_emitted += 1;

            if self.buffer.len() >= self.chunk_size
                && !self.limit.is_some_and(|l| self.deltas_emitted >= l)
            {
                match self.read_record() {
                    Ok(Some(data)) => {
                        let next_ts_init = parse_timestamp(data.local_timestamp);
                        if self.last_ts_init != Some(next_ts_init)
                            && let Some(last_delta) = self.buffer.last_mut()
                        {
                            last_delta.flags = RecordFlag::F_LAST as u8;
                        }
                        self.pending = Some(Ok(data));
                    }
                    Ok(None) => {
                        if let Some(last_delta) = self.buffer.last_mut() {
                            last_delta.flags = RecordFlag::F_LAST as u8;
                        }
                    }
                    Err(e) => self.pending = Some(Err(e)),
                }
                break;
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            // Only set F_LAST when limit reached (stream ending), not on chunk
            // boundary where more deltas from the same message may follow
            if let Some(limit) = self.limit
                && self.deltas_emitted >= limit
                && let Some(last_delta) = self.buffer.last_mut()
            {
                last_delta.flags = RecordFlag::F_LAST as u8;
            }
            Some(Ok(self.buffer.clone()))
        }
    }
}

impl DeltaStreamIterator {
    fn read_record(&mut self) -> anyhow::Result<Option<TardisBookUpdateRecord>> {
        if !self
            .reader
            .read_record(&mut self.record)
            .map_err(|e| anyhow::anyhow!("Failed to read record: {e}"))?
        {
            return Ok(None);
        }

        self.record
            .deserialize::<TardisBookUpdateRecord>(None)
            .map(Some)
            .map_err(|e| anyhow::anyhow!("Failed to deserialize record: {e}"))
    }
}

/// Streams [`OrderBookDelta`]s from a Tardis format CSV at the given `filepath`,
/// yielding chunks of the specified size.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference (not providing explicit precisions),
/// the inferred precision may differ from bulk loading the entire file. This is because
/// precision inference works within chunk boundaries, and different chunks may contain
/// values with different precision requirements. For deterministic precision behavior,
/// provide explicit `price_precision` and `size_precision` parameters.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_deltas<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<OrderBookDelta>>>> {
    DeltaStreamIterator::new(
        filepath,
        chunk_size,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

#[cfg(feature = "python")]
/// Streaming iterator over CSV records that yields chunks of parsed data.
struct BatchedDeltasStreamIterator {
    reader: Reader<Box<dyn std::io::Read>>,
    record: StringRecord,
    current_batch: Vec<OrderBookDelta>,
    pending_batches: Vec<Vec<OrderBookDelta>>,
    chunk_size: usize,
    instrument_id: InstrumentId,
    price_precision: u8,
    size_precision: u8,
    last_ts_init: Option<UnixNanos>,
    last_is_snapshot: bool,
    limit: Option<usize>,
    deltas_emitted: usize,
}

#[cfg(feature = "python")]
impl BatchedDeltasStreamIterator {
    /// Creates a new [`DeltaStreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read.
    fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let mut reader = create_csv_reader(&filepath)?;
        let mut record = StringRecord::new();

        let first_record = if reader.read_record(&mut record)? {
            record.deserialize::<TardisBookUpdateRecord>(None)?
        } else {
            anyhow::bail!("CSV file is empty");
        };

        let final_instrument_id = instrument_id
            .unwrap_or_else(|| parse_instrument_id(&first_record.exchange, first_record.symbol));

        let (final_price_precision, final_size_precision) =
            if let (Some(price_prec), Some(size_prec)) = (price_precision, size_precision) {
                // Both precisions provided, use them directly
                (price_prec, size_prec)
            } else {
                // One or both precisions missing, detect from sample including first record
                let (detected_price, detected_size) =
                    Self::detect_precision_from_sample(&mut reader, &mut record, 10_000);
                (
                    price_precision.unwrap_or(detected_price),
                    size_precision.unwrap_or(detected_size),
                )
            };

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            current_batch: Vec::new(),
            pending_batches: Vec::with_capacity(chunk_size),
            chunk_size,
            instrument_id: final_instrument_id,
            price_precision: final_price_precision,
            size_precision: final_size_precision,
            last_ts_init: None,
            last_is_snapshot: false,
            limit,
            deltas_emitted: 0,
        })
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn std::io::Read>>,
        record: &mut StringRecord,
        sample_size: usize,
    ) -> (u8, u8) {
        let mut max_price_precision = 0u8;
        let mut max_size_precision = 0u8;
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    if let Ok(data) = record.deserialize::<TardisBookUpdateRecord>(None) {
                        max_price_precision = max_price_precision.max(infer_precision(data.price));
                        max_size_precision = max_size_precision.max(infer_precision(data.amount));
                        records_scanned += 1;
                    }
                }
                Ok(false) => break,             // End of file
                Err(_) => records_scanned += 1, // Skip malformed records
            }
        }

        (max_price_precision, max_size_precision)
    }

    fn fill_pending_batches(&mut self) -> Option<anyhow::Result<()>> {
        self.pending_batches.clear();
        let mut batches_created = 0;

        while batches_created < self.chunk_size {
            if let Some(limit) = self.limit
                && self.deltas_emitted >= limit
            {
                break;
            }

            match self.reader.read_record(&mut self.record) {
                Ok(true) => {
                    let data = match self.record.deserialize::<TardisBookUpdateRecord>(None) {
                        Ok(data) => data,
                        Err(e) => {
                            return Some(Err(anyhow::anyhow!("Failed to deserialize record: {e}")));
                        }
                    };

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

                    // Parse before any state changes so invalid records
                    // don't corrupt batch boundaries or snapshot tracking
                    let delta = match parse_delta_record(
                        &data,
                        self.price_precision,
                        self.size_precision,
                        Some(self.instrument_id),
                    ) {
                        Ok(d) => d,
                        Err(e) => {
                            log::warn!("Skipping invalid delta record: {e}");
                            continue;
                        }
                    };

                    let starts_new_message = self.last_ts_init != Some(ts_init);

                    if starts_new_message && !self.current_batch.is_empty() {
                        // Set F_LAST on the last delta of the completed batch
                        if let Some(last_delta) = self.current_batch.last_mut() {
                            last_delta.flags = RecordFlag::F_LAST as u8;
                        }
                        self.pending_batches
                            .push(std::mem::take(&mut self.current_batch));
                        batches_created += 1;
                    }

                    // 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.
                    if data.is_snapshot && (!self.last_is_snapshot || starts_new_message) {
                        let clear_delta =
                            OrderBookDelta::clear(self.instrument_id, 0, ts_event, ts_init);
                        self.current_batch.push(clear_delta);
                        self.deltas_emitted += 1;

                        if let Some(limit) = self.limit
                            && self.deltas_emitted >= limit
                        {
                            self.last_is_snapshot = data.is_snapshot;
                            break;
                        }
                    }
                    self.last_ts_init = Some(ts_init);
                    self.last_is_snapshot = data.is_snapshot;

                    self.current_batch.push(delta);
                    self.deltas_emitted += 1;

                    if let Some(limit) = self.limit
                        && self.deltas_emitted >= limit
                    {
                        break;
                    }
                }
                Ok(false) => {
                    // End of file
                    break;
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if !self.current_batch.is_empty() && batches_created < self.chunk_size {
            // Ensure the last delta of the last batch has F_LAST set
            if let Some(last_delta) = self.current_batch.last_mut() {
                last_delta.flags = RecordFlag::F_LAST as u8;
            }
            self.pending_batches
                .push(std::mem::take(&mut self.current_batch));
        }

        if self.pending_batches.is_empty() {
            None
        } else {
            Some(Ok(()))
        }
    }
}

#[cfg(feature = "python")]
impl Iterator for BatchedDeltasStreamIterator {
    type Item = anyhow::Result<Vec<Py<PyAny>>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.deltas_emitted >= limit
        {
            return None;
        }

        if let Some(Err(e)) = self.fill_pending_batches() {
            return Some(Err(e));
        }

        if self.pending_batches.is_empty() {
            None
        } else {
            let batches = std::mem::take(&mut self.pending_batches);
            let result = Python::attach(|py| {
                batches
                    .into_iter()
                    .map(|batch| {
                        let deltas = OrderBookDeltas::new(self.instrument_id, batch);
                        let deltas = Box::new(deltas);
                        data_to_pyobject(py, Data::Deltas(deltas))
                    })
                    .collect::<PyResult<Vec<_>>>()
            })
            .map_err(|e| anyhow::anyhow!("Failed to convert batched deltas to Python: {e}"));
            Some(result)
        }
    }
}

#[cfg(feature = "python")]
/// Streams batches of `OrderBookDeltas` Python objects from a Tardis format CSV at the given
/// `filepath`, yielding chunks of the specified size.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_batched_deltas<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<Py<PyAny>>>>> {
    BatchedDeltasStreamIterator::new(
        filepath,
        chunk_size,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

////////////////////////////////////////////////////////////////////////////////
// Quote Streaming
////////////////////////////////////////////////////////////////////////////////

/// An iterator for streaming [`QuoteTick`]s from a Tardis CSV file in chunks.
struct QuoteStreamIterator {
    reader: Reader<Box<dyn Read>>,
    record: StringRecord,
    buffer: Vec<QuoteTick>,
    chunk_size: usize,
    instrument_id: Option<InstrumentId>,
    price_precision: u8,
    size_precision: u8,
    limit: Option<usize>,
    records_processed: usize,
}

impl QuoteStreamIterator {
    /// Creates a new [`QuoteStreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read.
    pub(crate) fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let (final_price_precision, final_size_precision) =
            if let (Some(price_prec), Some(size_prec)) = (price_precision, size_precision) {
                // Both precisions provided, use them directly
                (price_prec, size_prec)
            } else {
                // One or both precisions missing, detect only the missing ones
                let mut reader = create_csv_reader(&filepath)?;
                let mut record = StringRecord::new();
                let (detected_price, detected_size) =
                    Self::detect_precision_from_sample(&mut reader, &mut record, 10_000);
                (
                    price_precision.unwrap_or(detected_price),
                    size_precision.unwrap_or(detected_size),
                )
            };

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size),
            chunk_size,
            instrument_id,
            price_precision: final_price_precision,
            size_precision: final_size_precision,
            limit,
            records_processed: 0,
        })
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn std::io::Read>>,
        record: &mut StringRecord,
        sample_size: usize,
    ) -> (u8, u8) {
        let mut max_price_precision = 2u8;
        let mut max_size_precision = 0u8;
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    if let Ok(data) = record.deserialize::<TardisQuoteRecord>(None) {
                        if let Some(bid_price_val) = data.bid_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(bid_price_val));
                        }

                        if let Some(ask_price_val) = data.ask_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(ask_price_val));
                        }

                        if let Some(bid_amount_val) = data.bid_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(bid_amount_val));
                        }

                        if let Some(ask_amount_val) = data.ask_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(ask_amount_val));
                        }
                        records_scanned += 1;
                    }
                }
                Ok(false) => break,             // End of file
                Err(_) => records_scanned += 1, // Skip malformed records
            }
        }

        (max_price_precision, max_size_precision)
    }
}

impl Iterator for QuoteStreamIterator {
    type Item = anyhow::Result<Vec<QuoteTick>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.records_processed >= limit
        {
            return None;
        }

        self.buffer.clear();
        let mut records_read = 0;

        while records_read < self.chunk_size {
            match self.reader.read_record(&mut self.record) {
                Ok(true) => match self.record.deserialize::<TardisQuoteRecord>(None) {
                    Ok(data) => {
                        let quote = parse_quote_record(
                            &data,
                            self.price_precision,
                            self.size_precision,
                            self.instrument_id,
                        );

                        self.buffer.push(quote);
                        records_read += 1;
                        self.records_processed += 1;

                        if let Some(limit) = self.limit
                            && self.records_processed >= limit
                        {
                            break;
                        }
                    }
                    Err(e) => {
                        return Some(Err(anyhow::anyhow!("Failed to deserialize record: {e}")));
                    }
                },
                Ok(false) => {
                    if self.buffer.is_empty() {
                        return None;
                    }
                    return Some(Ok(self.buffer.clone()));
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            Some(Ok(self.buffer.clone()))
        }
    }
}

/// Streams [`QuoteTick`]s from a Tardis format CSV at the given `filepath`,
/// yielding chunks of the specified size.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference (not providing explicit precisions),
/// the inferred precision may differ from bulk loading the entire file. This is because
/// precision inference works within chunk boundaries, and different chunks may contain
/// values with different precision requirements. For deterministic precision behavior,
/// provide explicit `price_precision` and `size_precision` parameters.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_quotes<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<QuoteTick>>>> {
    QuoteStreamIterator::new(
        filepath,
        chunk_size,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

struct OptionsChainStreamIterator {
    reader: Reader<Box<dyn Read>>,
    record: StringRecord,
    buffer: Vec<Data>,
    chunk_size: usize,
    underlyings: Option<Vec<String>>,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    precision_by_instrument: AHashMap<InstrumentId, OptionsChainPrecision>,
    limit: Option<usize>,
    records_processed: usize,
}

impl OptionsChainStreamIterator {
    pub(crate) fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        underlyings: Option<Vec<String>>,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let underlyings = normalize_underlying_filters(underlyings);
        let mut precision_by_instrument = AHashMap::new();

        if price_precision.is_none() || size_precision.is_none() {
            let mut reader = create_csv_reader(&filepath)?;
            let mut record = StringRecord::new();
            Self::detect_precision_from_sample(
                &mut reader,
                &mut record,
                underlyings.as_deref(),
                price_precision,
                size_precision,
                &mut precision_by_instrument,
                10_000,
            );
        }

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size * 2),
            chunk_size,
            underlyings,
            price_precision,
            size_precision,
            precision_by_instrument,
            limit,
            records_processed: 0,
        })
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn Read>>,
        record: &mut StringRecord,
        underlyings: Option<&[String]>,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        precision_by_instrument: &mut AHashMap<InstrumentId, OptionsChainPrecision>,
        sample_size: usize,
    ) {
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    if let Some(underlyings) = underlyings {
                        let Some(symbol) = record.get(1) else {
                            records_scanned += 1;
                            continue;
                        };
                        let symbol = symbol.to_uppercase();
                        if !matches_underlying_filter(&symbol, Some(underlyings)) {
                            records_scanned += 1;
                            continue;
                        }
                    }

                    if let Ok(data) = record.deserialize::<TardisOptionsChainRecord>(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_scanned += 1;
                }
                Ok(false) => break,
                Err(_) => records_scanned += 1,
            }
        }
    }
}

impl Iterator for OptionsChainStreamIterator {
    type Item = anyhow::Result<Vec<Data>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.records_processed >= limit
        {
            return None;
        }

        self.buffer.clear();
        let mut records_read = 0;

        while records_read < self.chunk_size {
            match self.reader.read_record(&mut self.record) {
                Ok(true) => {
                    if let Some(underlyings) = self.underlyings.as_deref() {
                        let Some(symbol) = self.record.get(1) else {
                            continue;
                        };
                        let symbol = symbol.to_uppercase();
                        if !matches_underlying_filter(&symbol, Some(underlyings)) {
                            continue;
                        }
                    }

                    let data = match self.record.deserialize::<TardisOptionsChainRecord>(None) {
                        Ok(data) => data,
                        Err(e) => {
                            return Some(Err(anyhow::anyhow!("Failed to deserialize record: {e}")));
                        }
                    };
                    let instrument_id = parse_instrument_id(&data.exchange, data.symbol);
                    let precision = self
                        .precision_by_instrument
                        .entry(instrument_id)
                        .or_insert_with(|| {
                            OptionsChainPrecision::new(self.price_precision, self.size_precision)
                        });
                    precision.update(&data, self.price_precision, self.size_precision);

                    match parse_options_chain_record_as_quote(
                        &data,
                        precision.price,
                        precision.size,
                        instrument_id,
                    ) {
                        Ok(Some(quote)) => self.buffer.push(Data::Quote(quote)),
                        Ok(None) => {}
                        Err(e) => return Some(Err(e)),
                    }
                    self.buffer
                        .push(Data::OptionGreeks(parse_options_chain_record(
                            &data,
                            instrument_id,
                        )));

                    records_read += 1;
                    self.records_processed += 1;

                    if let Some(limit) = self.limit
                        && self.records_processed >= limit
                    {
                        break;
                    }
                }
                Ok(false) => {
                    if self.buffer.is_empty() {
                        return None;
                    }
                    return Some(Ok(self.buffer.clone()));
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            Some(Ok(self.buffer.clone()))
        }
    }
}

/// Streams Tardis `options_chain` CSV rows as quote and option greeks data.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference, later rows can raise the inferred precision for
/// their instrument after earlier chunks have already been emitted. Provide explicit precision
/// parameters for deterministic precision behavior.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_options_chain<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    underlyings: Option<Vec<String>>,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<Data>>>> {
    OptionsChainStreamIterator::new(
        filepath,
        chunk_size,
        underlyings,
        price_precision,
        size_precision,
        limit,
    )
}

////////////////////////////////////////////////////////////////////////////////
// Trade Streaming
////////////////////////////////////////////////////////////////////////////////

/// An iterator for streaming [`TradeTick`]s from a Tardis CSV file in chunks.
struct TradeStreamIterator {
    reader: Reader<Box<dyn Read>>,
    record: StringRecord,
    buffer: Vec<TradeTick>,
    chunk_size: usize,
    instrument_id: Option<InstrumentId>,
    price_precision: u8,
    size_precision: u8,
    limit: Option<usize>,
    records_processed: usize,
}

impl TradeStreamIterator {
    /// Creates a new [`TradeStreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read.
    pub(crate) fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let (final_price_precision, final_size_precision) =
            if let (Some(price_prec), Some(size_prec)) = (price_precision, size_precision) {
                // Both precisions provided, use them directly
                (price_prec, size_prec)
            } else {
                // One or both precisions missing, detect only the missing ones
                let mut reader = create_csv_reader(&filepath)?;
                let mut record = StringRecord::new();
                let (detected_price, detected_size) =
                    Self::detect_precision_from_sample(&mut reader, &mut record, 10_000);
                (
                    price_precision.unwrap_or(detected_price),
                    size_precision.unwrap_or(detected_size),
                )
            };

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size),
            chunk_size,
            instrument_id,
            price_precision: final_price_precision,
            size_precision: final_size_precision,
            limit,
            records_processed: 0,
        })
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn std::io::Read>>,
        record: &mut StringRecord,
        sample_size: usize,
    ) -> (u8, u8) {
        let mut max_price_precision = 2u8;
        let mut max_size_precision = 0u8;
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    if let Ok(data) = record.deserialize::<TardisTradeRecord>(None) {
                        max_price_precision = max_price_precision.max(infer_precision(data.price));
                        max_size_precision = max_size_precision.max(infer_precision(data.amount));
                        records_scanned += 1;
                    }
                }
                Ok(false) => break,             // End of file
                Err(_) => records_scanned += 1, // Skip malformed records
            }
        }

        (max_price_precision, max_size_precision)
    }
}

impl Iterator for TradeStreamIterator {
    type Item = anyhow::Result<Vec<TradeTick>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.records_processed >= limit
        {
            return None;
        }

        self.buffer.clear();
        let mut records_read = 0;

        while records_read < self.chunk_size {
            match self.reader.read_record(&mut self.record) {
                Ok(true) => match self.record.deserialize::<TardisTradeRecord>(None) {
                    Ok(data) => {
                        let size = Quantity::new(data.amount, self.size_precision);

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

                            self.buffer.push(trade);
                            records_read += 1;
                            self.records_processed += 1;

                            if let Some(limit) = self.limit
                                && self.records_processed >= limit
                            {
                                break;
                            }
                        } else {
                            log::warn!("Skipping zero-sized trade: {data:?}");
                        }
                    }
                    Err(e) => {
                        return Some(Err(anyhow::anyhow!("Failed to deserialize record: {e}")));
                    }
                },
                Ok(false) => {
                    if self.buffer.is_empty() {
                        return None;
                    }
                    return Some(Ok(self.buffer.clone()));
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            Some(Ok(self.buffer.clone()))
        }
    }
}

/// Streams [`TradeTick`]s from a Tardis format CSV at the given `filepath`,
/// yielding chunks of the specified size.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference (not providing explicit precisions),
/// the inferred precision may differ from bulk loading the entire file. This is because
/// precision inference works within chunk boundaries, and different chunks may contain
/// values with different precision requirements. For deterministic precision behavior,
/// provide explicit `price_precision` and `size_precision` parameters.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_trades<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<TradeTick>>>> {
    TradeStreamIterator::new(
        filepath,
        chunk_size,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

////////////////////////////////////////////////////////////////////////////////
// Depth10 Streaming
////////////////////////////////////////////////////////////////////////////////

/// An iterator for streaming [`OrderBookDepth10`]s from a Tardis CSV file in chunks.
struct Depth10StreamIterator {
    reader: Reader<Box<dyn Read>>,
    record: StringRecord,
    buffer: Vec<OrderBookDepth10>,
    chunk_size: usize,
    levels: u8,
    instrument_id: Option<InstrumentId>,
    price_precision: u8,
    size_precision: u8,
    limit: Option<usize>,
    records_processed: usize,
}

impl Depth10StreamIterator {
    /// Creates a new [`Depth10StreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read, or if `levels` is not 5 or 25.
    pub(crate) fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        levels: u8,
        price_precision: Option<u8>,
        size_precision: Option<u8>,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        anyhow::ensure!(
            levels == 5 || levels == 25,
            "Invalid levels: {levels}. Must be 5 or 25."
        );

        let (final_price_precision, final_size_precision) =
            if let (Some(price_prec), Some(size_prec)) = (price_precision, size_precision) {
                // Both precisions provided, use them directly
                (price_prec, size_prec)
            } else {
                // One or both precisions missing, detect only the missing ones
                let mut reader = create_csv_reader(&filepath)?;
                let mut record = StringRecord::new();
                let (detected_price, detected_size) =
                    Self::detect_precision_from_sample(&mut reader, &mut record, 10_000);
                (
                    price_precision.unwrap_or(detected_price),
                    size_precision.unwrap_or(detected_size),
                )
            };

        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size),
            chunk_size,
            levels,
            instrument_id,
            price_precision: final_price_precision,
            size_precision: final_size_precision,
            limit,
            records_processed: 0,
        })
    }

    fn process_snapshot5(&self, data: &TardisOrderBookSnapshot5Record) -> OrderBookDepth10 {
        let instrument_id = self
            .instrument_id
            .unwrap_or_else(|| parse_instrument_id(&data.exchange, data.symbol));

        let mut bids = [NULL_ORDER; DEPTH10_LEN];
        let mut asks = [NULL_ORDER; DEPTH10_LEN];
        let mut bid_counts = [0_u32; DEPTH10_LEN];
        let mut ask_counts = [0_u32; DEPTH10_LEN];

        // Process first 5 levels from snapshot5 data
        for i in 0..5 {
            let (bid_price, bid_amount) = match i {
                0 => (data.bids_0_price, data.bids_0_amount),
                1 => (data.bids_1_price, data.bids_1_amount),
                2 => (data.bids_2_price, data.bids_2_amount),
                3 => (data.bids_3_price, data.bids_3_amount),
                4 => (data.bids_4_price, data.bids_4_amount),
                _ => unreachable!(),
            };

            let (ask_price, ask_amount) = match i {
                0 => (data.asks_0_price, data.asks_0_amount),
                1 => (data.asks_1_price, data.asks_1_amount),
                2 => (data.asks_2_price, data.asks_2_amount),
                3 => (data.asks_3_price, data.asks_3_amount),
                4 => (data.asks_4_price, data.asks_4_amount),
                _ => unreachable!(),
            };

            let (bid_order, bid_count) = create_book_order(
                OrderSide::Buy,
                bid_price,
                bid_amount,
                self.price_precision,
                self.size_precision,
            );
            bids[i] = bid_order;
            bid_counts[i] = bid_count;

            let (ask_order, ask_count) = create_book_order(
                OrderSide::Sell,
                ask_price,
                ask_amount,
                self.price_precision,
                self.size_precision,
            );
            asks[i] = ask_order;
            ask_counts[i] = ask_count;
        }

        let flags = RecordFlag::F_SNAPSHOT as u8;
        let sequence = 0;
        let ts_event = parse_timestamp(data.timestamp);
        let ts_init = parse_timestamp(data.local_timestamp);

        OrderBookDepth10::new(
            instrument_id,
            bids,
            asks,
            bid_counts,
            ask_counts,
            flags,
            sequence,
            ts_event,
            ts_init,
        )
    }

    fn process_snapshot25(&self, data: &TardisOrderBookSnapshot25Record) -> OrderBookDepth10 {
        let instrument_id = self
            .instrument_id
            .unwrap_or_else(|| parse_instrument_id(&data.exchange, data.symbol));

        let mut bids = [NULL_ORDER; DEPTH10_LEN];
        let mut asks = [NULL_ORDER; DEPTH10_LEN];
        let mut bid_counts = [0_u32; DEPTH10_LEN];
        let mut ask_counts = [0_u32; DEPTH10_LEN];

        // Process first 10 levels from snapshot25 data
        for i in 0..DEPTH10_LEN {
            let (bid_price, bid_amount) = match i {
                0 => (data.bids_0_price, data.bids_0_amount),
                1 => (data.bids_1_price, data.bids_1_amount),
                2 => (data.bids_2_price, data.bids_2_amount),
                3 => (data.bids_3_price, data.bids_3_amount),
                4 => (data.bids_4_price, data.bids_4_amount),
                5 => (data.bids_5_price, data.bids_5_amount),
                6 => (data.bids_6_price, data.bids_6_amount),
                7 => (data.bids_7_price, data.bids_7_amount),
                8 => (data.bids_8_price, data.bids_8_amount),
                9 => (data.bids_9_price, data.bids_9_amount),
                _ => unreachable!(),
            };

            let (ask_price, ask_amount) = match i {
                0 => (data.asks_0_price, data.asks_0_amount),
                1 => (data.asks_1_price, data.asks_1_amount),
                2 => (data.asks_2_price, data.asks_2_amount),
                3 => (data.asks_3_price, data.asks_3_amount),
                4 => (data.asks_4_price, data.asks_4_amount),
                5 => (data.asks_5_price, data.asks_5_amount),
                6 => (data.asks_6_price, data.asks_6_amount),
                7 => (data.asks_7_price, data.asks_7_amount),
                8 => (data.asks_8_price, data.asks_8_amount),
                9 => (data.asks_9_price, data.asks_9_amount),
                _ => unreachable!(),
            };

            let (bid_order, bid_count) = create_book_order(
                OrderSide::Buy,
                bid_price,
                bid_amount,
                self.price_precision,
                self.size_precision,
            );
            bids[i] = bid_order;
            bid_counts[i] = bid_count;

            let (ask_order, ask_count) = create_book_order(
                OrderSide::Sell,
                ask_price,
                ask_amount,
                self.price_precision,
                self.size_precision,
            );
            asks[i] = ask_order;
            ask_counts[i] = ask_count;
        }

        let flags = RecordFlag::F_SNAPSHOT as u8;
        let sequence = 0;
        let ts_event = parse_timestamp(data.timestamp);
        let ts_init = parse_timestamp(data.local_timestamp);

        OrderBookDepth10::new(
            instrument_id,
            bids,
            asks,
            bid_counts,
            ask_counts,
            flags,
            sequence,
            ts_event,
            ts_init,
        )
    }

    fn detect_precision_from_sample(
        reader: &mut Reader<Box<dyn std::io::Read>>,
        record: &mut StringRecord,
        sample_size: usize,
    ) -> (u8, u8) {
        let mut max_price_precision = 2u8;
        let mut max_size_precision = 0u8;
        let mut records_scanned = 0;

        while records_scanned < sample_size {
            match reader.read_record(record) {
                Ok(true) => {
                    // Try to deserialize as snapshot5 record first
                    if let Ok(data) = record.deserialize::<TardisOrderBookSnapshot5Record>(None) {
                        if let Some(bid_price) = data.bids_0_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(bid_price));
                        }

                        if let Some(ask_price) = data.asks_0_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(ask_price));
                        }

                        if let Some(bid_amount) = data.bids_0_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(bid_amount));
                        }

                        if let Some(ask_amount) = data.asks_0_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(ask_amount));
                        }
                        records_scanned += 1;
                    } else if let Ok(data) =
                        record.deserialize::<TardisOrderBookSnapshot25Record>(None)
                    {
                        if let Some(bid_price) = data.bids_0_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(bid_price));
                        }

                        if let Some(ask_price) = data.asks_0_price {
                            max_price_precision =
                                max_price_precision.max(infer_precision(ask_price));
                        }

                        if let Some(bid_amount) = data.bids_0_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(bid_amount));
                        }

                        if let Some(ask_amount) = data.asks_0_amount {
                            max_size_precision =
                                max_size_precision.max(infer_precision(ask_amount));
                        }
                        records_scanned += 1;
                    }
                }
                Ok(false) => break,             // End of file
                Err(_) => records_scanned += 1, // Skip malformed records
            }
        }

        (max_price_precision, max_size_precision)
    }
}

impl Iterator for Depth10StreamIterator {
    type Item = anyhow::Result<Vec<OrderBookDepth10>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.records_processed >= limit
        {
            return None;
        }

        if !self.buffer.is_empty() {
            let chunk = self.buffer.split_off(0);
            return Some(Ok(chunk));
        }

        self.buffer.clear();
        let mut records_read = 0;

        while records_read < self.chunk_size {
            match self.reader.read_record(&mut self.record) {
                Ok(true) => {
                    let result = match self.levels {
                        5 => self
                            .record
                            .deserialize::<TardisOrderBookSnapshot5Record>(None)
                            .map(|data| self.process_snapshot5(&data)),
                        25 => self
                            .record
                            .deserialize::<TardisOrderBookSnapshot25Record>(None)
                            .map(|data| self.process_snapshot25(&data)),
                        _ => return Some(Err(anyhow::anyhow!("Invalid levels: {}", self.levels))),
                    };

                    match result {
                        Ok(depth) => {
                            self.buffer.push(depth);
                            records_read += 1;
                            self.records_processed += 1;

                            if let Some(limit) = self.limit
                                && self.records_processed >= limit
                            {
                                break;
                            }
                        }
                        Err(e) => {
                            return Some(Err(anyhow::anyhow!("Failed to deserialize record: {e}")));
                        }
                    }
                }
                Ok(false) => {
                    if self.buffer.is_empty() {
                        return None;
                    }
                    let chunk = self.buffer.split_off(0);
                    return Some(Ok(chunk));
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            let chunk = self.buffer.split_off(0);
            Some(Ok(chunk))
        }
    }
}

/// Streams [`OrderBookDepth10`]s from a Tardis format CSV at the given `filepath`,
/// yielding chunks of the specified size.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference (not providing explicit precisions),
/// the inferred precision may differ from bulk loading the entire file. This is because
/// precision inference works within chunk boundaries, and different chunks may contain
/// values with different precision requirements. For deterministic precision behavior,
/// provide explicit `price_precision` and `size_precision` parameters.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_depth10_from_snapshot5<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<OrderBookDepth10>>>> {
    Depth10StreamIterator::new(
        filepath,
        chunk_size,
        5,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

/// Streams [`OrderBookDepth10`]s from a Tardis format CSV at the given `filepath`,
/// yielding chunks of the specified size.
///
/// # Precision Inference Warning
///
/// When using streaming with precision inference (not providing explicit precisions),
/// the inferred precision may differ from bulk loading the entire file. This is because
/// precision inference works within chunk boundaries, and different chunks may contain
/// values with different precision requirements. For deterministic precision behavior,
/// provide explicit `price_precision` and `size_precision` parameters.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, read, or parsed as CSV.
pub fn stream_depth10_from_snapshot25<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    price_precision: Option<u8>,
    size_precision: Option<u8>,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<OrderBookDepth10>>>> {
    Depth10StreamIterator::new(
        filepath,
        chunk_size,
        25,
        price_precision,
        size_precision,
        instrument_id,
        limit,
    )
}

////////////////////////////////////////////////////////////////////////////////
// FundingRateUpdate Streaming
////////////////////////////////////////////////////////////////////////////////

use nautilus_model::data::FundingRateUpdate;

use crate::csv::record::TardisDerivativeTickerRecord;

/// An iterator for streaming [`FundingRateUpdate`]s from a Tardis CSV file in chunks.
struct FundingRateStreamIterator {
    reader: Reader<Box<dyn Read>>,
    record: StringRecord,
    buffer: Vec<FundingRateUpdate>,
    chunk_size: usize,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
    records_processed: usize,
}

impl FundingRateStreamIterator {
    /// Creates a new [`FundingRateStreamIterator`].
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or read.
    fn new<P: AsRef<Path>>(
        filepath: P,
        chunk_size: usize,
        instrument_id: Option<InstrumentId>,
        limit: Option<usize>,
    ) -> anyhow::Result<Self> {
        let reader = create_csv_reader(filepath)?;

        Ok(Self {
            reader,
            record: StringRecord::new(),
            buffer: Vec::with_capacity(chunk_size),
            chunk_size,
            instrument_id,
            limit,
            records_processed: 0,
        })
    }
}

impl Iterator for FundingRateStreamIterator {
    type Item = anyhow::Result<Vec<FundingRateUpdate>>;

    fn next(&mut self) -> Option<Self::Item> {
        if let Some(limit) = self.limit
            && self.records_processed >= limit
        {
            return None;
        }

        if !self.buffer.is_empty() {
            let chunk = self.buffer.split_off(0);
            return Some(Ok(chunk));
        }

        self.buffer.clear();
        let mut records_read = 0;

        while records_read < self.chunk_size {
            match self.reader.read_record(&mut self.record) {
                Ok(true) => {
                    let result = self
                        .record
                        .deserialize::<TardisDerivativeTickerRecord>(None)
                        .map_err(anyhow::Error::from)
                        .map(|data| parse_derivative_ticker_record(&data, self.instrument_id));

                    match result {
                        Ok(Some(funding_rate)) => {
                            self.buffer.push(funding_rate);
                            records_read += 1;
                            self.records_processed += 1;

                            if let Some(limit) = self.limit
                                && self.records_processed >= limit
                            {
                                break;
                            }
                        }
                        Ok(None) => {
                            // Skip this record as it has no funding data
                            self.records_processed += 1;
                        }
                        Err(e) => {
                            return Some(Err(anyhow::anyhow!(
                                "Failed to parse funding rate record: {e}"
                            )));
                        }
                    }
                }
                Ok(false) => {
                    if self.buffer.is_empty() {
                        return None;
                    }
                    let chunk = self.buffer.split_off(0);
                    return Some(Ok(chunk));
                }
                Err(e) => return Some(Err(anyhow::anyhow!("Failed to read record: {e}"))),
            }
        }

        if self.buffer.is_empty() {
            None
        } else {
            let chunk = self.buffer.split_off(0);
            Some(Ok(chunk))
        }
    }
}

/// Streams [`FundingRateUpdate`]s from a Tardis derivative ticker CSV file,
/// yielding chunks of the specified size.
///
/// 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 stream_funding_rates<P: AsRef<Path>>(
    filepath: P,
    chunk_size: usize,
    instrument_id: Option<InstrumentId>,
    limit: Option<usize>,
) -> anyhow::Result<impl Iterator<Item = anyhow::Result<Vec<FundingRateUpdate>>>> {
    FundingRateStreamIterator::new(filepath, chunk_size, instrument_id, limit)
}

#[cfg(test)]
mod tests {
    use nautilus_model::{
        enums::{AggressorSide, BookAction},
        identifiers::{InstrumentId, TradeId},
        types::Price,
    };
    use rstest::*;

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

    #[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_stream_deltas_chunked() {
        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_stream_deltas.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let stream = stream_deltas(&temp_file, 2, Some(4), Some(1), None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        // 5 data rows + 1 CLEAR = 6 deltas, in chunks of 2
        assert_eq!(chunks.len(), 3);

        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);
        assert_eq!(chunk1[0].action, BookAction::Clear); // CLEAR first
        assert_eq!(chunk1[1].order.price.precision, 4); // First data delta

        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 2);
        assert_eq!(chunk2[0].order.price.precision, 4);
        assert_eq!(chunk2[1].order.price.precision, 4);

        let chunk3 = chunks[2].as_ref().unwrap();
        assert_eq!(chunk3.len(), 2);
        assert_eq!(chunk3[0].order.price.precision, 4);
        assert_eq!(chunk3[1].order.price.precision, 4);

        let total_deltas: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_deltas, 6);

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

    #[rstest]
    #[case(2, vec![2, 2])]
    #[case(3, vec![3, 1])]
    fn test_stream_deltas_groups_messages_by_local_timestamp(
        #[case] chunk_size: usize,
        #[case] expected_chunk_lengths: Vec<usize>,
    ) {
        let filepath = get_test_data_path("csv/deltas_message_boundaries.csv");
        let expected = load_deltas(&filepath, Some(1), Some(1), None, None).unwrap();
        let chunks = stream_deltas(&filepath, chunk_size, Some(1), Some(1), None, None)
            .unwrap()
            .map(Result::unwrap)
            .collect::<Vec<_>>();

        assert_eq!(
            chunks.iter().map(Vec::len).collect::<Vec<_>>(),
            expected_chunk_lengths
        );
        assert_eq!(chunks.into_iter().flatten().collect::<Vec<_>>(), expected);
    }

    #[rstest]
    fn test_stream_deltas_defers_lookahead_error() {
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
deribit,BTC-PERPETUAL,1000,2000,false,bid,100.0,1.0
deribit,BTC-PERPETUAL,1000,2000,false,ask,101.0,2.0
deribit,BTC-PERPETUAL,invalid,2010,false,bid,99.0,3.0";
        let temp_file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(temp_file.path(), csv_data).unwrap();
        let mut stream = stream_deltas(temp_file.path(), 2, Some(1), Some(1), None, None).unwrap();

        let chunk = stream.next().unwrap().unwrap();
        let error = stream.next().unwrap().unwrap_err();

        assert_eq!(chunk.len(), 2);
        assert_eq!(chunk[0].order.price, Price::from("100.0"));
        assert_eq!(chunk[1].order.price, Price::from("101.0"));
        assert_eq!(chunk[1].flags, 0);
        assert!(
            error
                .to_string()
                .starts_with("Failed to deserialize record: CSV deserialize error:")
        );
        assert!(stream.next().is_none());
    }

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

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

        // limit=1 should return only the synthetic CLEAR delta
        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 10, Some(4), Some(1), None, Some(1))
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();
        assert_eq!(iterator.pending_batches.len(), 1);
        assert_eq!(iterator.pending_batches[0].len(), 1);
        assert_eq!(iterator.pending_batches[0][0].action, BookAction::Clear);

        // No limit should return all batches (first batch starts with CLEAR)
        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 10, Some(4), Some(1), None, None).unwrap();
        iterator.fill_pending_batches().transpose().unwrap();
        assert_eq!(iterator.pending_batches.len(), 5);
        assert_eq!(iterator.pending_batches[0].len(), 2);
        assert_eq!(iterator.pending_batches[0][0].action, BookAction::Clear);
        assert_ne!(iterator.pending_batches[0][1].action, BookAction::Clear);
        let total_deltas: usize = iterator
            .pending_batches
            .iter()
            .map(|batch| batch.len())
            .sum();
        assert_eq!(total_deltas, 6);

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

    #[cfg(feature = "python")]
    #[rstest]
    fn test_stream_batched_deltas_groups_messages_by_local_timestamp() {
        let filepath = get_test_data_path("csv/deltas_message_boundaries.csv");
        let expected = load_deltas(&filepath, Some(1), Some(1), None, None).unwrap();
        let mut iterator =
            BatchedDeltasStreamIterator::new(&filepath, 100, Some(1), Some(1), None, None).unwrap();

        iterator.fill_pending_batches().transpose().unwrap();

        assert_eq!(
            iterator
                .pending_batches
                .iter()
                .map(Vec::len)
                .collect::<Vec<_>>(),
            vec![2, 2]
        );
        assert_eq!(
            iterator
                .pending_batches
                .into_iter()
                .flatten()
                .collect::<Vec<_>>(),
            expected
        );
    }

    #[cfg(feature = "python")]
    #[rstest]
    fn test_stream_batched_deltas_returns_python_objects() {
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,true,ask,50000.0,1.0
binance,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.5,2.0";

        let temp_file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(temp_file.path(), csv_data).unwrap();
        Python::initialize();

        let mut stream =
            stream_batched_deltas(temp_file.path(), 10, Some(1), Some(1), None, None).unwrap();
        let objects = stream.next().unwrap().unwrap();

        Python::attach(|py| {
            let deltas: Vec<_> = objects
                .iter()
                .map(|obj| {
                    obj.bind(py)
                        .cast::<OrderBookDeltas>()
                        .unwrap()
                        .borrow()
                        .clone()
                })
                .collect();

            assert_eq!(deltas.len(), 2);
            assert_eq!(deltas[0].deltas.len(), 2);
            assert_eq!(deltas[0].deltas[0].action, BookAction::Clear);
            assert_eq!(deltas[0].deltas[1].action, BookAction::Add);
            assert_eq!(deltas[1].deltas.len(), 1);
            assert_eq!(deltas[1].deltas[0].action, BookAction::Update);
        });
        assert!(stream.next().is_none());
    }

    #[cfg(feature = "python")]
    #[rstest]
    pub fn test_stream_batched_deltas_with_mid_snapshot_inserts_clear() {
        // CSV with:
        // - Initial snapshot (is_snapshot=true) at start
        // - Some deltas (is_snapshot=false)
        // - Mid-day snapshot (is_snapshot=true) - should trigger CLEAR
        // - Back to deltas (is_snapshot=false)
        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,1640995300000000,1640995300100000,true,bid,50100.0,3.0
binance-futures,BTCUSDT,1640995300000000,1640995300100000,true,ask,50101.0,4.0
binance-futures,BTCUSDT,1640995301000000,1640995301100000,false,bid,50099.0,1.0";

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

        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 100, Some(1), Some(1), None, None)
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();

        let all_deltas: Vec<_> = iterator.pending_batches.iter().flatten().collect();
        let clear_count = all_deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        // Should have 2 CLEAR deltas: initial snapshot + mid-day snapshot
        assert_eq!(
            clear_count, 2,
            "Expected 2 CLEAR deltas (initial + mid-day snapshot), found {clear_count}"
        );

        // Verify CLEAR positions:
        // 0=CLEAR, 1=Add, 2=Add, 3=Update, 4=Update, 5=CLEAR, 6=Add, 7=Add, 8=Update
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[5].action, BookAction::Clear);

        // CLEAR deltas should NOT have F_LAST when followed by same-timestamp deltas
        assert_eq!(
            all_deltas[0].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 0 should not have F_LAST flag"
        );
        assert_eq!(
            all_deltas[5].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 5 should not have F_LAST flag"
        );

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

    #[cfg(feature = "python")]
    #[rstest]
    pub fn test_stream_batched_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_stream_batched_consecutive_snapshots.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 100, Some(1), Some(1), None, None)
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();

        let all_deltas: Vec<_> = iterator.pending_batches.iter().flatten().collect();
        let clear_count = all_deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        assert_eq!(clear_count, 2);
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[3].action, BookAction::Clear);
        assert_eq!(
            all_deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
        assert_eq!(all_deltas[3].flags & RecordFlag::F_LAST as u8, 0);
        assert_eq!(
            all_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();
    }

    #[cfg(feature = "python")]
    #[rstest]
    pub fn test_stream_batched_deltas_limit_includes_clear() {
        // Test that limit counts total emitted deltas (including CLEARs)
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
binance,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.0,0.5
binance,BTCUSDT,1640995202000000,1640995202100000,false,ask,50002.0,1.5
binance,BTCUSDT,1640995203000000,1640995203100000,false,bid,49998.0,0.5
binance,BTCUSDT,1640995204000000,1640995204100000,false,ask,50003.0,1.0";

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

        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 100, Some(1), Some(1), None, Some(4))
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();

        let all_deltas: Vec<_> = iterator.pending_batches.iter().flatten().collect();

        // limit=4 should get exactly 4 deltas: 1 CLEAR + 3 data deltas
        assert_eq!(all_deltas.len(), 4);
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[1].action, BookAction::Add);
        assert_eq!(all_deltas[2].action, BookAction::Update);
        assert_eq!(all_deltas[3].action, BookAction::Update);

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

    #[cfg(feature = "python")]
    #[rstest]
    pub fn test_stream_batched_deltas_limit_sets_f_last() {
        // Test that F_LAST is set on the final delta when limit is reached
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
binance,BTCUSDT,1640995201000000,1640995201100000,false,ask,50001.0,2.0
binance,BTCUSDT,1640995202000000,1640995202100000,false,bid,49999.0,0.5
binance,BTCUSDT,1640995203000000,1640995203100000,false,ask,50002.0,1.5";

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

        // limit=3 should get 3 deltas with F_LAST on the last one
        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 100, Some(1), Some(1), None, Some(3))
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();

        let all_deltas: Vec<_> = iterator.pending_batches.iter().flatten().collect();

        assert_eq!(all_deltas.len(), 3);
        assert_eq!(
            all_deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8,
            "Final delta should have F_LAST flag when limit is reached"
        );

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

    #[cfg(feature = "python")]
    #[rstest]
    pub fn test_stream_batched_deltas_snapshot_batch_flags() {
        // Test that CLEAR is first in batch and only the last delta has F_LAST
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
binance,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0
binance,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.0,0.5";

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

        let mut iterator =
            BatchedDeltasStreamIterator::new(&temp_file, 100, Some(1), Some(1), None, None)
                .unwrap();
        iterator.fill_pending_batches().transpose().unwrap();

        assert_eq!(iterator.pending_batches.len(), 2);
        let first_batch = &iterator.pending_batches[0];

        // First batch contains CLEAR + 2 snapshot deltas
        assert_eq!(first_batch.len(), 3);
        assert_eq!(first_batch[0].action, BookAction::Clear);
        assert_eq!(first_batch[0].flags & RecordFlag::F_LAST as u8, 0);
        assert_eq!(first_batch[1].flags & RecordFlag::F_LAST as u8, 0);
        assert_eq!(
            first_batch[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );

        // Second batch should have F_LAST set (end of file)
        assert_eq!(iterator.pending_batches[1].len(), 1);
        assert_eq!(
            iterator.pending_batches[1][0].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );

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

    #[rstest]
    pub fn test_stream_quotes_chunked() {
        let csv_data =
            "exchange,symbol,timestamp,local_timestamp,ask_amount,ask_price,bid_price,bid_amount
binance,BTCUSDT,1640995200000000,1640995200100000,1.0,50000.0,49999.0,1.5
binance,BTCUSDT,1640995201000000,1640995201100000,2.0,50000.5,49999.5,2.5
binance,BTCUSDT,1640995202000000,1640995202100000,1.5,50000.12,49999.12,1.8
binance,BTCUSDT,1640995203000000,1640995203100000,3.0,50000.123,49999.123,3.2
binance,BTCUSDT,1640995204000000,1640995204100000,0.5,50000.1234,49999.1234,0.8";

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

        let stream = stream_quotes(&temp_file, 2, Some(4), Some(1), None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

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

        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);
        assert_eq!(chunk1[0].bid_price.precision, 4);
        assert_eq!(chunk1[1].bid_price.precision, 4);

        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 2);
        assert_eq!(chunk2[0].bid_price.precision, 4);
        assert_eq!(chunk2[1].bid_price.precision, 4);

        let chunk3 = chunks[2].as_ref().unwrap();
        assert_eq!(chunk3.len(), 1);
        assert_eq!(chunk3[0].bid_price.precision, 4);

        let total_quotes: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_quotes, 5);

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

    #[rstest]
    pub fn test_stream_options_chain_filters_and_chunks() {
        let filepath = get_test_data_path("options_chain.csv");
        let stream = stream_options_chain(
            filepath,
            1,
            Some(vec!["ETH-".to_string()]),
            None,
            None,
            None,
        )
        .unwrap();
        let chunks: Vec<_> = stream.collect();

        assert_eq!(chunks.len(), 2);

        let first_chunk = chunks[0].as_ref().unwrap();
        assert_eq!(first_chunk.len(), 2);
        let Data::Quote(quote) = &first_chunk[0] else {
            panic!("Expected first data item to be Quote");
        };
        let Data::OptionGreeks(greeks) = &first_chunk[1] else {
            panic!("Expected second data item to be OptionGreeks");
        };

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

        let second_chunk = chunks[1].as_ref().unwrap();
        assert_eq!(second_chunk.len(), 1);
        assert!(matches!(second_chunk[0], Data::OptionGreeks(_)));
    }

    #[rstest]
    pub fn test_stream_trades_chunked() {
        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,2.0
binance,BTCUSDT,1640995202000000,1640995202100000,trade3,buy,50000.12,1.5
binance,BTCUSDT,1640995203000000,1640995203100000,trade4,sell,49999.123,3.0
binance,BTCUSDT,1640995204000000,1640995204100000,trade5,buy,50000.1234,0.5";

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

        let stream = stream_trades(&temp_file, 3, Some(4), Some(1), None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        assert_eq!(chunks.len(), 2);

        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 3);
        assert_eq!(chunk1[0].price.precision, 4);
        assert_eq!(chunk1[1].price.precision, 4);
        assert_eq!(chunk1[2].price.precision, 4);

        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 2);
        assert_eq!(chunk2[0].price.precision, 4);
        assert_eq!(chunk2[1].price.precision, 4);

        assert_eq!(chunk1[0].aggressor_side, AggressorSide::Buy);
        assert_eq!(chunk1[1].aggressor_side, AggressorSide::Sell);

        let total_trades: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_trades, 5);

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

    #[rstest]
    pub fn test_stream_trades_with_zero_sized_trade() {
        // 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_stream_trades_zero_size.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let stream = stream_trades(&temp_file, 3, Some(4), Some(1), None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 1 chunk with 3 valid trades (zero-sized trade skipped)
        assert_eq!(chunks.len(), 1);

        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 3);

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

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

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

    #[rstest]
    pub fn test_stream_depth10_from_snapshot5_chunked() {
        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,BTCUSDT,1640995200000000,1640995200100000,50001.0,1.0,49999.0,1.5,50002.0,2.0,49998.0,2.5,50003.0,3.0,49997.0,3.5,50004.0,4.0,49996.0,4.5,50005.0,5.0,49995.0,5.5
binance,BTCUSDT,1640995201000000,1640995201100000,50001.5,1.1,49999.5,1.6,50002.5,2.1,49998.5,2.6,50003.5,3.1,49997.5,3.6,50004.5,4.1,49996.5,4.6,50005.5,5.1,49995.5,5.6
binance,BTCUSDT,1640995202000000,1640995202100000,50001.12,1.12,49999.12,1.62,50002.12,2.12,49998.12,2.62,50003.12,3.12,49997.12,3.62,50004.12,4.12,49996.12,4.62,50005.12,5.12,49995.12,5.62";

        // Write to temporary file
        let temp_file = std::env::temp_dir().join("test_stream_depth10_snapshot5.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        // Stream with chunk size of 2
        let stream = stream_depth10_from_snapshot5(&temp_file, 2, None, None, None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 2 chunks: [2 items, 1 item]
        assert_eq!(chunks.len(), 2);

        // First chunk: 2 depth snapshots
        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);

        // Second chunk: 1 depth snapshot
        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 1);

        // Verify depth structure
        let first_depth = &chunk1[0];
        assert_eq!(first_depth.bids.len(), 10); // Should have 10 levels
        assert_eq!(first_depth.asks.len(), 10);

        // Verify some specific prices
        assert_eq!(first_depth.bids[0].price, parse_price(49999.0, 1));
        assert_eq!(first_depth.asks[0].price, parse_price(50001.0, 1));

        // Verify total count
        let total_depths: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_depths, 3);

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

    #[rstest]
    pub fn test_stream_depth10_from_snapshot25_chunked() {
        // Create minimal snapshot25 CSV data (first 10 levels only for testing)
        let mut header_parts = vec!["exchange", "symbol", "timestamp", "local_timestamp"];

        // Add bid and ask levels (we'll only populate first few for testing)
        let mut bid_headers = Vec::new();
        let mut ask_headers = Vec::new();

        for i in 0..25 {
            bid_headers.push(format!("bids[{i}].price"));
            bid_headers.push(format!("bids[{i}].amount"));
        }

        for i in 0..25 {
            ask_headers.push(format!("asks[{i}].price"));
            ask_headers.push(format!("asks[{i}].amount"));
        }

        for header in &bid_headers {
            header_parts.push(header);
        }

        for header in &ask_headers {
            header_parts.push(header);
        }

        let header = header_parts.join(",");

        // Create a row with data for first 5 levels (rest will be empty)
        let mut row1_parts = vec![
            "binance".to_string(),
            "BTCUSDT".to_string(),
            "1640995200000000".to_string(),
            "1640995200100000".to_string(),
        ];

        // Add bid levels (first 5 with data, rest empty)
        for i in 0..25 {
            if i < 5 {
                let bid_price = f64::from(i).mul_add(-0.01, 49999.0);
                let bid_amount = 1.0 + f64::from(i);
                row1_parts.push(bid_price.to_string());
                row1_parts.push(bid_amount.to_string());
            } else {
                row1_parts.push(String::new());
                row1_parts.push(String::new());
            }
        }

        // Add ask levels (first 5 with data, rest empty)
        for i in 0..25 {
            if i < 5 {
                let ask_price = f64::from(i).mul_add(0.01, 50000.0);
                let ask_amount = 1.0 + f64::from(i);
                row1_parts.push(ask_price.to_string());
                row1_parts.push(ask_amount.to_string());
            } else {
                row1_parts.push(String::new());
                row1_parts.push(String::new());
            }
        }

        let csv_data = format!("{}\n{}", header, row1_parts.join(","));

        // Write to temporary file
        let temp_file = std::env::temp_dir().join("test_stream_depth10_snapshot25.csv");
        std::fs::write(&temp_file, &csv_data).unwrap();

        // Stream with chunk size of 1
        let stream = stream_depth10_from_snapshot25(&temp_file, 1, None, None, None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 1 chunk with 1 item
        assert_eq!(chunks.len(), 1);

        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 1);

        // Verify depth structure
        let depth = &chunk1[0];
        assert_eq!(depth.bids.len(), 10); // Should have 10 levels
        assert_eq!(depth.asks.len(), 10);

        // Verify first level has data - check whatever we actually get
        let actual_bid_price = depth.bids[0].price;
        let actual_ask_price = depth.asks[0].price;
        assert!(actual_bid_price.as_f64() > 0.0);
        assert!(actual_ask_price.as_f64() > 0.0);

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

    #[rstest]
    pub fn test_stream_error_handling() {
        // Test with non-existent file
        let non_existent = std::path::Path::new("does_not_exist.csv");

        let result = stream_deltas(non_existent, 10, None, None, None, None);
        assert!(result.is_err());

        let result = stream_quotes(non_existent, 10, None, None, None, None);
        assert!(result.is_err());

        let result = stream_trades(non_existent, 10, None, None, None, None);
        assert!(result.is_err());

        let result = stream_depth10_from_snapshot5(non_existent, 10, None, None, None, None);
        assert!(result.is_err());

        let result = stream_depth10_from_snapshot25(non_existent, 10, None, None, None, None);
        assert!(result.is_err());
    }

    #[rstest]
    pub fn test_stream_empty_file() {
        // Test with empty CSV file
        let temp_file = std::env::temp_dir().join("test_empty.csv");
        std::fs::write(&temp_file, "").unwrap();

        let stream = stream_deltas(&temp_file, 10, None, None, None, None).unwrap();
        assert_eq!(stream.count(), 0);

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

    #[rstest]
    pub fn test_stream_precision_consistency() {
        // Test that streaming produces same results as bulk loading for 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";

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

        // Load all at once
        let bulk_deltas = load_deltas(&temp_file, None, None, None, None).unwrap();

        // Stream in chunks and collect
        let stream = stream_deltas(&temp_file, 2, None, None, None, None).unwrap();
        let streamed_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();

        // Should have same number of deltas
        assert_eq!(bulk_deltas.len(), streamed_deltas.len());

        // Compare key properties (precision inference will be different due to chunking)
        for (bulk, streamed) in bulk_deltas.iter().zip(streamed_deltas.iter()) {
            assert_eq!(bulk.instrument_id, streamed.instrument_id);
            assert_eq!(bulk.action, streamed.action);
            assert_eq!(bulk.order.side, streamed.order.side);
            assert_eq!(bulk.ts_event, streamed.ts_event);
            assert_eq!(bulk.ts_init, streamed.ts_init);
            // Note: precision may differ between bulk and streaming due to chunk boundaries
        }

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

    #[rstest]
    pub fn test_stream_trades_from_local_file() {
        let filepath = get_test_data_path("csv/trades_1.csv");
        let mut stream = stream_trades(filepath, 1, Some(1), Some(0), None, None).unwrap();

        let chunk1 = stream.next().unwrap().unwrap();
        assert_eq!(chunk1.len(), 1);
        assert_eq!(chunk1[0].price, Price::from("8531.5"));

        let chunk2 = stream.next().unwrap().unwrap();
        assert_eq!(chunk2.len(), 1);
        assert_eq!(chunk2[0].size, Quantity::from("1000"));

        assert!(stream.next().is_none());
    }

    #[rstest]
    pub fn test_stream_deltas_from_local_file() {
        let filepath = get_test_data_path("csv/deltas_1.csv");
        let mut stream = stream_deltas(filepath, 1, Some(1), Some(0), None, None).unwrap();

        // With chunk_size=1, each delta gets its own chunk
        // First chunk: CLEAR
        let chunk1 = stream.next().unwrap().unwrap();
        assert_eq!(chunk1.len(), 1);
        assert_eq!(chunk1[0].action, BookAction::Clear);

        // Second chunk: first data delta
        let chunk2 = stream.next().unwrap().unwrap();
        assert_eq!(chunk2.len(), 1);
        assert_eq!(chunk2[0].order.price, Price::from("6421.5"));

        // Third chunk: second data delta
        let chunk3 = stream.next().unwrap().unwrap();
        assert_eq!(chunk3.len(), 1);
        assert_eq!(chunk3[0].order.size, Quantity::from("10000"));

        assert!(stream.next().is_none());
    }

    #[rstest]
    pub fn test_stream_deltas_with_limit() {
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance,BTCUSDT,1640995200000000,1640995200100000,false,bid,50000.0,1.0
binance,BTCUSDT,1640995201000000,1640995201100000,false,ask,50001.0,2.0
binance,BTCUSDT,1640995202000000,1640995202100000,false,bid,49999.0,1.5
binance,BTCUSDT,1640995203000000,1640995203100000,false,ask,50002.0,3.0
binance,BTCUSDT,1640995204000000,1640995204100000,false,bid,49998.0,0.5";

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

        // Test with limit of 3 records
        let stream = stream_deltas(&temp_file, 2, Some(4), Some(1), None, Some(3)).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 2 chunks: [2 items, 1 item] = 3 total (limited)
        assert_eq!(chunks.len(), 2);
        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);
        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 1);

        // Total should be exactly 3 records due to limit
        let total_deltas: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_deltas, 3);

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

    #[rstest]
    pub fn test_stream_quotes_with_limit() {
        let csv_data =
            "exchange,symbol,timestamp,local_timestamp,ask_price,ask_amount,bid_price,bid_amount
binance,BTCUSDT,1640995200000000,1640995200100000,50001.0,1.0,50000.0,1.5
binance,BTCUSDT,1640995201000000,1640995201100000,50002.0,2.0,49999.0,2.5
binance,BTCUSDT,1640995202000000,1640995202100000,50003.0,1.5,49998.0,3.0
binance,BTCUSDT,1640995203000000,1640995203100000,50004.0,3.0,49997.0,3.5";

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

        // Test with limit of 2 records
        let stream = stream_quotes(&temp_file, 2, Some(4), Some(1), None, Some(2)).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 1 chunk with 2 items (limited)
        assert_eq!(chunks.len(), 1);
        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);

        // Verify we get exactly 2 records
        let total_quotes: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_quotes, 2);

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

    #[rstest]
    pub fn test_stream_trades_with_limit() {
        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,2.0
binance,BTCUSDT,1640995202000000,1640995202100000,trade3,buy,50000.12,1.5
binance,BTCUSDT,1640995203000000,1640995203100000,trade4,sell,49999.123,3.0
binance,BTCUSDT,1640995204000000,1640995204100000,trade5,buy,50000.1234,0.5";

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

        // Test with limit of 3 records
        let stream = stream_trades(&temp_file, 2, Some(4), Some(1), None, Some(3)).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 2 chunks: [2 items, 1 item] = 3 total (limited)
        assert_eq!(chunks.len(), 2);
        let chunk1 = chunks[0].as_ref().unwrap();
        assert_eq!(chunk1.len(), 2);
        let chunk2 = chunks[1].as_ref().unwrap();
        assert_eq!(chunk2.len(), 1);

        // Verify we get exactly 3 records
        let total_trades: usize = chunks.iter().map(|c| c.as_ref().unwrap().len()).sum();
        assert_eq!(total_trades, 3);

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

    #[rstest]
    pub fn test_depth10_invalid_levels_error_at_construction() {
        let temp_file = std::env::temp_dir().join("test_depth10_invalid_levels.csv");
        std::fs::write(&temp_file, "exchange,symbol,timestamp,local_timestamp\n").unwrap();

        let result = Depth10StreamIterator::new(&temp_file, 10, 10, None, None, None, None);
        assert!(result.is_err());
        let err_msg = result.err().unwrap().to_string();
        assert!(
            err_msg.contains("Invalid levels"),
            "Error should mention 'Invalid levels': {err_msg}"
        );

        let result = Depth10StreamIterator::new(&temp_file, 10, 3, None, None, None, None);
        assert!(result.is_err());

        let result = Depth10StreamIterator::new(&temp_file, 10, 5, None, None, None, None);
        assert!(result.is_ok());

        let result = Depth10StreamIterator::new(&temp_file, 10, 25, None, None, None, None);
        assert!(result.is_ok());

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

    #[rstest]
    pub fn test_stream_deltas_with_mid_snapshot_inserts_clear() {
        // CSV with:
        // - Initial snapshot (is_snapshot=true) at start
        // - Some deltas (is_snapshot=false)
        // - Mid-day snapshot (is_snapshot=true) - should trigger CLEAR
        // - Back to deltas (is_snapshot=false)
        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,1640995300000000,1640995300100000,true,bid,50100.0,3.0
binance-futures,BTCUSDT,1640995300000000,1640995300100000,true,ask,50101.0,4.0
binance-futures,BTCUSDT,1640995301000000,1640995301100000,false,bid,50099.0,1.0";

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

        let stream = stream_deltas(&temp_file, 100, Some(1), Some(1), None, None).unwrap();
        let all_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();

        let clear_count = all_deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        // Should have 2 CLEAR deltas: initial snapshot + mid-day snapshot
        assert_eq!(
            clear_count, 2,
            "Expected 2 CLEAR deltas (initial + mid-day snapshot), found {clear_count}"
        );

        // Verify CLEAR positions:
        // 0=CLEAR, 1=Add, 2=Add, 3=Update, 4=Update, 5=CLEAR, 6=Add, 7=Add, 8=Update
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[5].action, BookAction::Clear);

        // CLEAR deltas should NOT have F_LAST when followed by same-timestamp deltas
        assert_eq!(
            all_deltas[0].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 0 should not have F_LAST flag"
        );
        assert_eq!(
            all_deltas[5].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 5 should not have F_LAST flag"
        );

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

    #[rstest]
    pub fn test_stream_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_stream_deltas_consecutive_snapshots.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        let stream = stream_deltas(&temp_file, 100, Some(1), Some(1), None, None).unwrap();
        let all_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();
        let clear_count = all_deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        assert_eq!(clear_count, 2);
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[3].action, BookAction::Clear);
        assert_eq!(
            all_deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
        assert_eq!(all_deltas[3].flags & RecordFlag::F_LAST as u8, 0);
        assert_eq!(
            all_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]
    pub fn test_stream_deltas_consecutive_snapshots_clear_across_chunk_boundary() {
        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,1640995200000000,1640995200100000,true,ask,50001.0,2.0
hyperliquid,BTC,1640995201000000,1640995201100000,true,bid,49990.0,3.0
hyperliquid,BTC,1640995201000000,1640995201100000,true,ask,49991.0,4.0";

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

        // Chunk size 2 puts the second CLEAR at a chunk boundary and defers its snapshot row,
        // which must not insert another CLEAR.
        let stream = stream_deltas(&temp_file, 2, Some(1), Some(1), None, None).unwrap();
        let all_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();
        let clear_count = all_deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        assert_eq!(all_deltas.len(), 6);
        assert_eq!(clear_count, 2);
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[3].action, BookAction::Clear);
        assert_eq!(
            all_deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8
        );
        assert_eq!(all_deltas[3].flags & RecordFlag::F_LAST as u8, 0);

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

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

        let clear_count = deltas
            .iter()
            .filter(|d| d.action == BookAction::Clear)
            .count();

        // Should have 2 CLEAR deltas: initial snapshot + mid-day snapshot
        assert_eq!(
            clear_count, 2,
            "Expected 2 CLEAR deltas (initial + mid-day snapshot), found {clear_count}"
        );

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

        let second_clear_idx = deltas
            .iter()
            .enumerate()
            .filter(|(_, d)| d.action == BookAction::Clear)
            .nth(1)
            .map(|(i, _)| i)
            .expect("Should have second CLEAR");

        // 0=CLEAR, 1=Add, 2=Add, 3=Update, 4=Update, 5=Delete, 6=CLEAR
        assert_eq!(
            second_clear_idx, 6,
            "Second CLEAR should be at index 6, found {second_clear_idx}"
        );

        // CLEAR deltas should NOT have F_LAST when followed by same-timestamp deltas
        assert_eq!(
            deltas[0].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 0 should not have F_LAST flag"
        );
        assert_eq!(
            deltas[6].flags & RecordFlag::F_LAST as u8,
            0,
            "CLEAR at index 6 should not have F_LAST flag"
        );
    }

    #[rstest]
    fn test_stream_deltas_chunk_size_respects_clear() {
        // Test that chunk_size applies to total emitted deltas (including CLEARs)
        // With chunk_size=1, a snapshot boundary should emit CLEAR in one chunk
        // and the real delta in the next chunk
        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_stream_chunk_size_clear.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        // chunk_size=1 should produce separate chunks for CLEAR and real deltas
        let stream = stream_deltas(&temp_file, 1, Some(1), Some(1), None, None).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 3 chunks: [CLEAR], [data], [data]
        assert_eq!(chunks.len(), 3, "Expected 3 chunks with chunk_size=1");
        assert_eq!(chunks[0].as_ref().unwrap().len(), 1);
        assert_eq!(chunks[1].as_ref().unwrap().len(), 1);
        assert_eq!(chunks[2].as_ref().unwrap().len(), 1);

        // First chunk should be CLEAR
        assert_eq!(chunks[0].as_ref().unwrap()[0].action, BookAction::Clear);
        // Second and third chunks should be data deltas
        assert_eq!(chunks[1].as_ref().unwrap()[0].action, BookAction::Add);
        assert_eq!(chunks[2].as_ref().unwrap()[0].action, BookAction::Add);

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

    #[rstest]
    fn test_stream_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_stream_limit_stops_at_clear.csv");
        std::fs::write(&temp_file, csv_data).unwrap();

        // limit=1 should only get the CLEAR delta
        let stream = stream_deltas(&temp_file, 100, Some(1), Some(1), None, Some(1)).unwrap();
        let all_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();

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

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

    #[rstest]
    fn test_stream_deltas_limit_includes_clear() {
        // Test that limit counts total emitted deltas (including CLEARs)
        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";

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

        // limit=4 should get exactly 4 deltas: 1 CLEAR + 3 data deltas
        let stream = stream_deltas(&temp_file, 100, Some(1), Some(1), None, Some(4)).unwrap();
        let all_deltas: Vec<_> = stream.flat_map(|chunk| chunk.unwrap()).collect();

        assert_eq!(all_deltas.len(), 4);
        assert_eq!(all_deltas[0].action, BookAction::Clear);
        assert_eq!(all_deltas[1].action, BookAction::Add);
        assert_eq!(all_deltas[2].action, BookAction::Add);
        assert_eq!(all_deltas[3].action, BookAction::Update);

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

    #[rstest]
    fn test_stream_deltas_limit_sets_f_last() {
        // Test that F_LAST is set on the final delta when limit is reached
        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";

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

        // limit=3 should get 3 deltas with F_LAST on the last one
        let stream = stream_deltas(&temp_file, 100, Some(1), Some(1), None, Some(3)).unwrap();
        let chunks: Vec<_> = stream.collect();

        // Should have 1 chunk with 3 deltas
        assert_eq!(chunks.len(), 1);
        let deltas = chunks[0].as_ref().unwrap();
        assert_eq!(deltas.len(), 3);

        // Final delta should have F_LAST flag
        assert_eq!(
            deltas[2].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8,
            "Final delta should have F_LAST flag when limit is reached"
        );

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

    #[rstest]
    fn test_stream_deltas_chunk_boundary_no_f_last() {
        // Test that F_LAST is NOT set when only chunk_size boundary is hit (more data follows)
        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
binance-futures,BTCUSDT,1640995200000000,1640995200100000,false,bid,50000.0,1.0
binance-futures,BTCUSDT,1640995200000000,1640995200100000,false,ask,50001.0,2.0
binance-futures,BTCUSDT,1640995200000000,1640995200100000,false,bid,49999.0,0.5";

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

        // chunk_size=2, no limit - first chunk should NOT have F_LAST (more data follows)
        let mut stream = stream_deltas(&temp_file, 2, Some(1), Some(1), None, None).unwrap();

        let chunk1 = stream.next().unwrap().unwrap();
        assert_eq!(chunk1.len(), 2);

        // First chunk's last delta should NOT have F_LAST (more data follows with same timestamp)
        assert_eq!(
            chunk1[1].flags & RecordFlag::F_LAST as u8,
            0,
            "Mid-stream chunk should not have F_LAST flag"
        );

        // Second chunk exists and has F_LAST (end of file)
        let chunk2 = stream.next().unwrap().unwrap();
        assert_eq!(chunk2.len(), 1);
        assert_eq!(
            chunk2[0].flags & RecordFlag::F_LAST as u8,
            RecordFlag::F_LAST as u8,
            "Final chunk at EOF should have F_LAST flag"
        );

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