atelier_data 0.0.15

//! Multi-source time synchronizer with configurable clock modes.
//!
//! Aligns orderbook snapshots, trades, liquidations, funding rates, and open
//! interest to a uniform time grid. Produces [`MarketSnapshot`] at each period.
//!
//! # Clock Modes
//!
//! The synchronizer supports four clock modes via [`ClockMode`]:
//!
//! - **OrderbookDriven** (default): Grid periods are triggered by orderbook
//!   timestamp crossings, matching the original `MarketSynchronizer` behavior.
//! - **TradeDriven**: Grid periods are triggered by trade timestamps.
//! - **LiquidationDriven**: Grid periods are triggered by liquidation timestamps.
//! - **ExternalClock**: Grid periods are driven by explicit `on_time()` calls
//!   with an external nanosecond timestamp.
//!
//! # Semantics
//!
//! - **Orderbook, funding rate, OI** are *state-based*: the latest value is
//!   carried forward into each period.
//! - **Trades, liquidations** are *event-based*: all events within a period
//!   are collected into a single bucket.
//!
//! # Usage
//!
//! ```ignore
//! use atelier_data::synchronizers::{MarketSynchronizer, ClockMode};
//!
//! // Orderbook-driven (default, backward compatible)
//! let mut sync = MarketSynchronizer::new(100_000_000);
//!
//! // External clock
//! let mut sync = MarketSynchronizer::external_clock(100_000_000);
//! sync.on_orderbook("BTCUSDT", ts_ms, orderbook); // updates state only
//! sync.on_trade(trade);                            // accumulates only
//! sync.on_time(ts_ns);                             // drives the grid
//!
//! // Trade-driven
//! let mut sync = MarketSynchronizer::trade_driven(100_000_000);
//! sync.on_orderbook("BTCUSDT", ts_ms, orderbook); // updates state only
//! sync.on_trade(trade);                            // accumulates + drives grid
//!
//! // Liquidation-driven
//! let mut sync = MarketSynchronizer::liquidation_driven(100_000_000);
//! sync.on_liquidation(liq);                        // accumulates + drives grid
//!
//! // At end of stream:
//! sync.finalize();
//! let snapshots: Vec<MarketSnapshot> = sync.drain();
//! ```

use crate::{
    funding::FundingRate, liquidations::Liquidation, open_interest::OpenInterest,
    orderbooks::Orderbook, snapshots::MarketSnapshot, trades::Trade,
};

use std::collections::HashMap;
use tracing::warn;

/// Maximum periods to forward-fill during gaps.
const MAX_GAP_FILL: u64 = 10_000;

// ---------------------------------------------------------------------------
// ClockMode
// ---------------------------------------------------------------------------

/// Determines which data feed drives the grid clock.
///
/// The clock mode controls which `on_*` method triggers snapshot emission
/// at grid period boundaries. Non-driver feeds accumulate data passively
/// and their contents are included in whatever snapshot period they fall into.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ClockMode {
    /// Orderbook updates drive the grid clock.
    ///
    /// This is the default mode and preserves backward compatibility.
    /// Period boundaries are detected per-symbol using the orderbook's
    /// exchange-reported millisecond timestamp.
    OrderbookDriven,

    /// Trade events drive the grid clock.
    ///
    /// Each call to [`MarketSynchronizer::on_trade()`] checks if the trade's
    /// `trade_ts` crosses a grid period boundary. The trade is accumulated
    /// *before* the boundary check, so it is included in the emitted snapshot.
    TradeDriven,

    /// Liquidation events drive the grid clock.
    ///
    /// Each call to [`MarketSynchronizer::on_liquidation()`] checks if the
    /// liquidation's `liquidation_ts` crosses a grid period boundary.
    /// The liquidation is accumulated *before* the boundary check.
    LiquidationDriven,

    /// External timestamps via [`MarketSynchronizer::on_time()`] drive the grid.
    ///
    /// All data feeds are passive; the caller explicitly advances the grid by
    /// providing a nanosecond-resolution timestamp. This decouples snapshot
    /// emission from any particular data feed's arrival rate.
    ExternalClock,
}

// ---------------------------------------------------------------------------
// FlushResult
// ---------------------------------------------------------------------------

/// Paths of files written by [`MarketSynchronizer::flush_to_parquet`].
#[derive(Debug, Clone, Default)]
pub struct FlushResult {
    /// Path to the orderbook Parquet file, if any orderbooks were present.
    pub orderbook_path: Option<std::path::PathBuf>,
    /// Path to the trades Parquet file, if any trades were present.
    pub trades_path: Option<std::path::PathBuf>,
    /// Path to the liquidations Parquet file, if any liquidations were present.
    pub liquidations_path: Option<std::path::PathBuf>,
    /// Path to the funding rates Parquet file, if any funding rates were present.
    pub funding_path: Option<std::path::PathBuf>,
    /// Path to the open interest Parquet file, if any OI records were present.
    pub open_interest_path: Option<std::path::PathBuf>,
    /// Number of snapshots that were flushed.
    pub snapshot_count: usize,
}

// ---------------------------------------------------------------------------
// MarketSynchronizer
// ---------------------------------------------------------------------------

/// Multi-source time synchronizer that produces [`MarketSnapshot`] at each
/// grid period, combining all data sources.
///
/// See [module-level documentation](self) for usage examples.
pub struct MarketSynchronizer {
    /// Grid spacing in nanoseconds.
    pub period_ns: u64,

    /// Which data feed drives the grid clock.
    clock_mode: ClockMode,

    // -- OrderbookDriven state (per-symbol tracking) -------------------------
    /// Last completed period index per symbol.
    /// Only used in [`ClockMode::OrderbookDriven`] mode.
    last_period: HashMap<String, u64>,

    // -- Global clock state (TradeDriven / LiquidationDriven / ExternalClock)
    /// Global clock period index. `None` until the first clock-driving call.
    /// Used by all modes except [`ClockMode::OrderbookDriven`].
    clock_period: Option<u64>,

    // -- Data accumulators ---------------------------------------------------
    /// Most recent orderbook snapshot per symbol.
    last_orderbook: HashMap<String, Orderbook>,

    /// Trades accumulated in the current (incomplete) period.
    current_trades: Vec<Trade>,

    /// Liquidations accumulated in the current (incomplete) period.
    current_liquidations: Vec<Liquidation>,

    /// Most recent funding rates (state-based, carried forward).
    current_funding_rates: Vec<FundingRate>,

    /// Most recent open interest records (state-based, carried forward).
    current_open_interests: Vec<OpenInterest>,

    // -- Output --------------------------------------------------------------
    /// Buffered output snapshots.
    pub buffer: Vec<MarketSnapshot>,

    /// Total snapshots produced across all drains.
    pub total_captured: usize,
}

impl MarketSynchronizer {
    // -- Constructors --------------------------------------------------------

    /// Create a new synchronizer with the default [`ClockMode::OrderbookDriven`].
    ///
    /// Backward-compatible with the pre-`ClockMode` API.
    pub fn new(period_ns: u64) -> Self {
        Self::with_clock_mode(period_ns, ClockMode::OrderbookDriven)
    }

    /// Create a new synchronizer with an explicit clock mode.
    pub fn with_clock_mode(period_ns: u64, clock_mode: ClockMode) -> Self {
        assert!(period_ns > 0, "period_ns must be positive");
        Self {
            period_ns,
            clock_mode,
            last_period: HashMap::new(),
            clock_period: None,
            last_orderbook: HashMap::new(),
            current_trades: Vec::new(),
            current_liquidations: Vec::new(),
            current_funding_rates: Vec::new(),
            current_open_interests: Vec::new(),
            buffer: Vec::new(),
            total_captured: 0,
        }
    }

    /// Convenience constructor for [`ClockMode::ExternalClock`].
    #[inline]
    pub fn external_clock(period_ns: u64) -> Self {
        Self::with_clock_mode(period_ns, ClockMode::ExternalClock)
    }

    /// Convenience constructor for [`ClockMode::TradeDriven`].
    #[inline]
    pub fn trade_driven(period_ns: u64) -> Self {
        Self::with_clock_mode(period_ns, ClockMode::TradeDriven)
    }

    /// Convenience constructor for [`ClockMode::LiquidationDriven`].
    #[inline]
    pub fn liquidation_driven(period_ns: u64) -> Self {
        Self::with_clock_mode(period_ns, ClockMode::LiquidationDriven)
    }

    /// Returns the active clock mode.
    #[inline]
    pub fn clock_mode(&self) -> ClockMode {
        self.clock_mode
    }

    // -- Private: grid advancement -------------------------------------------

    /// Core clock advancement logic for non-OrderbookDriven modes.
    ///
    /// Computes the current period from `ts_ns` (passed as `u128` to avoid
    /// overflow from ms-to-ns widening), emits snapshots for all crossed grid
    /// periods, and drains event buffers into the first emitted snapshot.
    ///
    /// Returns the number of snapshots emitted.
    fn advance_grid(&mut self, ts_ns: u128) -> usize {
        let current_period = (ts_ns / self.period_ns as u128) as u64;

        let prev_period = match self.clock_period {
            Some(p) => p,
            None => {
                self.clock_period = Some(current_period);
                return 0;
            }
        };

        if current_period <= prev_period {
            return 0;
        }

        let gap = current_period - prev_period;
        let fill_count = gap.min(MAX_GAP_FILL);

        if gap > MAX_GAP_FILL {
            warn!(
                "[clock] gap of {} periods exceeds MAX_GAP_FILL ({}), capping",
                gap, MAX_GAP_FILL,
            );
        }

        // Grab the latest orderbook state (single-symbol case is unambiguous).
        let latest_ob = self.last_orderbook.values().next().cloned();
        let trades = std::mem::take(&mut self.current_trades);
        let liquidations = std::mem::take(&mut self.current_liquidations);
        let funding_rates = std::mem::take(&mut self.current_funding_rates);
        let open_interests = std::mem::take(&mut self.current_open_interests);

        for p in (prev_period + 1)..=(prev_period + fill_count) {
            let ts = p * self.period_ns;

            // First period gets accumulated events; gap-fill periods get empty.
            let snap = if p == prev_period + 1 {
                MarketSnapshot {
                    ts_ns: ts,
                    orderbook: latest_ob.clone().map(|mut ob| {
                        ob.orderbook_ts = ts;
                        ob
                    }),
                    trades: trades.clone(),
                    liquidations: liquidations.clone(),
                    funding_rate: funding_rates.clone(),
                    open_interest: open_interests.clone(),
                }
            } else {
                MarketSnapshot {
                    ts_ns: ts,
                    orderbook: latest_ob.clone().map(|mut ob| {
                        ob.orderbook_ts = ts;
                        ob
                    }),
                    trades: Vec::new(),
                    liquidations: Vec::new(),
                    funding_rate: Vec::new(),
                    open_interest: Vec::new(),
                }
            };

            self.buffer.push(snap);
        }

        self.clock_period = Some(current_period);
        self.total_captured += fill_count as usize;
        fill_count as usize
    }

    // -- Public: clock drivers -----------------------------------------------

    /// Advance the grid clock to `ts_ns` (nanoseconds).
    ///
    /// Only effective in [`ClockMode::ExternalClock`] mode. In other modes
    /// this is a no-op that returns 0.
    ///
    /// Emits snapshots for all grid periods crossed since the last clock
    /// advance. The clock is independent of any data feed — orderbook,
    /// trades, etc. are accumulated passively and drained into the emitted
    /// snapshot(s).
    ///
    /// Returns the number of snapshots emitted.
    pub fn on_time(&mut self, ts_ns: u64) -> usize {
        if self.clock_mode != ClockMode::ExternalClock {
            return 0;
        }
        self.advance_grid(ts_ns as u128)
    }

    /// Feed an orderbook snapshot.
    ///
    /// - In [`ClockMode::OrderbookDriven`]: this is the clock driver. Grid
    ///   periods are triggered by orderbook timestamp crossings (per-symbol).
    /// - In all other modes: only updates the latest orderbook state for
    ///   the given symbol, without advancing the grid.
    ///
    /// `exchange_ts_ms` is the exchange-reported timestamp in milliseconds.
    ///
    /// Returns the number of snapshots emitted (always 0 in non-OB modes).
    pub fn on_orderbook(
        &mut self,
        symbol: &str,
        exchange_ts_ms: u64,
        snapshot: Orderbook,
    ) -> usize {
        if self.clock_mode != ClockMode::OrderbookDriven {
            // Non-OB mode: passive state update only.
            self.last_orderbook.insert(symbol.to_string(), snapshot);
            return 0;
        }

        // --- OrderbookDriven: original per-symbol grid logic ----------------
        let ts_ns = exchange_ts_ms as u128 * 1_000_000;
        let current_period = (ts_ns / self.period_ns as u128) as u64;

        let prev_period = match self.last_period.get(symbol) {
            Some(&p) => p,
            None => {
                self.last_period.insert(symbol.to_string(), current_period);
                self.last_orderbook.insert(symbol.to_string(), snapshot);
                return 0;
            }
        };

        if current_period <= prev_period {
            // Still in the same period — update latest state.
            self.last_orderbook.insert(symbol.to_string(), snapshot);
            return 0;
        }

        // Grid boundary crossed — emit snapshot(s) for completed period(s).
        let gap = current_period - prev_period;
        let fill_count = gap.min(MAX_GAP_FILL);

        if gap > MAX_GAP_FILL {
            warn!(
                "[{}] gap of {} periods exceeds MAX_GAP_FILL ({}), capping",
                symbol, gap, MAX_GAP_FILL,
            );
        }

        // Emit MarketSnapshot for each completed period.
        let prev_ob = self.last_orderbook.get(symbol).cloned();
        let trades = std::mem::take(&mut self.current_trades);
        let liquidations = std::mem::take(&mut self.current_liquidations);
        let funding_rates = std::mem::take(&mut self.current_funding_rates);
        let open_interests = std::mem::take(&mut self.current_open_interests);

        for p in (prev_period + 1)..=(prev_period + fill_count) {
            let ts = p * self.period_ns;

            // First period gets accumulated events; gap-fill periods get empty.
            let snap = if p == prev_period + 1 {
                MarketSnapshot {
                    ts_ns: ts,
                    orderbook: prev_ob.clone().map(|mut ob| {
                        ob.orderbook_ts = ts;
                        ob
                    }),
                    trades: trades.clone(),
                    liquidations: liquidations.clone(),
                    funding_rate: funding_rates.clone(),
                    open_interest: open_interests.clone(),
                }
            } else {
                // Gap-fill: carry forward state, no events.
                MarketSnapshot {
                    ts_ns: ts,
                    orderbook: prev_ob.clone().map(|mut ob| {
                        ob.orderbook_ts = ts;
                        ob
                    }),
                    trades: Vec::new(),
                    liquidations: Vec::new(),
                    funding_rate: Vec::new(),
                    open_interest: Vec::new(),
                }
            };

            self.buffer.push(snap);
        }

        self.last_period.insert(symbol.to_string(), current_period);
        self.last_orderbook.insert(symbol.to_string(), snapshot);
        self.total_captured += fill_count as usize;
        fill_count as usize
    }

    /// Feed a trade event.
    ///
    /// The trade is always accumulated in the current period's buffer.
    ///
    /// - In [`ClockMode::TradeDriven`]: also checks if the trade's `trade_ts`
    ///   crosses a grid period boundary. The trade is included in the emitted
    ///   snapshot because it is accumulated *before* the boundary check.
    /// - In all other modes: accumulation only.
    ///
    /// Returns the number of snapshots emitted (always 0 in non-Trade modes).
    pub fn on_trade(&mut self, trade: Trade) -> usize {
        let ts_ms = trade.trade_ts;
        self.current_trades.push(trade);

        if self.clock_mode == ClockMode::TradeDriven {
            self.advance_grid(ts_ms as u128 * 1_000_000)
        } else {
            0
        }
    }

    /// Feed a liquidation event.
    ///
    /// The liquidation is always accumulated in the current period's buffer.
    ///
    /// - In [`ClockMode::LiquidationDriven`]: also checks if the liquidation's
    ///   `liquidation_ts` crosses a grid period boundary. The liquidation is
    ///   included in the emitted snapshot.
    /// - In all other modes: accumulation only.
    ///
    /// Returns the number of snapshots emitted (always 0 in non-Liq modes).
    pub fn on_liquidation(&mut self, liq: Liquidation) -> usize {
        let ts_ms = liq.liquidation_ts;
        self.current_liquidations.push(liq);

        if self.clock_mode == ClockMode::LiquidationDriven {
            self.advance_grid(ts_ms as u128 * 1_000_000)
        } else {
            0
        }
    }

    /// Feed a funding rate update. State-based — latest value is carried forward.
    ///
    /// Funding rates never drive the clock, regardless of mode.
    pub fn on_funding(&mut self, fr: FundingRate) {
        self.current_funding_rates.push(fr);
    }

    /// Feed an open interest update. State-based — latest value is carried forward.
    ///
    /// Open interest never drives the clock, regardless of mode.
    pub fn on_open_interest(&mut self, oi: OpenInterest) {
        self.current_open_interests.push(oi);
    }

    // -- Finalization --------------------------------------------------------

    /// Emit the final snapshot, closing the current (incomplete) period.
    /// Call when the data stream ends.
    ///
    /// In [`ClockMode::OrderbookDriven`], emits one snapshot per tracked symbol.
    /// In all other modes, emits a single final snapshot from the global clock.
    pub fn finalize(&mut self) {
        match self.clock_mode {
            ClockMode::OrderbookDriven => self.finalize_ob_driven(),
            _ => self.finalize_global_clock(),
        }
    }

    /// Finalize logic for [`ClockMode::OrderbookDriven`] — one snapshot per
    /// tracked symbol using the per-symbol period map.
    fn finalize_ob_driven(&mut self) {
        let periods: Vec<(String, u64)> = self
            .last_period
            .iter()
            .map(|(s, &p)| (s.clone(), p))
            .collect();

        let trades = std::mem::take(&mut self.current_trades);
        let liquidations = std::mem::take(&mut self.current_liquidations);
        let funding_rates = std::mem::take(&mut self.current_funding_rates);
        let open_interests = std::mem::take(&mut self.current_open_interests);

        for (symbol, period) in periods {
            let ts = (period + 1) * self.period_ns;

            let snap = MarketSnapshot {
                ts_ns: ts,
                orderbook: self.last_orderbook.get(&symbol).cloned().map(|mut ob| {
                    ob.orderbook_ts = ts;
                    ob
                }),
                trades: trades.clone(),
                liquidations: liquidations.clone(),
                funding_rate: funding_rates.clone(),
                open_interest: open_interests.clone(),
            };

            self.buffer.push(snap);
            self.total_captured += 1;
        }
    }

    /// Finalize logic for global-clock modes — single final snapshot from
    /// the shared `clock_period`.
    fn finalize_global_clock(&mut self) {
        let period = match self.clock_period {
            Some(p) => p,
            None => return, // Never received a clock-driving event.
        };

        let ts = (period + 1) * self.period_ns;
        let latest_ob = self.last_orderbook.values().next().cloned();
        let trades = std::mem::take(&mut self.current_trades);
        let liquidations = std::mem::take(&mut self.current_liquidations);
        let funding_rates = std::mem::take(&mut self.current_funding_rates);
        let open_interests = std::mem::take(&mut self.current_open_interests);

        let snap = MarketSnapshot {
            ts_ns: ts,
            orderbook: latest_ob.map(|mut ob| {
                ob.orderbook_ts = ts;
                ob
            }),
            trades,
            liquidations,
            funding_rate: funding_rates,
            open_interest: open_interests,
        };

        self.buffer.push(snap);
        self.total_captured += 1;
    }

    // -- Buffer access -------------------------------------------------------

    /// Drain the buffer, returning all accumulated snapshots.
    pub fn drain(&mut self) -> Vec<MarketSnapshot> {
        std::mem::take(&mut self.buffer)
    }

    #[inline]
    pub fn buffer_len(&self) -> usize {
        self.buffer.len()
    }

    #[inline]
    pub fn total_captured(&self) -> usize {
        self.total_captured
    }

    // -- Aggregation ---------------------------------------------------------

    /// Compute per-snapshot aggregated statistics from the buffer.
    ///
    /// Drains the buffer (like [`drain()`](Self::drain)) and returns one
    /// [`MarketAggregate`](crate::snapshots::aggregate::MarketAggregate) per
    /// [`MarketSnapshot`].
    ///
    /// Open interest change (`oi_change`) is computed relative to the
    /// previous snapshot in sequence; the first snapshot uses 0.0.
    pub fn market_aggregate(
        &mut self,
    ) -> Vec<crate::snapshots::aggregate::MarketAggregate> {
        use crate::snapshots::aggregate::MarketAggregate;

        let snapshots = std::mem::take(&mut self.buffer);
        let mut aggregates = Vec::with_capacity(snapshots.len());
        let mut prev_oi: f64 = 0.0;

        for snap in &snapshots {
            let agg = MarketAggregate::from_snapshot(snap, prev_oi);
            prev_oi = agg.oi_contracts;
            aggregates.push(agg);
        }

        aggregates
    }

    // -- Parquet persistence -------------------------------------------------

    /// Flush all buffered [`MarketSnapshot`]s to per-type Parquet files.
    ///
    /// Drains the buffer, decomposes each snapshot into its constituent data
    /// types, and writes one Parquet file per non-empty data type into
    /// `output_dir`. Each file uses the snapshot's `ts_ns` as its timestamp
    /// column, ensuring a common time index for downstream joins.
    ///
    /// Returns a [`FlushResult`] with the paths of all written files.
    ///
    /// # Feature Flag
    ///
    /// Requires `--features parquet`.
    #[cfg(feature = "parquet")]
    pub fn flush_to_parquet(
        &mut self,
        output_dir: &std::path::Path,
    ) -> anyhow::Result<FlushResult> {
        use crate::{
            funding::io::funding_parquet::write_funding_parquet_timestamped,
            liquidations::io::liq_parquet::write_liquidations_parquet_timestamped,
            open_interest::io::oi_parquet::write_oi_parquet_timestamped,
            orderbooks::io::ob_parquet::write_ob_parquet,
            trades::io::trades_parquet::write_trades_parquet_timestamped,
        };

        let snapshots = std::mem::take(&mut self.buffer);
        let n = snapshots.len();

        let mut orderbooks: Vec<Orderbook> = Vec::new();
        let mut trades: Vec<Trade> = Vec::new();
        let mut liquidations: Vec<Liquidation> = Vec::new();
        let mut funding_rates: Vec<FundingRate> = Vec::new();
        let mut open_interests: Vec<OpenInterest> = Vec::new();

        for snap in &snapshots {
            if let Some(ob) = &snap.orderbook {
                orderbooks.push(ob.clone());
            }
            trades.extend(snap.trades.iter().cloned());
            liquidations.extend(snap.liquidations.iter().cloned());
            funding_rates.extend(snap.funding_rate.iter().cloned());
            open_interests.extend(snap.open_interest.iter().cloned());
        }

        let mut result = FlushResult {
            snapshot_count: n,
            ..Default::default()
        };

        if !orderbooks.is_empty() {
            let orderbooks_dir = output_dir.join("orderbooks");
            std::fs::create_dir_all(&orderbooks_dir)?;
            let path = write_ob_parquet(&orderbooks, &orderbooks_dir, "sync")?;
            result.orderbook_path = Some(path);
        }
        if !trades.is_empty() {
            let trades_dir = output_dir.join("trades");
            std::fs::create_dir_all(&trades_dir)?;
            let path = write_trades_parquet_timestamped(&trades, &trades_dir, "sync")?;
            result.trades_path = Some(path);
        }
        if !liquidations.is_empty() {
            let liquidations_dir = output_dir.join("liquidations");
            std::fs::create_dir_all(&liquidations_dir)?;
            let path =
                write_liquidations_parquet_timestamped(&liquidations, &liquidations_dir, "sync")?;
            result.liquidations_path = Some(path);
        }
        if !funding_rates.is_empty() {
            let fundings_dir = output_dir.join("fundings");
            std::fs::create_dir_all(&fundings_dir)?;
            let path = write_funding_parquet_timestamped(&funding_rates, &fundings_dir, "sync")?;
            result.funding_path = Some(path);
        }
        if !open_interests.is_empty() {
            let open_interests_dir = output_dir.join("open_interests");
            std::fs::create_dir_all(&open_interests_dir)?;
            let path =
                write_oi_parquet_timestamped(&open_interests, &open_interests_dir, "sync")?;
            result.open_interest_path = Some(path);
        }

        Ok(result)
    }

    /// Stub for when the `parquet` feature is not enabled.
    #[cfg(not(feature = "parquet"))]
    pub fn flush_to_parquet(
        &mut self,
        _output_dir: &std::path::Path,
    ) -> Result<FlushResult, crate::errors::PersistError> {
        Err(crate::errors::PersistError::UnsupportedFormat(
            "parquet support not compiled in (enable 'parquet' feature)".to_string(),
        ))
    }

    /// Compute aggregated statistics and write to a Parquet file.
    ///
    /// Combines [`market_aggregate()`](Self::market_aggregate) and
    /// [`write_market_aggregate_parquet`](crate::snapshots::aggregate::write_market_aggregate_parquet)
    /// into a single call.
    ///
    /// Returns the path to the written Parquet file.
    #[cfg(feature = "parquet")]
    pub fn flush_aggregate_to_parquet(
        &mut self,
        output_dir: &std::path::Path,
    ) -> Result<std::path::PathBuf, crate::errors::PersistError> {
        use crate::snapshots::aggregate::write_market_aggregate_parquet;
        let aggregates = self.market_aggregate();
        write_market_aggregate_parquet(&aggregates, output_dir)
    }

    /// Stub for when the `parquet` feature is not enabled.
    #[cfg(not(feature = "parquet"))]
    pub fn flush_aggregate_to_parquet(
        &mut self,
        _output_dir: &std::path::Path,
    ) -> Result<std::path::PathBuf, crate::errors::PersistError> {
        Err(crate::errors::PersistError::UnsupportedFormat(
            "parquet support not compiled in (enable 'parquet' feature)".to_string(),
        ))
    }
}