nautilus_tardis/

replay.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::{
    fs,
    path::{Path, PathBuf},
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use arrow::record_batch::RecordBatch;
use futures_util::{StreamExt, pin_mut};
use jiff::{Timestamp, civil::Date, tz::Offset};
use nautilus_core::{UnixNanos, datetime::unix_nanos_to_iso8601, string::formatting::Separable};
use nautilus_model::{
    data::{
        Bar, BarType, Data, OptionGreeks, OrderBookDelta, OrderBookDeltas, OrderBookDepth,
        QuoteTick, TradeTick,
    },
    identifiers::InstrumentId,
};
use nautilus_persistence::common::paths::CatalogPathPrefix;
use nautilus_serialization::arrow::{
    bars_to_arrow_record_batch_bytes, book_deltas_to_arrow_record_batch_bytes,
    book_depths_to_arrow_record_batch_bytes, option_greeks_to_arrow_record_batch_bytes,
    quotes_to_arrow_record_batch_bytes, trades_to_arrow_record_batch_bytes,
};
use parquet::{arrow::ArrowWriter, basic::Compression, file::properties::WriterProperties};

use crate::{
    config::{BookSnapshotOutput, ParquetCompression, TardisReplayConfig},
    http::TardisHttpClient,
    machine::TardisMachineClient,
};

struct DateCursor {
    /// Cursor date UTC.
    date_utc: Date,
    /// Cursor end timestamp UNIX nanoseconds.
    end_ns: UnixNanos,
}

impl DateCursor {
    /// Creates a new [`DateCursor`] instance.
    fn new(current_ns: UnixNanos) -> Self {
        let current_utc = current_ns.to_datetime_utc();
        let date_utc = Offset::UTC.to_datetime(current_utc).date();

        // Calculate end of the current UTC day
        let end_utc = utc_timestamp(date_utc, 23, 59, 59, 999_999_999);
        let end_ns = UnixNanos::from(u64::try_from(end_utc.as_nanosecond()).unwrap_or(u64::MAX));

        Self { date_utc, end_ns }
    }
}

fn utc_timestamp(date: Date, hour: i8, minute: i8, second: i8, nanosecond: i32) -> Timestamp {
    Offset::UTC
        .to_timestamp(date.at(hour, minute, second, nanosecond))
        .expect("valid UTC civil datetime")
}

/// Runs the Tardis Machine replay from a JSON configuration file.
///
/// # Errors
///
/// Returns an error if reading or parsing the config file fails,
/// or if any downstream replay operation fails.
///
/// # Panics
///
/// Panics if unable to determine the output path (current directory fallback fails).
pub async fn run_tardis_machine_replay_from_config(config_filepath: &Path) -> anyhow::Result<()> {
    log::debug!("Starting replay");
    log::debug!("Config filepath: {}", config_filepath.display());

    // Load and parse the replay configuration
    let config_data = fs::read_to_string(config_filepath)
        .with_context(|| format!("Failed to read config file: {}", config_filepath.display()))?;
    let config: TardisReplayConfig = serde_json::from_str(&config_data)
        .context("failed to parse config JSON into TardisReplayConfig")?;

    let path = config
        .output_path
        .as_deref()
        .map(Path::new)
        .map(Path::to_path_buf)
        .or_else(|| {
            std::env::var("NAUTILUS_PATH")
                .ok()
                .map(|env_path| PathBuf::from(env_path).join("catalog").join("data"))
        })
        .unwrap_or_else(|| std::env::current_dir().expect("Failed to get current directory"));

    log::debug!("Output path: {}", path.display());

    let normalize_symbols = config.normalize_symbols.unwrap_or(true);
    log::debug!("normalize_symbols={normalize_symbols}");

    let book_snapshot_output = config
        .book_snapshot_output
        .clone()
        .unwrap_or(BookSnapshotOutput::Deltas);
    log::debug!("book_snapshot_output={book_snapshot_output:?}");

    let extract_bbo_as_quotes = config.extract_bbo_as_quotes.unwrap_or(false);
    log::debug!("extract_bbo_as_quotes={extract_bbo_as_quotes}");

    let compression = config
        .compression
        .clone()
        .unwrap_or(ParquetCompression::Zstd);
    log::debug!("compression={compression:?}");
    let compression = compression.as_parquet_compression();

    let http_client = TardisHttpClient::new(
        None,
        config
            .tardis_http_url
            .as_ref()
            .map(|value| value.expose_secret()),
        None,
        normalize_symbols,
        config
            .proxy_url
            .as_ref()
            .map(|value| value.expose_secret().to_owned()),
    )?;
    let mut machine_client = TardisMachineClient::new(
        config
            .tardis_ws_url
            .as_ref()
            .map(|value| value.expose_secret()),
        normalize_symbols,
        book_snapshot_output,
    )?;
    machine_client.extract_bbo_as_quotes = extract_bbo_as_quotes;

    let exchanges: AHashSet<_> = config.options.iter().map(|opt| opt.exchange).collect();
    let (instrument_map, _instruments) = http_client
        .bootstrap_instruments(&exchanges)
        .await
        .context("failed to bootstrap instruments")?;

    for (_, info) in &instrument_map {
        machine_client.add_instrument_info((**info).clone());
    }

    log::debug!("Starting tardis-machine stream");
    let stream = machine_client.replay(config.options).await?;
    pin_mut!(stream);

    // Initialize date cursors
    let mut deltas_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();
    let mut depths_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();
    let mut quotes_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();
    let mut trades_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();
    let mut bars_cursors: AHashMap<BarType, DateCursor> = AHashMap::new();
    let mut greeks_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();

    // Initialize date collection maps
    let mut deltas_map: AHashMap<InstrumentId, Vec<OrderBookDelta>> = AHashMap::new();
    let mut depths_map: AHashMap<InstrumentId, Vec<OrderBookDepth>> = AHashMap::new();
    let mut quotes_map: AHashMap<InstrumentId, Vec<QuoteTick>> = AHashMap::new();
    let mut trades_map: AHashMap<InstrumentId, Vec<TradeTick>> = AHashMap::new();
    let mut bars_map: AHashMap<BarType, Vec<Bar>> = AHashMap::new();
    let mut greeks_map: AHashMap<InstrumentId, Vec<OptionGreeks>> = AHashMap::new();

    let mut msg_count = 0;

    while let Some(result) = stream.next().await {
        match result {
            Ok(msg) => {
                match msg {
                    Data::BookDelta(delta) => {
                        log::warn!(
                            "Skipping individual delta message for {} (use Deltas batch instead)",
                            delta.instrument_id
                        );
                    }
                    Data::BookDeltas(msg) => {
                        handle_deltas_msg(
                            &msg,
                            &mut deltas_map,
                            &mut deltas_cursors,
                            &path,
                            compression,
                        );
                    }
                    Data::BookDepth(msg) => {
                        handle_depth_msg(
                            *msg,
                            &mut depths_map,
                            &mut depths_cursors,
                            &path,
                            compression,
                        );
                    }
                    Data::Quote(msg) => {
                        handle_quote_msg(
                            msg,
                            &mut quotes_map,
                            &mut quotes_cursors,
                            &path,
                            compression,
                        );
                    }
                    Data::Trade(msg) => {
                        handle_trade_msg(
                            msg,
                            &mut trades_map,
                            &mut trades_cursors,
                            &path,
                            compression,
                        );
                    }
                    Data::Bar(msg) => {
                        handle_bar_msg(msg, &mut bars_map, &mut bars_cursors, &path, compression);
                    }
                    Data::MarkPrice(_) | Data::IndexPrice(_) | Data::FundingRate(_) => {
                        log::debug!(
                            "Skipping unsupported data type for instrument {}",
                            msg.instrument_id()
                        );
                    }
                    Data::OptionGreeks(msg) => {
                        handle_option_greeks_msg(
                            msg,
                            &mut greeks_map,
                            &mut greeks_cursors,
                            &path,
                            compression,
                        );
                    }
                    Data::InstrumentStatus(_) | Data::InstrumentClose(_) | Data::Custom(_) => {
                        log::debug!(
                            "Skipping unsupported data type for instrument {}",
                            msg.instrument_id()
                        );
                    }
                    #[allow(unreachable_patterns)]
                    _ => {
                        log::debug!("Skipping unsupported data type");
                    }
                }

                msg_count += 1;
                if msg_count % 100_000 == 0 {
                    log::debug!("Processed {} messages", msg_count.separate_with_commas());
                }
            }
            Err(e) => {
                log::error!("Stream error: {e:?}");
                break;
            }
        }
    }

    // Iterate through every remaining type and instrument sequentially

    for (instrument_id, deltas) in &deltas_map {
        let cursor = deltas_cursors.get(instrument_id).expect("Expected cursor");
        batch_and_write_deltas(deltas, instrument_id, cursor.date_utc, &path, compression);
    }

    for (instrument_id, depths) in &depths_map {
        let cursor = depths_cursors.get(instrument_id).expect("Expected cursor");
        batch_and_write_depths(depths, instrument_id, cursor.date_utc, &path, compression);
    }

    for (instrument_id, quotes) in &quotes_map {
        let cursor = quotes_cursors.get(instrument_id).expect("Expected cursor");
        batch_and_write_quotes(quotes, instrument_id, cursor.date_utc, &path, compression);
    }

    for (instrument_id, trades) in &trades_map {
        let cursor = trades_cursors.get(instrument_id).expect("Expected cursor");
        batch_and_write_trades(trades, instrument_id, cursor.date_utc, &path, compression);
    }

    for (bar_type, bars) in &bars_map {
        let cursor = bars_cursors.get(bar_type).expect("Expected cursor");
        batch_and_write_bars(bars, bar_type, cursor.date_utc, &path, compression);
    }

    for (instrument_id, greeks) in &greeks_map {
        let cursor = greeks_cursors.get(instrument_id).expect("Expected cursor");
        batch_and_write_greeks(greeks, instrument_id, cursor.date_utc, &path, compression);
    }

    log::debug!(
        "Replay completed after {} messages",
        msg_count.separate_with_commas()
    );
    Ok(())
}

fn handle_deltas_msg(
    deltas: &OrderBookDeltas,
    map: &mut AHashMap<InstrumentId, Vec<OrderBookDelta>>,
    cursors: &mut AHashMap<InstrumentId, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(deltas.instrument_id)
        .or_insert_with(|| DateCursor::new(deltas.ts_init));

    if deltas.ts_init > cursor.end_ns {
        if let Some(deltas_vec) = map.remove(&deltas.instrument_id) {
            batch_and_write_deltas(
                &deltas_vec,
                &deltas.instrument_id,
                cursor.date_utc,
                path,
                compression,
            );
        }
        // Update cursor
        *cursor = DateCursor::new(deltas.ts_init);
    }

    map.entry(deltas.instrument_id)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .extend(&*deltas.deltas);
}

fn handle_depth_msg(
    depth: OrderBookDepth,
    map: &mut AHashMap<InstrumentId, Vec<OrderBookDepth>>,
    cursors: &mut AHashMap<InstrumentId, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(depth.instrument_id)
        .or_insert_with(|| DateCursor::new(depth.ts_init));

    if depth.ts_init > cursor.end_ns {
        if let Some(depths_vec) = map.remove(&depth.instrument_id) {
            batch_and_write_depths(
                &depths_vec,
                &depth.instrument_id,
                cursor.date_utc,
                path,
                compression,
            );
        }
        // Update cursor
        *cursor = DateCursor::new(depth.ts_init);
    }

    map.entry(depth.instrument_id)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .push(depth);
}

fn handle_quote_msg(
    quote: QuoteTick,
    map: &mut AHashMap<InstrumentId, Vec<QuoteTick>>,
    cursors: &mut AHashMap<InstrumentId, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(quote.instrument_id)
        .or_insert_with(|| DateCursor::new(quote.ts_init));

    if quote.ts_init > cursor.end_ns {
        if let Some(quotes_vec) = map.remove(&quote.instrument_id) {
            batch_and_write_quotes(
                &quotes_vec,
                &quote.instrument_id,
                cursor.date_utc,
                path,
                compression,
            );
        }
        // Update cursor
        *cursor = DateCursor::new(quote.ts_init);
    }

    map.entry(quote.instrument_id)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .push(quote);
}

fn handle_trade_msg(
    trade: TradeTick,
    map: &mut AHashMap<InstrumentId, Vec<TradeTick>>,
    cursors: &mut AHashMap<InstrumentId, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(trade.instrument_id)
        .or_insert_with(|| DateCursor::new(trade.ts_init));

    if trade.ts_init > cursor.end_ns {
        if let Some(trades_vec) = map.remove(&trade.instrument_id) {
            batch_and_write_trades(
                &trades_vec,
                &trade.instrument_id,
                cursor.date_utc,
                path,
                compression,
            );
        }
        // Update cursor
        *cursor = DateCursor::new(trade.ts_init);
    }

    map.entry(trade.instrument_id)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .push(trade);
}

fn handle_bar_msg(
    bar: Bar,
    map: &mut AHashMap<BarType, Vec<Bar>>,
    cursors: &mut AHashMap<BarType, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(bar.bar_type)
        .or_insert_with(|| DateCursor::new(bar.ts_init));

    if bar.ts_init > cursor.end_ns {
        if let Some(bars_vec) = map.remove(&bar.bar_type) {
            batch_and_write_bars(&bars_vec, &bar.bar_type, cursor.date_utc, path, compression);
        }
        // Update cursor
        *cursor = DateCursor::new(bar.ts_init);
    }

    map.entry(bar.bar_type)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .push(bar);
}

fn handle_option_greeks_msg(
    greeks: OptionGreeks,
    map: &mut AHashMap<InstrumentId, Vec<OptionGreeks>>,
    cursors: &mut AHashMap<InstrumentId, DateCursor>,
    path: &Path,
    compression: Compression,
) {
    let cursor = cursors
        .entry(greeks.instrument_id)
        .or_insert_with(|| DateCursor::new(greeks.ts_init));

    if greeks.ts_init > cursor.end_ns {
        if let Some(greeks_vec) = map.remove(&greeks.instrument_id) {
            batch_and_write_greeks(
                &greeks_vec,
                &greeks.instrument_id,
                cursor.date_utc,
                path,
                compression,
            );
        }
        // Update cursor
        *cursor = DateCursor::new(greeks.ts_init);
    }

    map.entry(greeks.instrument_id)
        .or_insert_with(|| Vec::with_capacity(100_000))
        .push(greeks);
}

fn batch_and_write_deltas(
    deltas: &[OrderBookDelta],
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    match book_deltas_to_arrow_record_batch_bytes(deltas) {
        Ok(batch) => write_batch(
            &batch,
            OrderBookDelta::path_prefix(),
            instrument_id,
            date,
            path,
            compression,
        ),
        Err(e) => {
            log::error!("Error converting OrderBookDeltas to Arrow: {e:?}");
        }
    }
}

fn batch_and_write_depths(
    depths: &[OrderBookDepth],
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    match book_depths_to_arrow_record_batch_bytes(depths) {
        Ok(batch) => write_batch(
            &batch,
            OrderBookDepth::path_prefix(),
            instrument_id,
            date,
            path,
            compression,
        ),
        Err(e) => {
            log::error!("Error converting OrderBookDepth to Arrow: {e:?}");
        }
    }
}

fn batch_and_write_quotes(
    quotes: &[QuoteTick],
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    match quotes_to_arrow_record_batch_bytes(quotes) {
        Ok(batch) => write_batch(
            &batch,
            QuoteTick::path_prefix(),
            instrument_id,
            date,
            path,
            compression,
        ),
        Err(e) => {
            log::error!("Error converting QuoteTick to Arrow: {e:?}");
        }
    }
}

fn batch_and_write_trades(
    trades: &[TradeTick],
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    match trades_to_arrow_record_batch_bytes(trades) {
        Ok(batch) => write_batch(
            &batch,
            TradeTick::path_prefix(),
            instrument_id,
            date,
            path,
            compression,
        ),
        Err(e) => {
            log::error!("Error converting TradeTick to Arrow: {e:?}");
        }
    }
}

fn batch_and_write_bars(
    bars: &[Bar],
    bar_type: &BarType,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    let batch = match bars_to_arrow_record_batch_bytes(bars) {
        Ok(batch) => batch,
        Err(e) => {
            log::error!("Error converting Bar to Arrow: {e:?}");
            return;
        }
    };

    let filepath = path.join(parquet_filepath_bars(bar_type, date));
    if let Err(e) = write_parquet_local(&batch, &filepath, compression) {
        log::error!("Error writing {}: {e}", filepath.display());
    } else {
        log::debug!("File written: {}", filepath.display());
    }
}

fn batch_and_write_greeks(
    greeks: &[OptionGreeks],
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    match option_greeks_to_arrow_record_batch_bytes(greeks) {
        Ok(batch) => write_batch(
            &batch,
            OptionGreeks::path_prefix(),
            instrument_id,
            date,
            path,
            compression,
        ),
        Err(e) => {
            log::error!("Error converting OptionGreeks to Arrow: {e:?}");
        }
    }
}

/// Asserts that the given date is on or after the UNIX epoch (1970-01-01).
///
/// # Panics
///
/// Panics if the date is before 1970-01-01, as pre-epoch dates cannot be
/// reliably represented as UnixNanos without overflow issues.
fn assert_post_epoch(date: Date) {
    let epoch = Date::constant(1970, 1, 1);
    assert!(
        date >= epoch,
        "Tardis replay filenames require dates on or after 1970-01-01; received {date}"
    );
}

/// Converts an ISO 8601 timestamp to a filesystem-safe format.
///
/// This function replaces colons and dots with hyphens to make the timestamp
/// safe for use in filenames across different filesystems.
fn iso_timestamp_to_file_timestamp(iso_timestamp: &str) -> String {
    iso_timestamp.replace([':', '.'], "-")
}

/// Converts timestamps to a filename using ISO 8601 format.
///
/// This function converts two Unix nanosecond timestamps to a filename that uses
/// ISO 8601 format with filesystem-safe characters, matching the catalog convention.
fn timestamps_to_filename(timestamp_1: UnixNanos, timestamp_2: UnixNanos) -> String {
    let datetime_1 = iso_timestamp_to_file_timestamp(&unix_nanos_to_iso8601(timestamp_1));
    let datetime_2 = iso_timestamp_to_file_timestamp(&unix_nanos_to_iso8601(timestamp_2));

    format!("{datetime_1}_{datetime_2}.parquet")
}

fn parquet_filepath(typename: &str, instrument_id: &InstrumentId, date: Date) -> PathBuf {
    assert_post_epoch(date);

    let instrument_id_str = instrument_id.to_string().replace('/', "");

    let start_nanos = utc_timestamp(date, 0, 0, 0, 0).as_nanosecond();
    let end_nanos = utc_timestamp(date, 23, 59, 59, 999_999_999).as_nanosecond();

    let filename = timestamps_to_filename(
        UnixNanos::from(u64::try_from(start_nanos).expect("date fits UnixNanos")),
        UnixNanos::from(u64::try_from(end_nanos).expect("date fits UnixNanos")),
    );

    PathBuf::new()
        .join(typename)
        .join(instrument_id_str)
        .join(filename)
}

fn parquet_filepath_bars(bar_type: &BarType, date: Date) -> PathBuf {
    assert_post_epoch(date);

    let bar_type_str = bar_type.to_string().replace('/', "");

    // Calculate start and end timestamps for the day (UTC)
    let start_nanos = utc_timestamp(date, 0, 0, 0, 0).as_nanosecond();
    let end_nanos = utc_timestamp(date, 23, 59, 59, 999_999_999).as_nanosecond();

    let filename = timestamps_to_filename(
        UnixNanos::from(u64::try_from(start_nanos).expect("date fits UnixNanos")),
        UnixNanos::from(u64::try_from(end_nanos).expect("date fits UnixNanos")),
    );

    PathBuf::new()
        .join(Bar::path_prefix())
        .join(bar_type_str)
        .join(filename)
}

fn write_batch(
    batch: &RecordBatch,
    typename: &str,
    instrument_id: &InstrumentId,
    date: Date,
    path: &Path,
    compression: Compression,
) {
    let filepath = path.join(parquet_filepath(typename, instrument_id, date));
    if let Err(e) = write_parquet_local(batch, &filepath, compression) {
        log::error!("Error writing {}: {e}", filepath.display());
    } else {
        log::debug!("File written: {}", filepath.display());
    }
}

fn write_parquet_local(
    batch: &RecordBatch,
    file_path: &Path,
    compression: Compression,
) -> anyhow::Result<()> {
    if let Some(parent) = file_path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    let file = std::fs::File::create(file_path)?;
    let props = WriterProperties::builder()
        .set_compression(compression)
        .build();

    let mut writer = ArrowWriter::try_new(file, batch.schema(), Some(props))?;
    writer.write(batch)?;
    writer.close()?;
    Ok(())
}

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

    use nautilus_core::DurationNanos;
    use nautilus_persistence::backend::catalog::ParquetDataCatalog;
    use rstest::rstest;

    use super::*;
    use crate::{
        common::{enums::TardisExchange, testing::load_test_json},
        config::BookSnapshotOutput,
        machine::{
            message::{BookSnapshotMsg, OptionSummaryMsg, TradeMsg, WsMessage},
            parse::parse_tardis_ws_message,
            types::TardisInstrumentMiniInfo,
        },
    };

    fn utc_nanos(
        year: i16,
        month: i8,
        day: i8,
        hour: i8,
        minute: i8,
        second: i8,
        nanosecond: i32,
    ) -> u64 {
        u64::try_from(
            utc_timestamp(
                Date::new(year, month, day).unwrap(),
                hour,
                minute,
                second,
                nanosecond,
            )
            .as_nanosecond(),
        )
        .unwrap()
    }

    #[rstest]
    #[case(
    // Start of day: 2024-01-01 00:00:00 UTC
    utc_nanos(2024, 1, 1, 0, 0, 0, 0),
    Date::new(2024, 1, 1).unwrap(),
    utc_nanos(2024, 1, 1, 23, 59, 59, 999_999_999)
)]
    #[case(
    // Midday: 2024-01-01 12:00:00 UTC
    utc_nanos(2024, 1, 1, 12, 0, 0, 0),
    Date::new(2024, 1, 1).unwrap(),
    utc_nanos(2024, 1, 1, 23, 59, 59, 999_999_999)
)]
    #[case(
    // End of day: 2024-01-01 23:59:59.999999999 UTC
    utc_nanos(2024, 1, 1, 23, 59, 59, 999_999_999),
    Date::new(2024, 1, 1).unwrap(),
    utc_nanos(2024, 1, 1, 23, 59, 59, 999_999_999)
)]
    #[case(
    // Start of new day: 2024-01-02 00:00:00 UTC
    utc_nanos(2024, 1, 2, 0, 0, 0, 0),
    Date::new(2024, 1, 2).unwrap(),
    utc_nanos(2024, 1, 2, 23, 59, 59, 999_999_999)
)]
    fn test_date_cursor(
        #[case] timestamp: u64,
        #[case] expected_date: Date,
        #[case] expected_end_ns: u64,
    ) {
        let unix_nanos = UnixNanos::from(timestamp);
        let cursor = DateCursor::new(unix_nanos);

        assert_eq!(cursor.date_utc, expected_date);
        assert_eq!(cursor.end_ns, UnixNanos::from(expected_end_ns));
    }

    #[rstest]
    fn test_option_greeks_replay_catalog_round_trip() {
        let instrument_id = InstrumentId::from("BTC-28JUN24-70000-C.DERIBIT");
        let compression = ParquetCompression::Zstd.as_parquet_compression();
        let info = Arc::new(TardisInstrumentMiniInfo::new(
            instrument_id,
            None,
            TardisExchange::Deribit,
            4,
            1,
        ));

        let option_summary: OptionSummaryMsg =
            serde_json::from_str(&load_test_json("option_summary.json")).unwrap();
        let Some(Data::OptionGreeks(greeks_1)) = parse_tardis_ws_message(
            WsMessage::OptionSummary(option_summary),
            &info,
            &BookSnapshotOutput::Deltas,
        ) else {
            panic!("Expected option_summary to route to Data::OptionGreeks");
        };

        let mut greeks_2 = greeks_1;
        greeks_2.ts_event = greeks_1.ts_event + DurationNanos::from_secs(1);
        greeks_2.ts_init = greeks_1.ts_init + DurationNanos::from_secs(1);
        greeks_2.greeks.delta = 0.26;

        let option_quote: BookSnapshotMsg =
            serde_json::from_str(&load_test_json("option_book_snapshot.json")).unwrap();
        let Some(Data::Quote(quote_1)) = parse_tardis_ws_message(
            WsMessage::BookSnapshot(option_quote),
            &info,
            &BookSnapshotOutput::Deltas,
        ) else {
            panic!("Expected depth-1 option book snapshot to route to Data::Quote");
        };

        let mut quote_2 = quote_1;
        quote_2.ts_event = quote_1.ts_event + DurationNanos::from_secs(1);
        quote_2.ts_init = quote_1.ts_init + DurationNanos::from_secs(1);

        let temp_dir = tempfile::tempdir().unwrap();
        let data_path = temp_dir.path().join("data");

        let mut quotes_map: AHashMap<InstrumentId, Vec<QuoteTick>> = AHashMap::new();
        let mut quotes_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();
        let mut greeks_map: AHashMap<InstrumentId, Vec<OptionGreeks>> = AHashMap::new();
        let mut greeks_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();

        for quote in [quote_1, quote_2] {
            handle_quote_msg(
                quote,
                &mut quotes_map,
                &mut quotes_cursors,
                &data_path,
                compression,
            );
        }

        for greeks in [greeks_1, greeks_2] {
            handle_option_greeks_msg(
                greeks,
                &mut greeks_map,
                &mut greeks_cursors,
                &data_path,
                compression,
            );
        }

        for (id, quotes) in &quotes_map {
            let cursor = quotes_cursors.get(id).expect("Expected cursor");
            batch_and_write_quotes(quotes, id, cursor.date_utc, &data_path, compression);
        }

        for (id, greeks) in &greeks_map {
            let cursor = greeks_cursors.get(id).expect("Expected cursor");
            batch_and_write_greeks(greeks, id, cursor.date_utc, &data_path, compression);
        }

        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
        let identifiers = Some(vec![instrument_id.to_string()]);

        let quotes_out = catalog
            .quote_ticks(identifiers.clone(), None, None)
            .unwrap();
        let greeks_out = catalog.option_greeks(identifiers, None, None).unwrap();

        assert_eq!(greeks_out, vec![greeks_1, greeks_2]);
        assert_eq!(greeks_out[0].instrument_id, instrument_id);
        assert_eq!(greeks_out[0].mark_iv, Some(0.565));
        assert_eq!(greeks_out[0].underlying_price, Some(63_500.0));
        assert!(greeks_out[0].ts_init < greeks_out[1].ts_init);

        assert_eq!(quotes_out, vec![quote_1, quote_2]);
        assert_eq!(quotes_out[0].instrument_id, instrument_id);
        assert!(quotes_out[0].ts_init < quotes_out[1].ts_init);
    }

    #[rstest]
    fn test_trades_replay_catalog_round_trip() {
        let instrument_id = InstrumentId::from("XBTUSD.BITMEX");
        let compression = ParquetCompression::Zstd.as_parquet_compression();
        let info = Arc::new(TardisInstrumentMiniInfo::new(
            instrument_id,
            None,
            TardisExchange::Bitmex,
            1,
            0,
        ));

        let trade_msg: TradeMsg = serde_json::from_str(&load_test_json("trade.json")).unwrap();
        let Some(Data::Trade(trade_1)) = parse_tardis_ws_message(
            WsMessage::Trade(trade_msg),
            &info,
            &BookSnapshotOutput::Deltas,
        ) else {
            panic!("Expected trade message to route to Data::Trade");
        };

        let mut trade_2 = trade_1;
        trade_2.ts_event = trade_1.ts_event + DurationNanos::from_secs(1);
        trade_2.ts_init = trade_1.ts_init + DurationNanos::from_secs(1);

        let temp_dir = tempfile::tempdir().unwrap();
        let data_path = temp_dir.path().join("data");

        let mut trades_map: AHashMap<InstrumentId, Vec<TradeTick>> = AHashMap::new();
        let mut trades_cursors: AHashMap<InstrumentId, DateCursor> = AHashMap::new();

        for trade in [trade_1, trade_2] {
            handle_trade_msg(
                trade,
                &mut trades_map,
                &mut trades_cursors,
                &data_path,
                compression,
            );
        }

        for (id, trades) in &trades_map {
            let cursor = trades_cursors.get(id).expect("Expected cursor");
            batch_and_write_trades(trades, id, cursor.date_utc, &data_path, compression);
        }

        // Trades must be written under the catalog convention path ("trades")
        assert!(data_path.join(TradeTick::path_prefix()).exists());
        assert!(!data_path.join("trade_tick").exists());

        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
        let trades_out = catalog
            .trade_ticks(Some(vec![instrument_id.to_string()]), None, None)
            .unwrap();

        assert_eq!(trades_out, vec![trade_1, trade_2]);
        assert_eq!(trades_out[0].instrument_id, instrument_id);
        assert!(trades_out[0].ts_init < trades_out[1].ts_init);
    }

    #[rstest]
    fn test_bars_replay_catalog_round_trip() {
        use nautilus_model::types::{Price, Quantity};

        let compression = ParquetCompression::Zstd.as_parquet_compression();
        let bar_type = BarType::from("BTCUSDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL");

        let ts_1 = UnixNanos::from(utc_nanos(2024, 1, 1, 0, 0, 0, 0));
        let ts_2 = ts_1 + DurationNanos::from_secs(1);

        let bar_1 = Bar::new(
            bar_type,
            Price::from("100.00"),
            Price::from("110.00"),
            Price::from("90.00"),
            Price::from("105.00"),
            Quantity::from("1000"),
            ts_1,
            ts_1,
        );
        let bar_2 = Bar::new(
            bar_type,
            Price::from("105.00"),
            Price::from("115.00"),
            Price::from("95.00"),
            Price::from("110.00"),
            Quantity::from("1000"),
            ts_2,
            ts_2,
        );

        let temp_dir = tempfile::tempdir().unwrap();
        let data_path = temp_dir.path().join("data");

        let mut bars_map: AHashMap<BarType, Vec<Bar>> = AHashMap::new();
        let mut bars_cursors: AHashMap<BarType, DateCursor> = AHashMap::new();

        for bar in [bar_1, bar_2] {
            handle_bar_msg(
                bar,
                &mut bars_map,
                &mut bars_cursors,
                &data_path,
                compression,
            );
        }

        for (bar_type, bars) in &bars_map {
            let cursor = bars_cursors.get(bar_type).expect("Expected cursor");
            batch_and_write_bars(bars, bar_type, cursor.date_utc, &data_path, compression);
        }

        // Bars must be written under the catalog convention path ("bars"), consistent
        // with the other data types so they are discoverable by `ParquetDataCatalog`.
        assert!(data_path.join(Bar::path_prefix()).exists());
        assert!(!data_path.join("bar").exists());

        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
        let bars_out = catalog.bars(None, None, None).unwrap();

        assert_eq!(bars_out, vec![bar_1, bar_2]);
        assert_eq!(bars_out[0].bar_type, bar_type);
        assert!(bars_out[0].ts_init < bars_out[1].ts_init);
    }
}