opendeviationbar_core/

processor.rs

// FILE-SIZE-OK: 1496 lines — tests are inline (access private OpenDeviationBarState)
//! Core open deviation bar processing algorithm
//!
//! Implements non-lookahead bias open deviation bar construction where bars close when
//! price moves ±threshold dbps from the bar's OPEN price.

use crate::checkpoint::{
    AnomalySummary, Checkpoint, CheckpointError, PositionVerification, PriceWindow,
};
use crate::fixed_point::FixedPoint;
use crate::interbar::{InterBarConfig, TradeHistory}; // Issue #59
// Issue #59: Intra-bar features - using compute_intra_bar_features_with_config (Issue #128)
use crate::types::{AggTrade, OpenDeviationBar};
use smallvec::SmallVec; // Issue #119: Trade accumulation with inline buffer (512 slots ≈ 29KB)
#[cfg(feature = "python")]
use pyo3::prelude::*;
// Re-export ProcessingError from errors.rs (Phase 2a extraction)
pub use crate::errors::ProcessingError;
// Re-export ExportOpenDeviationBarProcessor from export_processor.rs (Phase 2d extraction)
pub use crate::export_processor::ExportOpenDeviationBarProcessor;

/// Open deviation bar processor with non-lookahead bias guarantee
pub struct OpenDeviationBarProcessor {
    /// Threshold in decimal basis points (250 = 25bps, v3.0.0+)
    threshold_decimal_bps: u32,

    /// Issue #96 Task #98: Pre-computed threshold ratio for fast delta calculation
    /// Stores (threshold_decimal_bps * SCALE) / BASIS_POINTS_SCALE as fixed-point
    /// This allows compute_range_thresholds() to compute delta = (price * ratio) / SCALE
    /// without repeated division, avoiding i128 arithmetic in hot path.
    /// Performance: Eliminates BASIS_POINTS_SCALE division on every bar creation.
    /// Made public for testing purposes.
    pub threshold_ratio: i64,

    /// Current bar state for streaming processing (Q19)
    /// Enables get_incomplete_bar() and stateful process_single_trade()
    current_bar_state: Option<OpenDeviationBarState>,

    /// Price window for checkpoint hash verification
    price_window: PriceWindow,

    /// Last processed trade ID (for gap detection on resume)
    last_trade_id: Option<i64>,

    /// Last processed timestamp (for position verification)
    last_timestamp_us: i64,

    /// Anomaly tracking for debugging
    anomaly_summary: AnomalySummary,

    /// Flag indicating this processor was created from a checkpoint
    /// When true, process_agg_trade_records will continue from existing bar state
    resumed_from_checkpoint: bool,

    /// Prevent bars from closing on same timestamp as they opened (Issue #36)
    ///
    /// When true (default): A bar cannot close until a trade arrives with a
    /// different timestamp than the bar's open_time. This prevents "instant bars"
    /// during flash crashes where multiple trades occur at the same millisecond.
    ///
    /// When false: Legacy behavior - bars can close on any breach regardless
    /// of timestamp, which may produce bars with identical timestamps.
    prevent_same_timestamp_close: bool,

    /// Deferred bar open flag (Issue #46)
    ///
    /// When true: The previous trade triggered a threshold breach and closed a bar.
    /// The next trade arriving via `process_single_trade()` should open a new bar
    /// instead of being treated as a continuation.
    ///
    /// This matches the batch path's `defer_open` semantics in
    /// `process_agg_trade_records()` where the breaching trade closes the current
    /// bar and the NEXT trade opens the new bar.
    defer_open: bool,

    /// Trade history for inter-bar feature computation (Issue #59)
    ///
    /// Ring buffer of recent trades for computing lookback-based features.
    /// When Some, features are computed from trades BEFORE each bar's open_time.
    /// When None, inter-bar features are disabled (all lookback_* fields = None).
    trade_history: Option<TradeHistory>,

    /// Configuration for inter-bar features (Issue #59)
    ///
    /// Controls lookback mode (fixed count or time window) and which feature
    /// tiers to compute. When None, inter-bar features are disabled.
    inter_bar_config: Option<InterBarConfig>,

    /// Enable intra-bar feature computation (Issue #59)
    ///
    /// When true, the processor accumulates trades during bar construction
    /// and computes 22 features from trades WITHIN each bar at bar close.
    /// Features include ITH (Investment Time Horizon), statistical, and
    /// complexity metrics. When false, all intra_* fields are None.
    include_intra_bar_features: bool,

    /// Issue #128: Configuration for intra-bar feature computation.
    /// Controls which complexity features (Hurst, PE) are computed.
    intra_bar_config: crate::intrabar::IntraBarConfig,

    /// Issue #112: Maximum timestamp gap in microseconds before discarding a forming bar
    ///
    /// When resuming from checkpoint with a forming bar, if the gap between
    /// the forming bar's close_time and the first incoming trade exceeds this
    /// threshold, the forming bar is discarded as an orphan (same as ouroboros reset).
    /// This prevents "oversized" bars caused by large data gaps (e.g., 38-hour outages).
    ///
    /// Default: 3,600,000,000 μs (1 hour)
    max_gap_us: i64,
}

/// Cold path: scan trades to find first unsorted pair and return error
/// Extracted from validate_trade_ordering() to improve hot-path code layout
#[cold]
#[inline(never)]
fn find_unsorted_trade(trades: &[AggTrade]) -> Result<(), ProcessingError> {
    for i in 1..trades.len() {
        let prev = &trades[i - 1];
        let curr = &trades[i];
        if curr.timestamp < prev.timestamp
            || (curr.timestamp == prev.timestamp && curr.agg_trade_id <= prev.agg_trade_id)
        {
            return Err(ProcessingError::UnsortedTrades {
                index: i,
                prev_time: prev.timestamp,
                prev_id: prev.agg_trade_id,
                curr_time: curr.timestamp,
                curr_id: curr.agg_trade_id,
            });
        }
    }
    Ok(())
}

/// Cold path: construct unsorted trade error
/// Extracted to keep error construction out of the hot validation loop
#[cold]
#[inline(never)]
fn unsorted_trade_error(index: usize, prev: &AggTrade, curr: &AggTrade) -> Result<(), ProcessingError> {
    Err(ProcessingError::UnsortedTrades {
        index,
        prev_time: prev.timestamp,
        prev_id: prev.agg_trade_id,
        curr_time: curr.timestamp,
        curr_id: curr.agg_trade_id,
    })
}

impl OpenDeviationBarProcessor {
    /// Create new processor with given threshold
    ///
    /// Uses default behavior: `prevent_same_timestamp_close = true` (Issue #36)
    ///
    /// # Arguments
    ///
    /// * `threshold_decimal_bps` - Threshold in **decimal basis points**
    ///   - Example: `250` → 25bps = 0.25%
    ///   - Example: `10` → 1bps = 0.01%
    ///   - Minimum: `1` → 0.1bps = 0.001%
    ///
    /// # Breaking Change (v3.0.0)
    ///
    /// Prior to v3.0.0, `threshold_decimal_bps` was in 1bps units.
    /// **Migration**: Multiply all threshold values by 10.
    pub fn new(threshold_decimal_bps: u32) -> Result<Self, ProcessingError> {
        Self::with_options(threshold_decimal_bps, true)
    }

    /// Create new processor with explicit timestamp gating control
    ///
    /// # Arguments
    ///
    /// * `threshold_decimal_bps` - Threshold in **decimal basis points**
    /// * `prevent_same_timestamp_close` - If true, bars cannot close until
    ///   timestamp advances from open_time. This prevents "instant bars" during
    ///   flash crashes. Set to false for legacy behavior (pre-v9).
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Default behavior (v9+): timestamp gating enabled
    /// let processor = OpenDeviationBarProcessor::new(250)?;
    ///
    /// // Legacy behavior: allow instant bars
    /// let processor = OpenDeviationBarProcessor::with_options(250, false)?;
    /// ```
    pub fn with_options(
        threshold_decimal_bps: u32,
        prevent_same_timestamp_close: bool,
    ) -> Result<Self, ProcessingError> {
        // Validation bounds (v3.0.0: dbps units)
        // Min: 1 dbps = 0.001%
        // Max: 100,000 dbps = 100%
        if threshold_decimal_bps < 1 {
            return Err(ProcessingError::InvalidThreshold {
                threshold_decimal_bps,
            });
        }
        if threshold_decimal_bps > 100_000 {
            return Err(ProcessingError::InvalidThreshold {
                threshold_decimal_bps,
            });
        }

        // Issue #96 Task #98: Pre-compute threshold ratio
        // Ratio = (threshold_decimal_bps * SCALE) / BASIS_POINTS_SCALE
        // This is used in compute_range_thresholds() for fast delta calculation
        let threshold_ratio = ((threshold_decimal_bps as i64) * crate::fixed_point::SCALE)
            / (crate::fixed_point::BASIS_POINTS_SCALE as i64);

        Ok(Self {
            threshold_decimal_bps,
            threshold_ratio,
            current_bar_state: None,
            price_window: PriceWindow::new(),
            last_trade_id: None,
            last_timestamp_us: 0,
            anomaly_summary: AnomalySummary::default(),
            resumed_from_checkpoint: false,
            prevent_same_timestamp_close,
            defer_open: false,
            trade_history: None,               // Issue #59: disabled by default
            inter_bar_config: None,            // Issue #59: disabled by default
            include_intra_bar_features: false, // Issue #59: disabled by default
            intra_bar_config: crate::intrabar::IntraBarConfig::default(), // Issue #128
            max_gap_us: 3_600_000_000,         // Issue #112: 1 hour default
        })
    }

    /// Get the prevent_same_timestamp_close setting
    pub fn prevent_same_timestamp_close(&self) -> bool {
        self.prevent_same_timestamp_close
    }

    /// Enable inter-bar feature computation with the given configuration (Issue #59)
    ///
    /// When enabled, the processor maintains a trade history buffer and computes
    /// lookback-based microstructure features on each bar close. Features are
    /// computed from trades that occurred BEFORE each bar's open_time, ensuring
    /// no lookahead bias.
    ///
    /// Uses a local entropy cache (default behavior, backward compatible).
    /// For multi-symbol workloads, use `with_inter_bar_config_and_cache()` with a global cache.
    ///
    /// # Arguments
    ///
    /// * `config` - Configuration controlling lookback mode and feature tiers
    ///
    /// # Example
    ///
    /// ```ignore
    /// use opendeviationbar_core::processor::OpenDeviationBarProcessor;
    /// use opendeviationbar_core::interbar::{InterBarConfig, LookbackMode};
    ///
    /// let processor = OpenDeviationBarProcessor::new(1000)?
    ///     .with_inter_bar_config(InterBarConfig {
    ///         lookback_mode: LookbackMode::FixedCount(500),
    ///         compute_tier2: true,
    ///         compute_tier3: true,
    ///         ..Default::default()
    ///     });
    /// ```
    pub fn with_inter_bar_config(self, config: InterBarConfig) -> Self {
        self.with_inter_bar_config_and_cache(config, None)
    }

    /// Enable inter-bar feature computation with optional external entropy cache
    ///
    /// Issue #145 Phase 3: Multi-Symbol Entropy Cache Sharing
    ///
    /// # Arguments
    ///
    /// * `config` - Configuration controlling lookback mode and feature tiers
    /// * `external_cache` - Optional shared entropy cache from `get_global_entropy_cache()`
    ///   - If provided: Uses the shared global cache (recommended for multi-symbol)
    ///   - If None: Creates a local 128-entry cache (default, backward compatible)
    ///
    /// # Usage
    ///
    /// ```ignore
    /// use opendeviationbar_core::{processor::OpenDeviationBarProcessor, entropy_cache_global::get_global_entropy_cache, interbar::InterBarConfig};
    ///
    /// // Single-symbol: use local cache (default)
    /// let processor = OpenDeviationBarProcessor::new(1000)?
    ///     .with_inter_bar_config(config);
    ///
    /// // Multi-symbol: share global cache
    /// let global_cache = get_global_entropy_cache();
    /// let processor = OpenDeviationBarProcessor::new(1000)?
    ///     .with_inter_bar_config_and_cache(config, Some(global_cache));
    /// ```
    pub fn with_inter_bar_config_and_cache(
        mut self,
        config: InterBarConfig,
        external_cache: Option<std::sync::Arc<parking_lot::RwLock<crate::interbar_math::EntropyCache>>>,
    ) -> Self {
        self.trade_history = Some(TradeHistory::new_with_cache(config.clone(), external_cache));
        self.inter_bar_config = Some(config);
        self
    }

    /// Check if inter-bar features are enabled
    pub fn inter_bar_enabled(&self) -> bool {
        self.inter_bar_config.is_some()
    }

    /// Issue #112: Configure maximum timestamp gap for checkpoint recovery
    ///
    /// When resuming from checkpoint with a forming bar, if the gap between
    /// the forming bar's close_time and the first incoming trade exceeds this
    /// threshold, the forming bar is discarded as an orphan.
    ///
    /// # Arguments
    ///
    /// * `max_gap_us` - Maximum gap in microseconds (default: 3,600,000,000 = 1 hour)
    pub fn with_max_gap(mut self, max_gap_us: i64) -> Self {
        self.max_gap_us = max_gap_us;
        self
    }

    /// Get the maximum gap threshold in microseconds
    pub fn max_gap_us(&self) -> i64 {
        self.max_gap_us
    }

    /// Enable intra-bar feature computation (Issue #59)
    ///
    /// When enabled, the processor accumulates trades during bar construction
    /// and computes 22 features from trades WITHIN each bar at bar close:
    /// - 8 ITH features (Investment Time Horizon)
    /// - 12 statistical features (OFI, intensity, Kyle lambda, etc.)
    /// - 2 complexity features (Hurst exponent, permutation entropy)
    ///
    /// # Memory Note
    ///
    /// Trades are accumulated per-bar and freed when the bar closes.
    /// Typical 1000 dbps bar: ~50-500 trades, ~2-24 KB overhead.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let processor = OpenDeviationBarProcessor::new(1000)?
    ///     .with_intra_bar_features();
    /// ```
    pub fn with_intra_bar_features(mut self) -> Self {
        self.include_intra_bar_features = true;
        self
    }

    /// Check if intra-bar features are enabled
    pub fn intra_bar_enabled(&self) -> bool {
        self.include_intra_bar_features
    }

    /// Re-enable inter-bar features on an existing processor (Issue #97).
    ///
    /// Used after `from_checkpoint()` to restore microstructure config that
    /// is not preserved in checkpoint state. Uses a local entropy cache by default.
    /// For multi-symbol workloads, use `set_inter_bar_config_with_cache()` with a global cache.
    pub fn set_inter_bar_config(&mut self, config: InterBarConfig) {
        self.set_inter_bar_config_with_cache(config, None);
    }

    /// Re-enable inter-bar features with optional external entropy cache (Issue #145 Phase 3).
    ///
    /// Used after `from_checkpoint()` to restore microstructure config that
    /// is not preserved in checkpoint state. Allows specifying a shared entropy cache
    /// for multi-symbol processors.
    pub fn set_inter_bar_config_with_cache(
        &mut self,
        config: InterBarConfig,
        external_cache: Option<std::sync::Arc<parking_lot::RwLock<crate::interbar_math::EntropyCache>>>,
    ) {
        self.trade_history = Some(TradeHistory::new_with_cache(config.clone(), external_cache));
        self.inter_bar_config = Some(config);
    }

    /// Re-enable intra-bar features on an existing processor (Issue #97).
    pub fn set_intra_bar_features(&mut self, enabled: bool) {
        self.include_intra_bar_features = enabled;
    }

    /// Issue #128: Set intra-bar feature configuration.
    /// Controls which complexity features (Hurst, PE) are computed.
    pub fn with_intra_bar_config(mut self, config: crate::intrabar::IntraBarConfig) -> Self {
        self.intra_bar_config = config;
        self
    }

    /// Issue #128: Set intra-bar feature configuration on existing processor.
    pub fn set_intra_bar_config(&mut self, config: crate::intrabar::IntraBarConfig) {
        self.intra_bar_config = config;
    }

    /// Process a single trade and return completed bar if any
    ///
    /// Maintains internal state for streaming use case. State persists across calls
    /// until a bar completes (threshold breach), enabling get_incomplete_bar().
    ///
    /// # Arguments
    ///
    /// * `trade` - Single aggregated trade to process
    ///
    /// # Returns
    ///
    /// `Some(OpenDeviationBar)` if a bar was completed, `None` otherwise
    ///
    /// # State Management
    ///
    /// - First trade: Initializes new bar state
    /// - Subsequent trades: Updates existing bar or closes on breach
    /// - Breach: Returns completed bar, starts new bar with breaching trade
    ///
    /// Issue #96 Task #78: Accept borrowed AggTrade to eliminate clones in fan-out loops.
    /// Streaming pipelines (4+ thresholds) were cloning ~57 byte trades per processor.
    /// Signature change to `&AggTrade` eliminates 4-8x unnecessary allocations.
    /// Issue #96 Task #84: `#[inline]` — main hot-path entry point called on every trade.
    #[inline]
    pub fn process_single_trade(
        &mut self,
        trade: &AggTrade,
    ) -> Result<Option<OpenDeviationBar>, ProcessingError> {
        // Track price and position for checkpoint
        self.price_window.push(trade.price);
        self.last_trade_id = Some(trade.agg_trade_id);
        self.last_timestamp_us = trade.timestamp;

        // Issue #59: Push trade to history buffer for inter-bar feature computation
        // This must happen BEFORE bar processing so lookback window includes recent trades
        if let Some(ref mut history) = self.trade_history {
            history.push(trade);
        }

        // Issue #46: If previous call triggered a breach, this trade opens the new bar.
        // This matches the batch path's defer_open semantics - the breaching trade
        // closes the current bar, and the NEXT trade opens the new bar.
        if self.defer_open {
            // Issue #68: Notify history that new bar is opening (preserves pre-bar trades)
            if let Some(ref mut history) = self.trade_history {
                history.on_bar_open(trade.timestamp);
            }
            self.current_bar_state = Some(if self.include_intra_bar_features {
                OpenDeviationBarState::new_with_trade_accumulation(trade, self.threshold_ratio)
            } else {
                OpenDeviationBarState::new(trade, self.threshold_ratio)
            });
            self.defer_open = false;
            return Ok(None);
        }

        match &mut self.current_bar_state {
            None => {
                // First trade - initialize new bar
                // Issue #68: Notify history that new bar is opening (preserves pre-bar trades)
                if let Some(ref mut history) = self.trade_history {
                    history.on_bar_open(trade.timestamp);
                }
                self.current_bar_state = Some(if self.include_intra_bar_features {
                    OpenDeviationBarState::new_with_trade_accumulation(trade, self.threshold_ratio)
                } else {
                    OpenDeviationBarState::new(trade, self.threshold_ratio)
                });
                Ok(None)
            }
            Some(bar_state) => {
                // Issue #59 & #96 Task #44: Accumulate trade for intra-bar features (before breach check)
                // Only accumulates if features enabled, avoiding unnecessary clones
                bar_state.accumulate_trade(trade, self.include_intra_bar_features);

                // Check for threshold breach
                let price_breaches = bar_state.bar.is_breach(
                    trade.price,
                    bar_state.upper_threshold,
                    bar_state.lower_threshold,
                );

                // Timestamp gate (Issue #36): prevent bars from closing on same timestamp
                // This eliminates "instant bars" during flash crashes where multiple trades
                // occur at the same millisecond.
                let timestamp_allows_close = !self.prevent_same_timestamp_close
                    || trade.timestamp != bar_state.bar.open_time;

                if price_breaches && timestamp_allows_close {
                    // Breach detected AND timestamp changed - close current bar
                    bar_state.bar.update_with_trade(trade);

                    // Validation: Ensure high/low include open/close extremes
                    debug_assert!(
                        bar_state.bar.high >= bar_state.bar.open.max(bar_state.bar.close)
                    );
                    debug_assert!(bar_state.bar.low <= bar_state.bar.open.min(bar_state.bar.close));

                    // Compute microstructure features at bar finalization (Issue #25)
                    bar_state.bar.compute_microstructure_features();

                    // Issue #59: Compute inter-bar features from lookback window
                    // Features are computed from trades BEFORE bar.open_time (no lookahead)
                    if let Some(ref mut history) = self.trade_history {
                        let inter_bar_features = history.compute_features(bar_state.bar.open_time);
                        bar_state.bar.set_inter_bar_features(&inter_bar_features);
                        // Issue #68: Notify history that bar is closing (resumes normal pruning)
                        history.on_bar_close();
                    }

                    // Issue #59: Compute intra-bar features from accumulated trades
                    if self.include_intra_bar_features {
                        // Issue #96 Task #173: Use reusable scratch buffers from bar_state
                        let intra_bar_features = crate::intrabar::compute_intra_bar_features_with_config(
                            &bar_state.accumulated_trades,
                            &mut bar_state.scratch_prices,
                            &mut bar_state.scratch_volumes,
                            &self.intra_bar_config, // Issue #128
                        );
                        bar_state.bar.set_intra_bar_features(&intra_bar_features);
                    }

                    // Move bar out instead of cloning — bar_state borrow ends after
                    // last use above (NLL), so take() is safe here.
                    let completed_bar = self.current_bar_state.take().unwrap().bar;

                    // Issue #46: Don't start new bar with breaching trade.
                    // Next trade will open the new bar via defer_open.
                    self.defer_open = true;

                    Ok(Some(completed_bar))
                } else {
                    // Either no breach OR same timestamp (gate active) - update existing bar
                    bar_state.bar.update_with_trade(trade);
                    Ok(None)
                }
            }
        }
    }

    /// Get any incomplete bar currently being processed
    ///
    /// Returns clone of current bar state for inspection without consuming it.
    /// Useful for final bar at stream end or progress monitoring.
    ///
    /// # Returns
    ///
    /// `Some(OpenDeviationBar)` if bar is in progress, `None` if no active bar
    pub fn get_incomplete_bar(&self) -> Option<OpenDeviationBar> {
        self.current_bar_state
            .as_ref()
            .map(|state| state.bar.clone())
    }

    /// Process AggTrade records into open deviation bars including incomplete bars for analysis
    ///
    /// # Arguments
    ///
    /// * `agg_trade_records` - Slice of AggTrade records sorted by (timestamp, agg_trade_id)
    ///
    /// # Returns
    ///
    /// Vector of open deviation bars including incomplete bars at end of data
    ///
    /// # Warning
    ///
    /// This method is for analysis purposes only. Incomplete bars violate the
    /// fundamental open deviation bar algorithm and should not be used for production trading.
    pub fn process_agg_trade_records_with_incomplete(
        &mut self,
        agg_trade_records: &[AggTrade],
    ) -> Result<Vec<OpenDeviationBar>, ProcessingError> {
        self.process_agg_trade_records_with_options(agg_trade_records, true)
    }

    /// Process Binance aggregated trade records into open deviation bars
    ///
    /// This is the primary method for converting AggTrade records (which aggregate
    /// multiple individual trades) into open deviation bars based on price movement thresholds.
    ///
    /// # Parameters
    ///
    /// * `agg_trade_records` - Slice of AggTrade records sorted by (timestamp, agg_trade_id)
    ///   Each record represents multiple individual trades aggregated at same price
    ///
    /// # Returns
    ///
    /// Vector of completed open deviation bars (ONLY bars that breached thresholds).
    /// Each bar tracks both individual trade count and AggTrade record count.
    pub fn process_agg_trade_records(
        &mut self,
        agg_trade_records: &[AggTrade],
    ) -> Result<Vec<OpenDeviationBar>, ProcessingError> {
        self.process_agg_trade_records_with_options(agg_trade_records, false)
    }

    /// Process AggTrade records with options for including incomplete bars
    ///
    /// Batch processing mode: Clears any existing state before processing.
    /// Use process_single_trade() for stateful streaming instead.
    ///
    /// # Parameters
    ///
    /// * `agg_trade_records` - Slice of AggTrade records sorted by (timestamp, agg_trade_id)
    /// * `include_incomplete` - Whether to include incomplete bars at end of processing
    ///
    /// # Returns
    ///
    /// Vector of open deviation bars (completed + incomplete if requested)
    pub fn process_agg_trade_records_with_options(
        &mut self,
        agg_trade_records: &[AggTrade],
        include_incomplete: bool,
    ) -> Result<Vec<OpenDeviationBar>, ProcessingError> {
        if agg_trade_records.is_empty() {
            return Ok(Vec::new());
        }

        // Validate records are sorted
        self.validate_trade_ordering(agg_trade_records)?;

        // Use existing bar state if resuming from checkpoint, otherwise start fresh
        // This is CRITICAL for cross-file continuation (Issues #2, #3)
        let mut current_bar: Option<OpenDeviationBarState> = if self.resumed_from_checkpoint {
            // Continue from checkpoint's incomplete bar
            self.resumed_from_checkpoint = false; // Consume the flag
            let restored_bar = self.current_bar_state.take();

            // Issue #112: Gap-aware checkpoint recovery
            // If the forming bar's close_time is too far from the first incoming trade,
            // discard it as an orphan to prevent oversized bars from data gaps.
            if let Some(ref bar_state) = restored_bar {
                let first_trade_ts = agg_trade_records[0].timestamp;
                let gap = first_trade_ts - bar_state.bar.close_time;
                if gap > self.max_gap_us {
                    self.anomaly_summary.record_gap();
                    // Discard forming bar — same treatment as ouroboros reset
                    None
                } else {
                    restored_bar
                }
            } else {
                restored_bar
            }
        } else {
            // Start fresh for normal batch processing
            self.current_bar_state = None;
            None
        };

        let mut bars = Vec::with_capacity(agg_trade_records.len() / 50); // Heuristic: 50 trades/bar covers consolidation regimes
        let mut defer_open = false;

        for agg_record in agg_trade_records {
            // Track price and position for checkpoint
            self.price_window.push(agg_record.price);
            self.last_trade_id = Some(agg_record.agg_trade_id);
            self.last_timestamp_us = agg_record.timestamp;

            // Issue #59: Push trade to history buffer for inter-bar feature computation
            if let Some(ref mut history) = self.trade_history {
                history.push(agg_record);
            }

            if defer_open {
                // Previous bar closed, this agg_record opens new bar
                // Issue #68: Notify history that new bar is opening (preserves pre-bar trades)
                if let Some(ref mut history) = self.trade_history {
                    history.on_bar_open(agg_record.timestamp);
                }
                current_bar = Some(if self.include_intra_bar_features {
                    OpenDeviationBarState::new_with_trade_accumulation(
                        agg_record,
                        self.threshold_ratio,
                    )
                } else {
                    OpenDeviationBarState::new(agg_record, self.threshold_ratio)
                });
                defer_open = false;
                continue;
            }

            match current_bar {
                None => {
                    // First bar initialization
                    // Issue #68: Notify history that new bar is opening (preserves pre-bar trades)
                    if let Some(ref mut history) = self.trade_history {
                        history.on_bar_open(agg_record.timestamp);
                    }
                    current_bar = Some(if self.include_intra_bar_features {
                        OpenDeviationBarState::new_with_trade_accumulation(
                            agg_record,
                            self.threshold_ratio,
                        )
                    } else {
                        OpenDeviationBarState::new(agg_record, self.threshold_ratio)
                    });
                }
                Some(ref mut bar_state) => {
                    // Issue #59 & #96 Task #44: Accumulate trade for intra-bar features (before breach check)
                    // Only accumulates if features enabled, avoiding unnecessary clones
                    bar_state.accumulate_trade(agg_record, self.include_intra_bar_features);

                    // Check if this AggTrade record breaches the threshold
                    let price_breaches = bar_state.bar.is_breach(
                        agg_record.price,
                        bar_state.upper_threshold,
                        bar_state.lower_threshold,
                    );

                    // Timestamp gate (Issue #36): prevent bars from closing on same timestamp
                    // This eliminates "instant bars" during flash crashes where multiple trades
                    // occur at the same millisecond.
                    let timestamp_allows_close = !self.prevent_same_timestamp_close
                        || agg_record.timestamp != bar_state.bar.open_time;

                    if price_breaches && timestamp_allows_close {
                        // Breach detected AND timestamp changed - update bar with breaching record
                        bar_state.bar.update_with_trade(agg_record);

                        // Validation: Ensure high/low include open/close extremes
                        debug_assert!(
                            bar_state.bar.high >= bar_state.bar.open.max(bar_state.bar.close)
                        );
                        debug_assert!(
                            bar_state.bar.low <= bar_state.bar.open.min(bar_state.bar.close)
                        );

                        // Compute microstructure features at bar finalization (Issue #34)
                        bar_state.bar.compute_microstructure_features();

                        // Issue #59: Compute inter-bar features from lookback window
                        if let Some(ref mut history) = self.trade_history {
                            let inter_bar_features =
                                history.compute_features(bar_state.bar.open_time);
                            bar_state.bar.set_inter_bar_features(&inter_bar_features);
                            // Issue #68: Notify history that bar is closing (resumes normal pruning)
                            history.on_bar_close();
                        }

                        // Issue #59: Compute intra-bar features from accumulated trades
                        if self.include_intra_bar_features {
                            // Issue #96 Task #173: Use reusable scratch buffers from bar_state
                            let intra_bar_features = crate::intrabar::compute_intra_bar_features_with_config(
                                &bar_state.accumulated_trades,
                                &mut bar_state.scratch_prices,
                                &mut bar_state.scratch_volumes,
                                &self.intra_bar_config, // Issue #128
                            );
                            bar_state.bar.set_intra_bar_features(&intra_bar_features);
                        }

                        // Move bar out instead of cloning — bar_state borrow ends
                        // after last use above (NLL), so take() is safe here.
                        bars.push(current_bar.take().unwrap().bar);
                        defer_open = true; // Next record will open new bar
                    } else {
                        // Either no breach OR same timestamp (gate active) - normal update
                        bar_state.bar.update_with_trade(agg_record);
                    }
                }
            }
        }

        // Save current bar state for checkpoint and optionally append incomplete bar.
        // When include_incomplete=true, clone for checkpoint then consume for output.
        // When include_incomplete=false, move directly (no clone needed).
        if include_incomplete {
            // Issue #96 Task #95: Optimize checkpoint cloning with take() to avoid Vec allocation
            // (accumulated_trades not needed after intra-bar features computed).
            // Using std::mem::take() instead of clone+clear reduces allocation overhead.
            if let Some(ref state) = current_bar {
                let mut checkpoint_state = state.clone();
                // Use take() to avoid cloning the Vec - directly moves empty Vec into place
                let _ = std::mem::take(&mut checkpoint_state.accumulated_trades);
                self.current_bar_state = Some(checkpoint_state);
            }

            // Add final partial bar only if explicitly requested
            // This preserves algorithm integrity: bars should only close on threshold breach
            if let Some(mut bar_state) = current_bar {
                // Compute microstructure features for incomplete bar (Issue #34)
                bar_state.bar.compute_microstructure_features();

                // Issue #59: Compute inter-bar features from lookback window
                if let Some(ref history) = self.trade_history {
                    let inter_bar_features = history.compute_features(bar_state.bar.open_time);
                    bar_state.bar.set_inter_bar_features(&inter_bar_features);
                }

                // Issue #59: Compute intra-bar features from accumulated trades
                if self.include_intra_bar_features {
                    // Issue #96 Task #173: Use reusable scratch buffers from bar_state
                    let intra_bar_features = crate::intrabar::compute_intra_bar_features_with_config(
                        &bar_state.accumulated_trades,
                        &mut bar_state.scratch_prices,
                        &mut bar_state.scratch_volumes,
                        &self.intra_bar_config, // Issue #128
                    );
                    bar_state.bar.set_intra_bar_features(&intra_bar_features);
                }

                bars.push(bar_state.bar);
            }
        } else {
            // No incomplete bar appended — move ownership directly, no clone needed
            self.current_bar_state = current_bar;
        }

        Ok(bars)
    }

    // === CHECKPOINT METHODS ===

    /// Create checkpoint for cross-file continuation
    ///
    /// Captures current processing state for seamless continuation:
    /// - Incomplete bar (if any) with FIXED thresholds
    /// - Position tracking (timestamp, trade_id if available)
    /// - Price hash for verification
    ///
    /// # Arguments
    ///
    /// * `symbol` - Symbol being processed (e.g., "BTCUSDT", "EURUSD")
    ///
    /// # Example
    ///
    /// ```ignore
    /// let bars = processor.process_agg_trade_records(&trades)?;
    /// let checkpoint = processor.create_checkpoint("BTCUSDT");
    /// let json = serde_json::to_string(&checkpoint)?;
    /// std::fs::write("checkpoint.json", json)?;
    /// ```
    pub fn create_checkpoint(&self, symbol: &str) -> Checkpoint {
        let (incomplete_bar, thresholds) = match &self.current_bar_state {
            Some(state) => (
                Some(state.bar.clone()),
                Some((state.upper_threshold, state.lower_threshold)),
            ),
            None => (None, None),
        };

        let mut checkpoint = Checkpoint::new(
            symbol.to_string(),
            self.threshold_decimal_bps,
            incomplete_bar,
            thresholds,
            self.last_timestamp_us,
            self.last_trade_id,
            self.price_window.compute_hash(),
            self.prevent_same_timestamp_close,
        );
        // Issue #46: Persist defer_open state for cross-session continuity
        checkpoint.defer_open = self.defer_open;
        checkpoint
    }

    /// Resume processing from checkpoint
    ///
    /// Restores incomplete bar state with IMMUTABLE thresholds.
    /// Next trade continues building the bar until threshold breach.
    ///
    /// # Errors
    ///
    /// - `CheckpointError::MissingThresholds` - Checkpoint has bar but no thresholds
    ///
    /// # Example
    ///
    /// ```ignore
    /// let json = std::fs::read_to_string("checkpoint.json")?;
    /// let checkpoint: Checkpoint = serde_json::from_str(&json)?;
    /// let mut processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint)?;
    /// let bars = processor.process_agg_trade_records(&next_file_trades)?;
    /// ```
    pub fn from_checkpoint(checkpoint: Checkpoint) -> Result<Self, CheckpointError> {
        // Issue #85 Phase 2: Apply checkpoint schema migration if needed
        let checkpoint = Self::migrate_checkpoint(checkpoint);

        // Issue #62: Validate threshold range before restoring from checkpoint
        // Valid range: 1-100,000 dbps (0.0001% to 10%)
        const THRESHOLD_MIN: u32 = 1;
        const THRESHOLD_MAX: u32 = 100_000;
        if checkpoint.threshold_decimal_bps < THRESHOLD_MIN
            || checkpoint.threshold_decimal_bps > THRESHOLD_MAX
        {
            return Err(CheckpointError::InvalidThreshold {
                threshold: checkpoint.threshold_decimal_bps,
                min_threshold: THRESHOLD_MIN,
                max_threshold: THRESHOLD_MAX,
            });
        }

        // Validate checkpoint consistency
        if checkpoint.incomplete_bar.is_some() && checkpoint.thresholds.is_none() {
            return Err(CheckpointError::MissingThresholds);
        }

        // Restore bar state if there's an incomplete bar
        // Note: accumulated_trades is reset to empty - intra-bar features won't be
        // accurate for bars resumed from checkpoint (partial trade history lost)
        let current_bar_state = match (checkpoint.incomplete_bar, checkpoint.thresholds) {
            (Some(bar), Some((upper, lower))) => Some(OpenDeviationBarState {
                bar,
                upper_threshold: upper,
                lower_threshold: lower,
                accumulated_trades: SmallVec::new(), // Lost on checkpoint - features may be partial
                scratch_prices: SmallVec::new(),
                scratch_volumes: SmallVec::new(),
            }),
            _ => None,
        };

        // Issue #96 Task #98: Pre-compute threshold ratio (same as with_options)
        let threshold_ratio = ((checkpoint.threshold_decimal_bps as i64) * crate::fixed_point::SCALE)
            / (crate::fixed_point::BASIS_POINTS_SCALE as i64);

        Ok(Self {
            threshold_decimal_bps: checkpoint.threshold_decimal_bps,
            threshold_ratio,
            current_bar_state,
            price_window: PriceWindow::new(), // Reset - will be rebuilt from new trades
            last_trade_id: checkpoint.last_trade_id,
            last_timestamp_us: checkpoint.last_timestamp_us,
            anomaly_summary: checkpoint.anomaly_summary,
            resumed_from_checkpoint: true, // Signal to continue from existing bar state
            prevent_same_timestamp_close: checkpoint.prevent_same_timestamp_close,
            defer_open: checkpoint.defer_open, // Issue #46: Restore deferred open state
            trade_history: None,               // Issue #59: Must be re-enabled after restore
            inter_bar_config: None,            // Issue #59: Must be re-enabled after restore
            include_intra_bar_features: false, // Issue #59: Must be re-enabled after restore
            intra_bar_config: crate::intrabar::IntraBarConfig::default(), // Issue #128
            max_gap_us: 3_600_000_000,         // Issue #112: 1 hour default
        })
    }

    /// Migrate checkpoint between schema versions
    /// Issue #85 Phase 2: Handle v1 → v2 migration
    /// Safe: JSON deserialization is field-name-based, so old v1 checkpoints load correctly
    fn migrate_checkpoint(mut checkpoint: Checkpoint) -> Checkpoint {
        match checkpoint.version {
            1 => {
                // v1 → v2: OpenDeviationBar struct field reordering (no behavioral changes)
                // JSON serialization is position-independent, so no transformation needed
                checkpoint.version = 2;
                checkpoint
            }
            2 => {
                // Already current version
                checkpoint
            }
            _ => {
                // Unknown version - log warning and continue with best effort
                eprintln!(
                    "Warning: Checkpoint has unknown version {}, treating as v2",
                    checkpoint.version
                );
                checkpoint.version = 2;
                checkpoint
            }
        }
    }

    /// Verify we're at the right position in the data stream
    ///
    /// Call with first trade of new file to verify continuity.
    /// Returns verification result indicating if there's a gap or exact match.
    ///
    /// # Arguments
    ///
    /// * `first_trade` - First trade of the new file/chunk
    ///
    /// # Example
    ///
    /// ```ignore
    /// let processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint)?;
    /// let verification = processor.verify_position(&next_file_trades[0]);
    /// match verification {
    ///     PositionVerification::Exact => println!("Perfect continuation!"),
    ///     PositionVerification::Gap { missing_count, .. } => {
    ///         println!("Warning: {} trades missing", missing_count);
    ///     }
    ///     PositionVerification::TimestampOnly { gap_ms } => {
    ///         println!("Exness data: {}ms gap", gap_ms);
    ///     }
    /// }
    /// ```
    pub fn verify_position(&self, first_trade: &AggTrade) -> PositionVerification {
        match self.last_trade_id {
            Some(last_id) => {
                // Binance: has trade IDs - check for gaps
                let expected_id = last_id + 1;
                if first_trade.agg_trade_id == expected_id {
                    PositionVerification::Exact
                } else {
                    let missing_count = first_trade.agg_trade_id - expected_id;
                    PositionVerification::Gap {
                        expected_id,
                        actual_id: first_trade.agg_trade_id,
                        missing_count,
                    }
                }
            }
            None => {
                // Exness: no trade IDs - use timestamp only
                let gap_us = first_trade.timestamp - self.last_timestamp_us;
                let gap_ms = gap_us / 1000;
                PositionVerification::TimestampOnly { gap_ms }
            }
        }
    }

    /// Get the current anomaly summary
    pub fn anomaly_summary(&self) -> &AnomalySummary {
        &self.anomaly_summary
    }

    /// Get the threshold in decimal basis points
    pub fn threshold_decimal_bps(&self) -> u32 {
        self.threshold_decimal_bps
    }

    /// Validate that trades are properly sorted for deterministic processing
    ///
    /// Issue #96 Task #62: Early-exit optimization for sorted data
    /// For typical workloads (95%+ sorted), quick first/last check identifies
    /// unsorted batches immediately without full O(n) validation.
    fn validate_trade_ordering(&self, trades: &[AggTrade]) -> Result<(), ProcessingError> {
        if trades.is_empty() {
            return Ok(());
        }

        // Issue #96 Task #62: Fast-path check for obviously unsorted data
        // If first and last trades are not ordered, data is definitely unsorted
        // This early-exit catches common failures without full validation
        let first = &trades[0];
        let last = &trades[trades.len() - 1];

        if last.timestamp < first.timestamp
            || (last.timestamp == first.timestamp && last.agg_trade_id <= first.agg_trade_id)
        {
            // Definitely unsorted - find exact error location (cold path)
            return find_unsorted_trade(trades);
        }

        // Full validation for typical sorted case
        for i in 1..trades.len() {
            let prev = &trades[i - 1];
            let curr = &trades[i];

            // Check ordering: (timestamp, agg_trade_id) ascending
            if curr.timestamp < prev.timestamp
                || (curr.timestamp == prev.timestamp && curr.agg_trade_id <= prev.agg_trade_id)
            {
                return unsorted_trade_error(i, prev, curr);
            }
        }

        Ok(())
    }

    /// Reset processor state at an ouroboros boundary (year/month/week).
    ///
    /// Clears the incomplete bar and position tracking while preserving
    /// the threshold configuration. Use this when starting fresh at a
    /// known boundary for reproducibility.
    ///
    /// # Returns
    ///
    /// The orphaned incomplete bar (if any) so caller can decide
    /// whether to include it in results with `is_orphan=True` flag.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // At year boundary (Jan 1 00:00:00 UTC)
    /// let orphaned = processor.reset_at_ouroboros();
    /// if let Some(bar) = orphaned {
    ///     // Handle incomplete bar from previous year
    /// }
    /// // Continue processing new year's data with clean state
    /// ```
    pub fn reset_at_ouroboros(&mut self) -> Option<OpenDeviationBar> {
        let orphaned = self.current_bar_state.take().map(|state| state.bar);
        self.price_window = PriceWindow::new();
        self.last_trade_id = None;
        self.last_timestamp_us = 0;
        self.resumed_from_checkpoint = false;
        self.defer_open = false;
        // Issue #81: Clear bar boundary tracking at ouroboros reset.
        // Trades are preserved — still valid lookback for first bar of new segment.
        if let Some(ref mut history) = self.trade_history {
            history.reset_bar_boundaries();
        }
        orphaned
    }
}

/// Internal state for a open deviation bar being built
#[derive(Clone)]
struct OpenDeviationBarState {
    /// The open deviation bar being constructed
    pub bar: OpenDeviationBar,

    /// Upper breach threshold (FIXED from bar open)
    pub upper_threshold: FixedPoint,

    /// Lower breach threshold (FIXED from bar open)
    pub lower_threshold: FixedPoint,

    /// Accumulated trades for intra-bar feature computation (Issue #59)
    ///
    /// When intra-bar features are enabled, trades are accumulated here
    /// during bar construction and used to compute features at bar close.
    /// Cleared when bar closes to free memory.
    /// Issue #136: Optimized from 512 to 64 slots (saves 87.5% stack memory).
    /// Profile data: max trades/bar = 26 (P99 = 14), so 64 slots provides
    /// 2.5x safety margin while reducing from 32KB to 4KB per bar.
    pub accumulated_trades: SmallVec<[AggTrade; 64]>,

    /// Issue #96: Scratch buffer for intra-bar price extraction
    /// SmallVec<[f64; 64]> keeps 95%+ of bars on stack (P99 trades/bar = 14, max = 26)
    /// Eliminates heap allocation for typical bars, spills transparently for large ones
    pub scratch_prices: SmallVec<[f64; 64]>,

    /// Issue #96: Scratch buffer for intra-bar volume extraction
    /// Same sizing rationale as scratch_prices
    pub scratch_volumes: SmallVec<[f64; 64]>,
}

impl OpenDeviationBarState {
    /// Create new open deviation bar state from opening trade
    /// Issue #96 Task #98: Accept pre-computed threshold_ratio for fast threshold calculation
    #[inline]
    fn new(trade: &AggTrade, threshold_ratio: i64) -> Self {
        let bar = OpenDeviationBar::new(trade);

        // Issue #96 Task #98: Use cached ratio instead of repeated division
        // This avoids BASIS_POINTS_SCALE division on every bar creation
        let (upper_threshold, lower_threshold) =
            bar.open.compute_range_thresholds_cached(threshold_ratio);

        Self {
            bar,
            upper_threshold,
            lower_threshold,
            accumulated_trades: SmallVec::new(),
            scratch_prices: SmallVec::new(),
            scratch_volumes: SmallVec::new(),
        }
    }

    /// Create new open deviation bar state with intra-bar feature accumulation
    /// Issue #96 Task #98: Accept pre-computed threshold_ratio for fast threshold calculation
    #[inline]
    fn new_with_trade_accumulation(trade: &AggTrade, threshold_ratio: i64) -> Self {
        let bar = OpenDeviationBar::new(trade);

        // Issue #96 Task #98: Use cached ratio instead of repeated division
        // This avoids BASIS_POINTS_SCALE division on every bar creation
        let (upper_threshold, lower_threshold) =
            bar.open.compute_range_thresholds_cached(threshold_ratio);

        Self {
            bar,
            upper_threshold,
            lower_threshold,
            accumulated_trades: {
                let mut sv = SmallVec::new();
                sv.push(trade.clone());
                sv
            },
            scratch_prices: SmallVec::new(),
            scratch_volumes: SmallVec::new(),
        }
    }

    /// Accumulate a trade for intra-bar feature computation
    ///
    /// Issue #96 Task #44: Only accumulates if intra-bar features are enabled,
    /// avoiding unnecessary clones for the majority of use cases where they're disabled.
    /// Issue #96 Task #79: #[inline] allows compiler to fold invariant branch
    /// (include_intra is constant for processor lifetime)
    #[inline]
    fn accumulate_trade(&mut self, trade: &AggTrade, include_intra: bool) {
        if include_intra {
            self.accumulated_trades.push(trade.clone());
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_utils::{self, scenarios};

    #[test]
    fn test_single_bar_no_breach() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 dbps = 0.25%

        // Create trades that stay within 250 dbps threshold
        let trades = scenarios::no_breach_sequence(250);

        // Test strict algorithm compliance: no bars should be created without breach
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(
            bars.len(),
            0,
            "Strict algorithm should not create bars without breach"
        );

        // Test analysis mode: incomplete bar should be available for analysis
        let bars_with_incomplete = processor
            .process_agg_trade_records_with_incomplete(&trades)
            .unwrap();
        assert_eq!(
            bars_with_incomplete.len(),
            1,
            "Analysis mode should include incomplete bar"
        );

        let bar = &bars_with_incomplete[0];
        assert_eq!(bar.open.to_string(), "50000.00000000");
        assert_eq!(bar.high.to_string(), "50100.00000000");
        assert_eq!(bar.low.to_string(), "49900.00000000");
        assert_eq!(bar.close.to_string(), "49900.00000000");
    }

    #[test]
    fn test_exact_breach_upward() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 dbps = 0.25%

        let trades = scenarios::exact_breach_upward(250);

        // Test strict algorithm: only completed bars (with breach)
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(
            bars.len(),
            1,
            "Strict algorithm should only return completed bars"
        );

        // First bar should close at breach
        let bar1 = &bars[0];
        assert_eq!(bar1.open.to_string(), "50000.00000000");
        // Breach at 250 dbps = 0.25% = 50000 * 1.0025 = 50125
        assert_eq!(bar1.close.to_string(), "50125.00000000"); // Breach tick included
        assert_eq!(bar1.high.to_string(), "50125.00000000");
        assert_eq!(bar1.low.to_string(), "50000.00000000");

        // Test analysis mode: includes incomplete second bar
        let bars_with_incomplete = processor
            .process_agg_trade_records_with_incomplete(&trades)
            .unwrap();
        assert_eq!(
            bars_with_incomplete.len(),
            2,
            "Analysis mode should include incomplete bars"
        );

        // Second bar should start at next tick price (not breach price)
        let bar2 = &bars_with_incomplete[1];
        assert_eq!(bar2.open.to_string(), "50500.00000000"); // Next tick after breach
        assert_eq!(bar2.close.to_string(), "50500.00000000");
    }

    #[test]
    fn test_exact_breach_downward() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 × 0.1bps = 25bps = 0.25%

        let trades = scenarios::exact_breach_downward(250);

        let bars = processor.process_agg_trade_records(&trades).unwrap();

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

        let bar = &bars[0];
        assert_eq!(bar.open.to_string(), "50000.00000000");
        assert_eq!(bar.close.to_string(), "49875.00000000"); // Breach tick included
        assert_eq!(bar.high.to_string(), "50000.00000000");
        assert_eq!(bar.low.to_string(), "49875.00000000");
    }

    #[test]
    fn test_large_gap_single_bar() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 × 0.1bps = 25bps = 0.25%

        let trades = scenarios::large_gap_sequence();

        let bars = processor.process_agg_trade_records(&trades).unwrap();

        // Should create exactly ONE bar, not multiple bars to "fill the gap"
        assert_eq!(bars.len(), 1);

        let bar = &bars[0];
        assert_eq!(bar.open.to_string(), "50000.00000000");
        assert_eq!(bar.close.to_string(), "51000.00000000");
        assert_eq!(bar.high.to_string(), "51000.00000000");
        assert_eq!(bar.low.to_string(), "50000.00000000");
    }

    #[test]
    fn test_unsorted_trades_error() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 × 0.1bps = 25bps

        let trades = scenarios::unsorted_sequence();

        let result = processor.process_agg_trade_records(&trades);
        assert!(result.is_err());

        match result {
            Err(ProcessingError::UnsortedTrades { index, .. }) => {
                assert_eq!(index, 1);
            }
            _ => panic!("Expected UnsortedTrades error"),
        }
    }

    #[test]
    fn test_threshold_calculation() {
        let processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 × 0.1bps = 25bps = 0.25%

        let trade = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let bar_state = OpenDeviationBarState::new(&trade, processor.threshold_ratio);

        // 50000 * 0.0025 = 125 (25bps = 0.25%)
        assert_eq!(bar_state.upper_threshold.to_string(), "50125.00000000");
        assert_eq!(bar_state.lower_threshold.to_string(), "49875.00000000");
    }

    #[test]
    fn test_empty_trades() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap(); // 250 × 0.1bps = 25bps
        let trades = scenarios::empty_sequence();
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 0);
    }

    #[test]
    fn test_debug_streaming_data() {
        let mut processor = OpenDeviationBarProcessor::new(100).unwrap(); // 100 × 0.1bps = 10bps = 0.1%

        // Create trades similar to our test data
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50014.00859087", "0.12019569", 1756710002083),
            test_utils::create_test_agg_trade(2, "50163.87750994", "1.01283708", 1756710005113), // ~0.3% increase
            test_utils::create_test_agg_trade(3, "50032.44128269", "0.69397094", 1756710008770),
        ];

        println!("Test data prices: 50014 -> 50163 -> 50032");
        println!("Expected price movements: +0.3% then -0.26%");

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        println!("Generated {} open deviation bars", bars.len());

        for (i, bar) in bars.iter().enumerate() {
            println!(
                "  Bar {}: O={} H={} L={} C={}",
                i + 1,
                bar.open,
                bar.high,
                bar.low,
                bar.close
            );
        }

        // With a 0.1% threshold and 0.3% price movement, we should get at least 1 bar
        assert!(
            !bars.is_empty(),
            "Expected at least 1 open deviation bar with 0.3% price movement and 0.1% threshold"
        );
    }

    #[test]
    fn test_threshold_validation() {
        // Valid threshold
        assert!(OpenDeviationBarProcessor::new(250).is_ok());

        // Invalid: too low (0 × 0.1bps = 0%)
        assert!(matches!(
            OpenDeviationBarProcessor::new(0),
            Err(ProcessingError::InvalidThreshold {
                threshold_decimal_bps: 0
            })
        ));

        // Invalid: too high (150,000 × 0.1bps = 15,000bps = 150%)
        assert!(matches!(
            OpenDeviationBarProcessor::new(150_000),
            Err(ProcessingError::InvalidThreshold {
                threshold_decimal_bps: 150_000
            })
        ));

        // Valid boundary: minimum (1 × 0.1bps = 0.1bps = 0.001%)
        assert!(OpenDeviationBarProcessor::new(1).is_ok());

        // Valid boundary: maximum (100,000 × 0.1bps = 10,000bps = 100%)
        assert!(OpenDeviationBarProcessor::new(100_000).is_ok());
    }

    #[test]
    fn test_export_processor_with_manual_trades() {
        println!("Testing ExportOpenDeviationBarProcessor with same trade data...");

        let mut export_processor = ExportOpenDeviationBarProcessor::new(100).unwrap(); // 100 × 0.1bps = 10bps = 0.1%

        // Use same trades as the working basic test
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50014.00859087", "0.12019569", 1756710002083),
            test_utils::create_test_agg_trade(2, "50163.87750994", "1.01283708", 1756710005113), // ~0.3% increase
            test_utils::create_test_agg_trade(3, "50032.44128269", "0.69397094", 1756710008770),
        ];

        println!(
            "Processing {} trades with ExportOpenDeviationBarProcessor...",
            trades.len()
        );

        export_processor.process_trades_continuously(&trades);
        let bars = export_processor.get_all_completed_bars();

        println!(
            "ExportOpenDeviationBarProcessor generated {} open deviation bars",
            bars.len()
        );
        for (i, bar) in bars.iter().enumerate() {
            println!(
                "  Bar {}: O={} H={} L={} C={}",
                i + 1,
                bar.open,
                bar.high,
                bar.low,
                bar.close
            );
        }

        // Should match the basic processor results (1 bar)
        assert!(
            !bars.is_empty(),
            "ExportOpenDeviationBarProcessor should generate same results as basic processor"
        );
    }

    // === CHECKPOINT TESTS (Issues #2 and #3) ===

    #[test]
    fn test_checkpoint_creation() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Process some trades that don't complete a bar
        let trades = scenarios::no_breach_sequence(250);
        let _bars = processor.process_agg_trade_records(&trades).unwrap();

        // Create checkpoint
        let checkpoint = processor.create_checkpoint("BTCUSDT");

        assert_eq!(checkpoint.symbol, "BTCUSDT");
        assert_eq!(checkpoint.threshold_decimal_bps, 250);
        assert!(checkpoint.has_incomplete_bar()); // Should have incomplete bar
        assert!(checkpoint.thresholds.is_some()); // Thresholds should be saved
        assert!(checkpoint.last_trade_id.is_some()); // Should track last trade
    }

    #[test]
    fn test_checkpoint_serialization_roundtrip() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Process trades
        let trades = scenarios::no_breach_sequence(250);
        let _bars = processor.process_agg_trade_records(&trades).unwrap();

        // Create checkpoint
        let checkpoint = processor.create_checkpoint("BTCUSDT");

        // Serialize to JSON
        let json = serde_json::to_string(&checkpoint).expect("Serialization should succeed");

        // Deserialize back
        let restored: Checkpoint =
            serde_json::from_str(&json).expect("Deserialization should succeed");

        assert_eq!(restored.symbol, checkpoint.symbol);
        assert_eq!(
            restored.threshold_decimal_bps,
            checkpoint.threshold_decimal_bps
        );
        assert_eq!(
            restored.incomplete_bar.is_some(),
            checkpoint.incomplete_bar.is_some()
        );
    }

    #[test]
    fn test_cross_file_bar_continuation() {
        // This is the PRIMARY test for Issues #2 and #3
        // Verifies that incomplete bars continue correctly across file boundaries

        // Create trades that span multiple bars
        let mut all_trades = Vec::new();

        // Generate enough trades to produce multiple bars
        // Using 100bps threshold (1%) for clearer price movements
        let base_timestamp = 1640995200000000i64; // Microseconds

        // Create a sequence where we'll have ~3-4 completed bars with remainder
        for i in 0..20 {
            let price = 50000.0 + (i as f64 * 100.0) * if i % 4 < 2 { 1.0 } else { -1.0 };
            let trade = test_utils::create_test_agg_trade(
                i + 1,
                &format!("{:.8}", price),
                "1.0",
                base_timestamp + (i * 1000000),
            );
            all_trades.push(trade);
        }

        // === FULL PROCESSING (baseline) ===
        let mut processor_full = OpenDeviationBarProcessor::new(100).unwrap(); // 100 × 0.1bps = 10bps = 0.1%
        let bars_full = processor_full
            .process_agg_trade_records(&all_trades)
            .unwrap();

        // === SPLIT PROCESSING WITH CHECKPOINT ===
        let split_point = 10; // Split in the middle

        // Part 1: Process first half
        let mut processor_1 = OpenDeviationBarProcessor::new(100).unwrap();
        let part1_trades = &all_trades[0..split_point];
        let bars_1 = processor_1.process_agg_trade_records(part1_trades).unwrap();

        // Create checkpoint
        let checkpoint = processor_1.create_checkpoint("TEST");

        // Part 2: Resume from checkpoint and process second half
        let mut processor_2 = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        let part2_trades = &all_trades[split_point..];
        let bars_2 = processor_2.process_agg_trade_records(part2_trades).unwrap();

        // === VERIFY CONTINUATION ===
        // Total completed bars should match full processing
        let split_total = bars_1.len() + bars_2.len();

        println!("Full processing: {} bars", bars_full.len());
        println!(
            "Split processing: {} + {} = {} bars",
            bars_1.len(),
            bars_2.len(),
            split_total
        );

        assert_eq!(
            split_total,
            bars_full.len(),
            "Split processing should produce same bar count as full processing"
        );

        // Verify the bars themselves match
        let all_split_bars: Vec<_> = bars_1.iter().chain(bars_2.iter()).collect();
        for (i, (full, split)) in bars_full.iter().zip(all_split_bars.iter()).enumerate() {
            assert_eq!(full.open.0, split.open.0, "Bar {} open price mismatch", i);
            assert_eq!(
                full.close.0, split.close.0,
                "Bar {} close price mismatch",
                i
            );
        }
    }

    #[test]
    fn test_verify_position_exact() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Process some trades
        let trade1 = test_utils::create_test_agg_trade(100, "50000.0", "1.0", 1640995200000000);
        let trade2 = test_utils::create_test_agg_trade(101, "50010.0", "1.0", 1640995201000000);

        let _ = processor.process_single_trade(&trade1);
        let _ = processor.process_single_trade(&trade2);

        // Create next trade in sequence
        let next_trade = test_utils::create_test_agg_trade(102, "50020.0", "1.0", 1640995202000000);

        // Verify position
        let verification = processor.verify_position(&next_trade);

        assert_eq!(verification, PositionVerification::Exact);
    }

    #[test]
    fn test_verify_position_gap() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Process some trades
        let trade1 = test_utils::create_test_agg_trade(100, "50000.0", "1.0", 1640995200000000);
        let trade2 = test_utils::create_test_agg_trade(101, "50010.0", "1.0", 1640995201000000);

        let _ = processor.process_single_trade(&trade1);
        let _ = processor.process_single_trade(&trade2);

        // Create next trade with gap (skip IDs 102-104)
        let next_trade = test_utils::create_test_agg_trade(105, "50020.0", "1.0", 1640995202000000);

        // Verify position
        let verification = processor.verify_position(&next_trade);

        match verification {
            PositionVerification::Gap {
                expected_id,
                actual_id,
                missing_count,
            } => {
                assert_eq!(expected_id, 102);
                assert_eq!(actual_id, 105);
                assert_eq!(missing_count, 3);
            }
            _ => panic!("Expected Gap verification, got {:?}", verification),
        }
    }

    // Issue #96 Task #97: Test TimestampOnly branch (Exness path, no trade IDs)
    #[test]
    fn test_verify_position_timestamp_only() {
        // Fresh processor has last_trade_id = None (simulates Exness path)
        let processor = OpenDeviationBarProcessor::new(250).unwrap();

        let trade = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 5000000);
        let verification = processor.verify_position(&trade);

        // last_trade_id is None → TimestampOnly branch
        // gap_ms = (5000000 - 0) / 1000 = 5000
        match verification {
            PositionVerification::TimestampOnly { gap_ms } => {
                assert_eq!(gap_ms, 5000, "gap_ms should be (timestamp - 0) / 1000");
            }
            _ => panic!("Expected TimestampOnly verification, got {:?}", verification),
        }
    }

    #[test]
    fn test_checkpoint_clean_completion() {
        // Test when last trade completes a bar with no remainder
        // In open deviation bar algorithm: breach trade closes bar, NEXT trade opens new bar
        // If there's no next trade, there's no incomplete bar
        let mut processor = OpenDeviationBarProcessor::new(100).unwrap(); // 10bps

        // Create trades that complete exactly one bar
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200000000),
            test_utils::create_test_agg_trade(2, "50100.0", "1.0", 1640995201000000), // ~0.2% move, breaches 0.1%
        ];

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 1, "Should have exactly one completed bar");

        // Create checkpoint - should NOT have incomplete bar
        // (breach trade closes bar, no next trade to open new bar)
        let checkpoint = processor.create_checkpoint("TEST");

        // With defer_open logic, the next bar isn't started until the next trade
        assert!(
            !checkpoint.has_incomplete_bar(),
            "No incomplete bar when last trade was a breach with no following trade"
        );
    }

    #[test]
    fn test_checkpoint_with_remainder() {
        // Test when we have trades remaining after a completed bar
        let mut processor = OpenDeviationBarProcessor::new(100).unwrap(); // 10bps

        // Create trades: bar completes at trade 2, trade 3 starts new bar
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200000000),
            test_utils::create_test_agg_trade(2, "50100.0", "1.0", 1640995201000000), // Breach
            test_utils::create_test_agg_trade(3, "50110.0", "1.0", 1640995202000000), // Opens new bar
        ];

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 1, "Should have exactly one completed bar");

        // Create checkpoint - should have incomplete bar from trade 3
        let checkpoint = processor.create_checkpoint("TEST");

        assert!(
            checkpoint.has_incomplete_bar(),
            "Should have incomplete bar from trade 3"
        );

        // Verify the incomplete bar has correct data
        let incomplete = checkpoint.incomplete_bar.unwrap();
        assert_eq!(
            incomplete.open.to_string(),
            "50110.00000000",
            "Incomplete bar should open at trade 3 price"
        );
    }

    /// Issue #46: Verify streaming and batch paths produce identical bars
    ///
    /// The batch path (`process_agg_trade_records`) and streaming path
    /// (`process_single_trade`) must produce identical OHLCV output for
    /// the same input trades. This test catches regressions where the
    /// breaching trade is double-counted or bar boundaries differ.
    #[test]
    fn test_streaming_batch_parity() {
        let threshold = 250; // 250 dbps = 0.25%

        // Build a sequence with multiple breaches
        let trades = test_utils::AggTradeBuilder::new()
            .add_trade(1, 1.0, 0) // Open first bar at 50000
            .add_trade(2, 1.001, 1000) // +0.1% - accumulate
            .add_trade(3, 1.003, 2000) // +0.3% - breach (>0.25%)
            .add_trade(4, 1.004, 3000) // Opens second bar
            .add_trade(5, 1.005, 4000) // Accumulate
            .add_trade(6, 1.008, 5000) // +0.4% from bar 2 open - breach
            .add_trade(7, 1.009, 6000) // Opens third bar
            .build();

        // === BATCH PATH ===
        let mut batch_processor = OpenDeviationBarProcessor::new(threshold).unwrap();
        let batch_bars = batch_processor.process_agg_trade_records(&trades).unwrap();
        let batch_incomplete = batch_processor.get_incomplete_bar();

        // === STREAMING PATH ===
        let mut stream_processor = OpenDeviationBarProcessor::new(threshold).unwrap();
        let mut stream_bars: Vec<OpenDeviationBar> = Vec::new();
        for trade in &trades {
            if let Some(bar) = stream_processor
                .process_single_trade(trade)
                .unwrap()
            {
                stream_bars.push(bar);
            }
        }
        let stream_incomplete = stream_processor.get_incomplete_bar();

        // === VERIFY PARITY ===
        assert_eq!(
            batch_bars.len(),
            stream_bars.len(),
            "Batch and streaming should produce same number of completed bars"
        );

        for (i, (batch_bar, stream_bar)) in batch_bars.iter().zip(stream_bars.iter()).enumerate() {
            assert_eq!(
                batch_bar.open, stream_bar.open,
                "Bar {i}: open price mismatch"
            );
            assert_eq!(
                batch_bar.close, stream_bar.close,
                "Bar {i}: close price mismatch"
            );
            assert_eq!(
                batch_bar.high, stream_bar.high,
                "Bar {i}: high price mismatch"
            );
            assert_eq!(batch_bar.low, stream_bar.low, "Bar {i}: low price mismatch");
            assert_eq!(
                batch_bar.volume, stream_bar.volume,
                "Bar {i}: volume mismatch (double-counting?)"
            );
            assert_eq!(
                batch_bar.open_time, stream_bar.open_time,
                "Bar {i}: open_time mismatch"
            );
            assert_eq!(
                batch_bar.close_time, stream_bar.close_time,
                "Bar {i}: close_time mismatch"
            );
            assert_eq!(
                batch_bar.individual_trade_count, stream_bar.individual_trade_count,
                "Bar {i}: trade count mismatch"
            );
        }

        // Verify incomplete bars match
        match (batch_incomplete, stream_incomplete) {
            (Some(b), Some(s)) => {
                assert_eq!(b.open, s.open, "Incomplete bar: open mismatch");
                assert_eq!(b.close, s.close, "Incomplete bar: close mismatch");
                assert_eq!(b.volume, s.volume, "Incomplete bar: volume mismatch");
            }
            (None, None) => {} // Both finished cleanly
            _ => panic!("Incomplete bar presence mismatch between batch and streaming"),
        }
    }

    /// Issue #96: Proptest-enhanced batch vs streaming parity
    ///
    /// Generates random trade sequences (200-500 trades) with realistic price
    /// movements and verifies bit-exact parity between batch and streaming paths
    /// on ALL bar fields including microstructure features.
    mod proptest_batch_streaming_parity {
        use super::*;
        use proptest::prelude::*;

        /// Generate a realistic trade sequence with random price movements
        fn trade_sequence(
            n: usize,
            base_price: f64,
            volatility: f64,
        ) -> Vec<AggTrade> {
            let mut trades = Vec::with_capacity(n);
            let mut price = base_price;
            let base_ts = 1640995200000i64; // 2022-01-01

            for i in 0..n {
                // Deterministic price walk using sin/cos (proptest handles seed)
                let step = ((i as f64 * 0.3).sin() * volatility)
                    + ((i as f64 * 0.07).cos() * volatility * 0.5);
                price += step;
                // Clamp to avoid negative prices
                if price < 100.0 {
                    price = 100.0 + (i as f64 * 0.01).sin().abs() * 50.0;
                }

                let trade = test_utils::create_test_agg_trade_with_range(
                    i as i64 + 1,
                    &format!("{:.8}", price),
                    "1.50000000",
                    base_ts + (i as i64 * 500), // 500ms apart
                    (i as i64 + 1) * 10,
                    (i as i64 + 1) * 10,
                    i % 3 != 0, // Mix of buy/sell sides
                );
                trades.push(trade);
            }
            trades
        }

        /// Assert two bars are bit-exact on all OHLCV and microstructure fields
        fn assert_bar_parity(i: usize, batch: &OpenDeviationBar, stream: &OpenDeviationBar) {
            // Tier 1: OHLCV core
            assert_eq!(batch.open_time, stream.open_time, "Bar {i}: open_time");
            assert_eq!(batch.close_time, stream.close_time, "Bar {i}: close_time");
            assert_eq!(batch.open, stream.open, "Bar {i}: open");
            assert_eq!(batch.high, stream.high, "Bar {i}: high");
            assert_eq!(batch.low, stream.low, "Bar {i}: low");
            assert_eq!(batch.close, stream.close, "Bar {i}: close");

            // Tier 2: Volume accumulators
            assert_eq!(batch.volume, stream.volume, "Bar {i}: volume");
            assert_eq!(batch.turnover, stream.turnover, "Bar {i}: turnover");
            assert_eq!(batch.buy_volume, stream.buy_volume, "Bar {i}: buy_volume");
            assert_eq!(batch.sell_volume, stream.sell_volume, "Bar {i}: sell_volume");
            assert_eq!(batch.buy_turnover, stream.buy_turnover, "Bar {i}: buy_turnover");
            assert_eq!(batch.sell_turnover, stream.sell_turnover, "Bar {i}: sell_turnover");

            // Tier 3: Trade tracking
            assert_eq!(batch.individual_trade_count, stream.individual_trade_count, "Bar {i}: trade_count");
            assert_eq!(batch.agg_record_count, stream.agg_record_count, "Bar {i}: agg_record_count");
            assert_eq!(batch.first_trade_id, stream.first_trade_id, "Bar {i}: first_trade_id");
            assert_eq!(batch.last_trade_id, stream.last_trade_id, "Bar {i}: last_trade_id");
            assert_eq!(batch.first_agg_trade_id, stream.first_agg_trade_id, "Bar {i}: first_agg_trade_id");
            assert_eq!(batch.last_agg_trade_id, stream.last_agg_trade_id, "Bar {i}: last_agg_trade_id");
            assert_eq!(batch.buy_trade_count, stream.buy_trade_count, "Bar {i}: buy_trade_count");
            assert_eq!(batch.sell_trade_count, stream.sell_trade_count, "Bar {i}: sell_trade_count");

            // Tier 4: VWAP
            assert_eq!(batch.vwap, stream.vwap, "Bar {i}: vwap");

            // Tier 5: Microstructure f64 features (bit-exact comparison)
            assert_eq!(batch.duration_us, stream.duration_us, "Bar {i}: duration_us");
            assert_eq!(batch.ofi.to_bits(), stream.ofi.to_bits(), "Bar {i}: ofi");
            assert_eq!(batch.vwap_close_deviation.to_bits(), stream.vwap_close_deviation.to_bits(), "Bar {i}: vwap_close_dev");
            assert_eq!(batch.price_impact.to_bits(), stream.price_impact.to_bits(), "Bar {i}: price_impact");
            assert_eq!(batch.kyle_lambda_proxy.to_bits(), stream.kyle_lambda_proxy.to_bits(), "Bar {i}: kyle_lambda");
            assert_eq!(batch.trade_intensity.to_bits(), stream.trade_intensity.to_bits(), "Bar {i}: trade_intensity");
            assert_eq!(batch.volume_per_trade.to_bits(), stream.volume_per_trade.to_bits(), "Bar {i}: vol_per_trade");
            assert_eq!(batch.aggression_ratio.to_bits(), stream.aggression_ratio.to_bits(), "Bar {i}: aggression_ratio");
            assert_eq!(batch.aggregation_density_f64.to_bits(), stream.aggregation_density_f64.to_bits(), "Bar {i}: agg_density");
            assert_eq!(batch.turnover_imbalance.to_bits(), stream.turnover_imbalance.to_bits(), "Bar {i}: turnover_imbalance");
        }

        proptest! {
            /// 200-500 random trades, 250 dbps threshold
            #[test]
            fn batch_streaming_parity_random(
                n in 200usize..500,
                volatility in 10.0f64..200.0,
            ) {
                let trades = trade_sequence(n, 50000.0, volatility);

                // Batch path
                let mut batch_proc = OpenDeviationBarProcessor::new(250).unwrap();
                let batch_bars = batch_proc.process_agg_trade_records(&trades).unwrap();
                let batch_incomplete = batch_proc.get_incomplete_bar();

                // Streaming path
                let mut stream_proc = OpenDeviationBarProcessor::new(250).unwrap();
                let mut stream_bars: Vec<OpenDeviationBar> = Vec::new();
                for trade in &trades {
                    if let Some(bar) = stream_proc.process_single_trade(trade).unwrap() {
                        stream_bars.push(bar);
                    }
                }
                let stream_incomplete = stream_proc.get_incomplete_bar();

                // Bar count parity
                prop_assert_eq!(batch_bars.len(), stream_bars.len(),
                    "Completed bar count mismatch: batch={}, stream={} for n={}, vol={}",
                    batch_bars.len(), stream_bars.len(), n, volatility);

                // Field-level parity for all completed bars
                for (i, (b, s)) in batch_bars.iter().zip(stream_bars.iter()).enumerate() {
                    assert_bar_parity(i, b, s);
                }

                // Incomplete bar parity
                match (&batch_incomplete, &stream_incomplete) {
                    (Some(b), Some(s)) => assert_bar_parity(batch_bars.len(), b, s),
                    (None, None) => {}
                    _ => prop_assert!(false,
                        "Incomplete bar presence mismatch: batch={}, stream={}",
                        batch_incomplete.is_some(), stream_incomplete.is_some()),
                }
            }

            /// Vary threshold: 100-1000 dbps
            #[test]
            fn batch_streaming_parity_thresholds(
                threshold in 100u32..1000,
            ) {
                let trades = trade_sequence(300, 50000.0, 80.0);

                let mut batch_proc = OpenDeviationBarProcessor::new(threshold).unwrap();
                let batch_bars = batch_proc.process_agg_trade_records(&trades).unwrap();

                let mut stream_proc = OpenDeviationBarProcessor::new(threshold).unwrap();
                let mut stream_bars: Vec<OpenDeviationBar> = Vec::new();
                for trade in &trades {
                    if let Some(bar) = stream_proc.process_single_trade(trade).unwrap() {
                        stream_bars.push(bar);
                    }
                }

                prop_assert_eq!(batch_bars.len(), stream_bars.len(),
                    "Bar count mismatch at threshold={}", threshold);

                for (i, (b, s)) in batch_bars.iter().zip(stream_bars.iter()).enumerate() {
                    assert_bar_parity(i, b, s);
                }
            }
        }
    }

    /// Issue #46: After breach, next trade opens new bar (not breaching trade)
    #[test]
    fn test_defer_open_new_bar_opens_with_next_trade() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Trade 1: Opens bar at 50000
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        assert!(processor.process_single_trade(&t1).unwrap().is_none());

        // Trade 2: Breaches threshold (+0.3%)
        let t2 = test_utils::create_test_agg_trade(2, "50150.0", "2.0", 2000);
        let bar = processor.process_single_trade(&t2).unwrap();
        assert!(bar.is_some(), "Should close bar on breach");

        let closed_bar = bar.unwrap();
        assert_eq!(closed_bar.open.to_string(), "50000.00000000");
        assert_eq!(closed_bar.close.to_string(), "50150.00000000");

        // After breach, no incomplete bar should exist
        assert!(
            processor.get_incomplete_bar().is_none(),
            "No incomplete bar after breach - defer_open is true"
        );

        // Trade 3: Should open NEW bar (not the breaching trade)
        let t3 = test_utils::create_test_agg_trade(3, "50100.0", "3.0", 3000);
        assert!(processor.process_single_trade(&t3).unwrap().is_none());

        let incomplete = processor.get_incomplete_bar().unwrap();
        assert_eq!(
            incomplete.open.to_string(),
            "50100.00000000",
            "New bar should open at trade 3's price, not trade 2's"
        );
    }

    // === Memory efficiency tests (R1/R2/R3) ===

    #[test]
    fn test_bar_close_take_single_trade() {
        // R1: Verify bar close via single-trade path produces correct OHLCV after
        // clone→take optimization. Uses single_breach_sequence that triggers breach.
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trades = scenarios::single_breach_sequence(250);

        for trade in &trades[..trades.len() - 1] {
            let result = processor.process_single_trade(trade).unwrap();
            assert!(result.is_none());
        }

        // Last trade triggers breach
        let bar = processor
            .process_single_trade(trades.last().unwrap())
            .unwrap()
            .expect("Should produce completed bar");

        // Verify OHLCV integrity after take() optimization
        assert_eq!(bar.open.to_string(), "50000.00000000");
        assert!(bar.high >= bar.open.max(bar.close));
        assert!(bar.low <= bar.open.min(bar.close));
        assert!(bar.volume > 0);

        // Verify processor state is clean after bar close
        assert!(processor.get_incomplete_bar().is_none());
    }

    #[test]
    fn test_bar_close_take_batch() {
        // R2: Verify batch path produces correct bars after clone→take optimization.
        // large_sequence generates enough trades to trigger multiple breaches.
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trades = scenarios::large_sequence(500);

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert!(
            !bars.is_empty(),
            "Should produce at least one completed bar"
        );

        // Verify every bar has valid OHLCV invariants
        for bar in &bars {
            assert!(bar.high >= bar.open.max(bar.close));
            assert!(bar.low <= bar.open.min(bar.close));
            assert!(bar.volume > 0);
            assert!(bar.close_time >= bar.open_time);
        }
    }

    #[test]
    fn test_checkpoint_conditional_clone() {
        // R3: Verify checkpoint state is preserved correctly with both
        // include_incomplete=true and include_incomplete=false.
        let trades = scenarios::no_breach_sequence(250);

        // Test with include_incomplete=false (move, no clone)
        let mut processor1 = OpenDeviationBarProcessor::new(250).unwrap();
        let bars_without = processor1.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars_without.len(), 0);
        // Checkpoint should be preserved
        assert!(processor1.get_incomplete_bar().is_some());

        // Test with include_incomplete=true (clone + consume)
        let mut processor2 = OpenDeviationBarProcessor::new(250).unwrap();
        let bars_with = processor2
            .process_agg_trade_records_with_incomplete(&trades)
            .unwrap();
        assert_eq!(bars_with.len(), 1);
        // Checkpoint should ALSO be preserved (cloned before consume)
        assert!(processor2.get_incomplete_bar().is_some());

        // Both checkpoints should have identical bar content
        let cp1 = processor1.get_incomplete_bar().unwrap();
        let cp2 = processor2.get_incomplete_bar().unwrap();
        assert_eq!(cp1.open, cp2.open);
        assert_eq!(cp1.close, cp2.close);
        assert_eq!(cp1.high, cp2.high);
        assert_eq!(cp1.low, cp2.low);
    }

    #[test]
    fn test_checkpoint_v1_to_v2_migration() {
        // Issue #85 Phase 2: Verify v1→v2 checkpoint migration
        // Simulate loading an old v1 checkpoint (without version field)
        let v1_json = r#"{
            "symbol": "BTCUSDT",
            "threshold_decimal_bps": 250,
            "incomplete_bar": null,
            "thresholds": null,
            "last_timestamp_us": 1640995200000000,
            "last_trade_id": 5000,
            "price_hash": 0,
            "anomaly_summary": {"gaps_detected": 0, "overlaps_detected": 0, "timestamp_anomalies": 0},
            "prevent_same_timestamp_close": true,
            "defer_open": false
        }"#;

        // Deserialize old v1 checkpoint
        let checkpoint: Checkpoint = serde_json::from_str(v1_json).unwrap();
        assert_eq!(checkpoint.version, 1, "Old checkpoints should default to v1");
        assert_eq!(checkpoint.symbol, "BTCUSDT");
        assert_eq!(checkpoint.threshold_decimal_bps, 250);

        // Restore processor from v1 checkpoint (triggers migration)
        let mut processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();

        // Verify processor is ready to continue
        assert!(!processor.get_incomplete_bar().is_some(), "No incomplete bar before processing");

        // Process some trades to verify migration worked
        let trades = scenarios::single_breach_sequence(250);
        let bars = processor.process_agg_trade_records(&trades).unwrap();

        // Should produce a bar after migration
        assert!(!bars.is_empty(), "Should produce bars after v1→v2 migration");
        // Verify bar has valid OHLCV (symbol is in Checkpoint, not OpenDeviationBar)
        assert!(bars[0].volume > 0, "Bar should have volume after migration");
        assert!(bars[0].close_time >= bars[0].open_time, "Bar times should be valid");

        // Create new checkpoint from processor after migration
        let new_checkpoint = processor.create_checkpoint("BTCUSDT");
        assert_eq!(new_checkpoint.version, 2, "New checkpoints should be v2");
        assert_eq!(new_checkpoint.symbol, "BTCUSDT");

        // Verify new checkpoint can be serialized and deserialized
        let json = serde_json::to_string(&new_checkpoint).unwrap();
        let restored: Checkpoint = serde_json::from_str(&json).unwrap();
        assert_eq!(restored.version, 2);
        assert_eq!(restored.symbol, "BTCUSDT");
    }

    // =========================================================================
    // Issue #96: Checkpoint error path tests
    // =========================================================================

    #[test]
    fn test_from_checkpoint_invalid_threshold_zero() {
        let checkpoint = Checkpoint::new(
            "BTCUSDT".to_string(), 0, None, None, 0, None, 0, true,
        );
        match OpenDeviationBarProcessor::from_checkpoint(checkpoint) {
            Err(CheckpointError::InvalidThreshold { threshold: 0, .. }) => {}
            other => panic!("Expected InvalidThreshold(0), got {:?}", other.err()),
        }
    }

    #[test]
    fn test_from_checkpoint_invalid_threshold_too_high() {
        let checkpoint = Checkpoint::new(
            "BTCUSDT".to_string(), 200_000, None, None, 0, None, 0, true,
        );
        match OpenDeviationBarProcessor::from_checkpoint(checkpoint) {
            Err(CheckpointError::InvalidThreshold { threshold: 200_000, .. }) => {}
            other => panic!("Expected InvalidThreshold(200000), got {:?}", other.err()),
        }
    }

    #[test]
    fn test_from_checkpoint_missing_thresholds() {
        let bar = OpenDeviationBar::new(&test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000));
        let mut checkpoint = Checkpoint::new(
            "BTCUSDT".to_string(), 250, None, None, 0, None, 0, true,
        );
        checkpoint.incomplete_bar = Some(bar);
        checkpoint.thresholds = None;

        match OpenDeviationBarProcessor::from_checkpoint(checkpoint) {
            Err(CheckpointError::MissingThresholds) => {}
            other => panic!("Expected MissingThresholds, got {:?}", other.err()),
        }
    }

    #[test]
    fn test_from_checkpoint_unknown_version_treated_as_v2() {
        let mut checkpoint = Checkpoint::new(
            "BTCUSDT".to_string(), 250, None, None, 0, None, 0, true,
        );
        checkpoint.version = 99;

        let processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        assert_eq!(processor.threshold_decimal_bps(), 250);
    }

    #[test]
    fn test_from_checkpoint_valid_with_incomplete_bar() {
        use crate::fixed_point::FixedPoint;
        let bar = OpenDeviationBar::new(&test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000));
        let upper = FixedPoint::from_str("50125.0").unwrap();
        let lower = FixedPoint::from_str("49875.0").unwrap();

        let checkpoint = Checkpoint::new(
            "BTCUSDT".to_string(), 250, Some(bar), Some((upper, lower)), 0, None, 0, true,
        );

        let processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        assert!(processor.get_incomplete_bar().is_some(), "Should restore incomplete bar");
    }

    // =========================================================================
    // Issue #96: Ouroboros reset tests
    // =========================================================================

    #[test]
    fn test_reset_at_ouroboros_with_orphan() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Feed trades to create incomplete bar
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let t2 = test_utils::create_test_agg_trade(2, "50050.0", "1.0", 2000);
        assert!(processor.process_single_trade(&t1).unwrap().is_none());
        assert!(processor.process_single_trade(&t2).unwrap().is_none());
        assert!(processor.get_incomplete_bar().is_some(), "Should have incomplete bar");

        // Reset at ouroboros boundary - should return orphaned bar
        let orphan = processor.reset_at_ouroboros();
        assert!(orphan.is_some(), "Should return orphaned bar");
        let orphan_bar = orphan.unwrap();
        assert_eq!(orphan_bar.open.to_string(), "50000.00000000");

        // After reset, no incomplete bar
        assert!(processor.get_incomplete_bar().is_none(), "No bar after reset");
    }

    #[test]
    fn test_reset_at_ouroboros_clean_state() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Reset without any trades processed - should return None
        let orphan = processor.reset_at_ouroboros();
        assert!(orphan.is_none(), "No orphan when state is clean");
        assert!(processor.get_incomplete_bar().is_none());
    }

    #[test]
    fn test_reset_at_ouroboros_clears_defer_open() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Create a breach to set defer_open = true
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let t2 = test_utils::create_test_agg_trade(2, "50200.0", "1.0", 2000); // +0.4% breach
        processor.process_single_trade(&t1).unwrap();
        let bar = processor.process_single_trade(&t2).unwrap();
        assert!(bar.is_some(), "Should breach");

        // After breach, defer_open is true - no incomplete bar
        assert!(processor.get_incomplete_bar().is_none());

        // Reset at ouroboros should clear defer_open
        processor.reset_at_ouroboros();

        // New trade should open fresh bar (defer_open was cleared)
        let t3 = test_utils::create_test_agg_trade(3, "50000.0", "1.0", 3000);
        processor.process_single_trade(&t3).unwrap();
        assert!(processor.get_incomplete_bar().is_some(), "Should have new bar after reset");
    }

    // === EDGE CASE TESTS (Issue #96 Task #21) ===

    #[test]
    fn test_single_trade_no_bar() {
        // A single trade cannot breach — no bar should be produced
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trade = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let bars = processor.process_agg_trade_records(&[trade]).unwrap();
        assert_eq!(bars.len(), 0, "Single trade should not produce a completed bar");
        assert!(processor.get_incomplete_bar().is_some(), "Should have incomplete bar");
    }

    #[test]
    fn test_identical_timestamps_no_close() {
        // Issue #36: Bar cannot close when breach tick has same timestamp as open
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let t2 = test_utils::create_test_agg_trade(2, "50200.0", "1.0", 1000); // Same timestamp, breaches
        let bars = processor.process_agg_trade_records(&[t1, t2]).unwrap();
        assert_eq!(bars.len(), 0, "Bar should not close on same timestamp as open (Issue #36)");
    }

    #[test]
    fn test_identical_timestamps_then_different_closes() {
        // Same-timestamp trades followed by different-timestamp breach should close
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let t2 = test_utils::create_test_agg_trade(2, "50050.0", "1.0", 1000); // Same ts
        let t3 = test_utils::create_test_agg_trade(3, "50200.0", "1.0", 2000); // Different ts, breach
        let bars = processor.process_agg_trade_records(&[t1, t2, t3]).unwrap();
        assert_eq!(bars.len(), 1, "Should close when breach at different timestamp");
    }

    #[test]
    fn test_streaming_defer_open_semantics() {
        // After breach via process_single_trade, next trade should open new bar
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let t1 = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let t2 = test_utils::create_test_agg_trade(2, "50200.0", "1.0", 2000); // Breach
        let t3 = test_utils::create_test_agg_trade(3, "51000.0", "1.0", 3000); // Opens new bar

        processor.process_single_trade(&t1).unwrap();
        let bar = processor.process_single_trade(&t2).unwrap();
        assert!(bar.is_some(), "Trade 2 should cause a breach");

        // After breach, no incomplete bar (defer_open state)
        assert!(processor.get_incomplete_bar().is_none());

        // Next trade opens a fresh bar
        let bar2 = processor.process_single_trade(&t3).unwrap();
        assert!(bar2.is_none(), "Trade 3 should open new bar, not breach");
        let incomplete = processor.get_incomplete_bar().unwrap();
        assert_eq!(incomplete.open.to_f64(), 51000.0, "New bar should open at t3 price");
    }

    #[test]
    fn test_process_empty_then_trades() {
        // Processing empty slice should be no-op, then normal processing works
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let bars = processor.process_agg_trade_records(&[]).unwrap();
        assert_eq!(bars.len(), 0);
        assert!(processor.get_incomplete_bar().is_none());

        // Now process a real trade
        let trade = test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000);
        let bars = processor.process_agg_trade_records(&[trade]).unwrap();
        assert_eq!(bars.len(), 0);
        assert!(processor.get_incomplete_bar().is_some());
    }

    #[test]
    fn test_multiple_breaches_in_batch() {
        // Multiple bars should form from a batch with repeated breaches
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000),
            test_utils::create_test_agg_trade(2, "50200.0", "1.0", 2000),  // Breach 1
            test_utils::create_test_agg_trade(3, "50500.0", "1.0", 3000),  // Opens bar 2
            test_utils::create_test_agg_trade(4, "50700.0", "1.0", 4000),  // Breach 2
            test_utils::create_test_agg_trade(5, "51000.0", "1.0", 5000),  // Opens bar 3
        ];
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 2, "Should produce 2 completed bars from 2 breaches");
    }

    #[test]
    fn test_streaming_batch_parity_extended() {
        // Task #29: Comprehensive streaming/batch parity with 20 trades producing 5+ bars
        // Uses 100 dbps (0.1%) threshold for more frequent breaches
        let threshold = 100;

        // Build a zigzag price sequence that repeatedly breaches 0.1%
        let mut trades = Vec::new();
        let mut price = 50000.0;
        for i in 0..20 {
            // Alternate up and down movements exceeding 0.1%
            if i % 3 == 0 && i > 0 {
                price *= 1.002; // +0.2% → breach upward
            } else if i % 3 == 1 && i > 1 {
                price *= 0.998; // -0.2% → may breach downward
            } else {
                price *= 1.0005; // Small move, no breach
            }
            trades.push(test_utils::create_test_agg_trade(
                (i + 1) as i64,
                &format!("{:.8}", price),
                "1.0",
                (i as i64 + 1) * 1000,
            ));
        }

        // === BATCH PATH ===
        let mut batch_processor = OpenDeviationBarProcessor::new(threshold).unwrap();
        let batch_bars = batch_processor.process_agg_trade_records(&trades).unwrap();

        // === STREAMING PATH ===
        let mut stream_processor = OpenDeviationBarProcessor::new(threshold).unwrap();
        let mut stream_bars: Vec<OpenDeviationBar> = Vec::new();
        for trade in &trades {
            if let Some(bar) = stream_processor.process_single_trade(trade).unwrap() {
                stream_bars.push(bar);
            }
        }

        // === VERIFY PARITY ===
        assert!(batch_bars.len() >= 3, "Should produce at least 3 bars from zigzag pattern");
        assert_eq!(
            batch_bars.len(), stream_bars.len(),
            "Batch ({}) and streaming ({}) bar count mismatch",
            batch_bars.len(), stream_bars.len()
        );

        for (i, (b, s)) in batch_bars.iter().zip(stream_bars.iter()).enumerate() {
            assert_eq!(b.open, s.open, "Bar {i}: open mismatch");
            assert_eq!(b.close, s.close, "Bar {i}: close mismatch");
            assert_eq!(b.high, s.high, "Bar {i}: high mismatch");
            assert_eq!(b.low, s.low, "Bar {i}: low mismatch");
            assert_eq!(b.volume, s.volume, "Bar {i}: volume mismatch");
            assert_eq!(b.open_time, s.open_time, "Bar {i}: open_time mismatch");
            assert_eq!(b.close_time, s.close_time, "Bar {i}: close_time mismatch");
            assert_eq!(b.individual_trade_count, s.individual_trade_count, "Bar {i}: trade_count mismatch");
        }

        // Verify incomplete bars match
        let batch_inc = batch_processor.get_incomplete_bar();
        let stream_inc = stream_processor.get_incomplete_bar();
        match (&batch_inc, &stream_inc) {
            (Some(b), Some(s)) => {
                assert_eq!(b.open, s.open, "Incomplete: open mismatch");
                assert_eq!(b.volume, s.volume, "Incomplete: volume mismatch");
            }
            (None, None) => {}
            _ => panic!("Incomplete bar presence mismatch"),
        }
    }

    #[test]
    fn test_multi_batch_sequential_state_continuity() {
        // Send 3 separate batches, each producing 1+ bars
        // Verify state carries correctly across batch boundaries
        let mut processor = OpenDeviationBarProcessor::new(100).unwrap(); // 100 dbps = 0.10%
        let mut all_bars = Vec::new();

        // Batch 1: open at 50000, breach needs > 0.10% = price > 50050
        let batch1 = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1000),
            test_utils::create_test_agg_trade(2, "50020.0", "1.0", 2000),
            test_utils::create_test_agg_trade(3, "50060.0", "1.0", 3000), // Breach (>0.10%)
        ];
        let bars1 = processor.process_agg_trade_records(&batch1).unwrap();
        all_bars.extend(bars1);

        // Batch 2: next trade opens new bar, breach again
        let batch2 = vec![
            test_utils::create_test_agg_trade(4, "50100.0", "1.0", 4000), // Opens new bar
            test_utils::create_test_agg_trade(5, "50120.0", "1.0", 5000),
            test_utils::create_test_agg_trade(6, "50170.0", "1.0", 6000), // Breach (>0.10%)
        ];
        let bars2 = processor.process_agg_trade_records(&batch2).unwrap();
        all_bars.extend(bars2);

        // Batch 3: another new bar from fresh state
        let batch3 = vec![
            test_utils::create_test_agg_trade(7, "50200.0", "1.0", 7000), // Opens new bar
            test_utils::create_test_agg_trade(8, "50220.0", "1.0", 8000),
            test_utils::create_test_agg_trade(9, "50280.0", "1.0", 9000), // Breach (>0.10%)
        ];
        let bars3 = processor.process_agg_trade_records(&batch3).unwrap();
        all_bars.extend(bars3);

        // Should have produced at least 3 bars (one per batch boundary)
        assert!(
            all_bars.len() >= 3,
            "Expected at least 3 bars from 3 batches, got {}",
            all_bars.len()
        );

        // Timestamps must be strictly monotonic
        for i in 1..all_bars.len() {
            assert!(
                all_bars[i].close_time >= all_bars[i - 1].close_time,
                "Bar {i}: close_time {} < previous {}",
                all_bars[i].close_time,
                all_bars[i - 1].close_time
            );
        }

        // Trade IDs should be continuous across batches
        for i in 1..all_bars.len() {
            assert_eq!(
                all_bars[i].first_agg_trade_id,
                all_bars[i - 1].last_agg_trade_id + 1,
                "Bar {i}: trade ID gap (first={}, prev last={})",
                all_bars[i].first_agg_trade_id,
                all_bars[i - 1].last_agg_trade_id
            );
        }
    }

    // Issue #96 Task #93: Edge case tests for processor algorithm invariants

    #[test]
    fn test_same_timestamp_prevents_bar_close() {
        // Issue #36: Bar cannot close on same timestamp as it opened
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        processor.prevent_same_timestamp_close = true;

        // All trades at same timestamp but price breaches threshold
        let trades: Vec<AggTrade> = (0..5)
            .map(|i| {
                let price_str = if i == 0 {
                    "50000.0".to_string()
                } else {
                    // Price far above threshold to trigger breach
                    format!("{}.0", 50000 + (i + 1) * 200)
                };
                AggTrade {
                    agg_trade_id: i as i64,
                    price: FixedPoint::from_str(&price_str).unwrap(),
                    volume: FixedPoint::from_str("1.0").unwrap(),
                    first_trade_id: i as i64,
                    last_trade_id: i as i64,
                    timestamp: 1000000, // ALL same timestamp
                    is_buyer_maker: false,
                    is_best_match: None,
                }
            })
            .collect();

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        // No bars should close because timestamp gate blocks it
        assert_eq!(bars.len(), 0, "Same timestamp should prevent bar close (Issue #36)");
    }

    #[test]
    fn test_single_trade_incomplete_bar() {
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        let trade = AggTrade {
            agg_trade_id: 1,
            price: FixedPoint::from_str("50000.0").unwrap(),
            volume: FixedPoint::from_str("10.0").unwrap(),
            first_trade_id: 1,
            last_trade_id: 1,
            timestamp: 1000000,
            is_buyer_maker: false,
            is_best_match: None,
        };

        // Strict mode: 0 completed bars
        let bars = processor.process_agg_trade_records(&[trade.clone()]).unwrap();
        assert_eq!(bars.len(), 0, "Single trade cannot complete a bar");

        // With incomplete: should return 1 incomplete bar
        let mut processor2 = OpenDeviationBarProcessor::new(250).unwrap();
        let bars_incl = processor2
            .process_agg_trade_records_with_incomplete(&[trade])
            .unwrap();
        assert_eq!(bars_incl.len(), 1, "Should return 1 incomplete bar");
        assert_eq!(bars_incl[0].open, bars_incl[0].close);
        assert_eq!(bars_incl[0].high, bars_incl[0].low);
    }

    // === Issue #96: Configuration method coverage tests ===

    #[test]
    fn test_with_options_gate_disabled_same_timestamp_closes() {
        // Issue #36: with prevent_same_timestamp_close=false, bar should close
        // even when breach trade has same timestamp as open
        let mut processor = OpenDeviationBarProcessor::with_options(250, false).unwrap();
        assert!(!processor.prevent_same_timestamp_close());

        let trades = vec![
            AggTrade {
                agg_trade_id: 1, price: FixedPoint::from_str("50000.0").unwrap(),
                volume: FixedPoint::from_str("1.0").unwrap(),
                first_trade_id: 1, last_trade_id: 1, timestamp: 1000000,
                is_buyer_maker: false, is_best_match: None,
            },
            AggTrade {
                agg_trade_id: 2, price: FixedPoint::from_str("50200.0").unwrap(), // +0.4% > 0.25%
                volume: FixedPoint::from_str("1.0").unwrap(),
                first_trade_id: 2, last_trade_id: 2, timestamp: 1000000, // Same timestamp!
                is_buyer_maker: false, is_best_match: None,
            },
        ];
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 1, "Gate disabled: same-timestamp breach should close bar");
    }

    #[test]
    fn test_inter_bar_config_enables_features() {
        use crate::interbar::LookbackMode;
        let processor = OpenDeviationBarProcessor::new(250).unwrap();
        assert!(!processor.inter_bar_enabled(), "Default: inter-bar disabled");

        let processor = processor.with_inter_bar_config(InterBarConfig {
            lookback_mode: LookbackMode::FixedCount(100),
            compute_tier2: false,
            compute_tier3: false,
            ..Default::default()
        });
        assert!(processor.inter_bar_enabled(), "After config: inter-bar enabled");
    }

    #[test]
    fn test_intra_bar_feature_toggle() {
        let processor = OpenDeviationBarProcessor::new(250).unwrap();
        assert!(!processor.intra_bar_enabled(), "Default: intra-bar disabled");

        let processor = processor.with_intra_bar_features();
        assert!(processor.intra_bar_enabled(), "After toggle: intra-bar enabled");
    }

    #[test]
    fn test_set_inter_bar_config_after_construction() {
        use crate::interbar::LookbackMode;
        let mut processor = OpenDeviationBarProcessor::new(500).unwrap();
        assert!(!processor.inter_bar_enabled());

        processor.set_inter_bar_config(InterBarConfig {
            lookback_mode: LookbackMode::FixedCount(200),
            compute_tier2: true,
            compute_tier3: false,
            ..Default::default()
        });
        assert!(processor.inter_bar_enabled(), "set_inter_bar_config should enable");
    }

    #[test]
    fn test_process_with_options_incomplete_false_vs_true() {
        let trades = scenarios::single_breach_sequence(250);

        // Without incomplete: only completed bars
        let mut p1 = OpenDeviationBarProcessor::new(250).unwrap();
        let bars_strict = p1.process_agg_trade_records_with_options(&trades, false).unwrap();

        // With incomplete: completed + 1 partial
        let mut p2 = OpenDeviationBarProcessor::new(250).unwrap();
        let bars_incl = p2.process_agg_trade_records_with_options(&trades, true).unwrap();

        assert!(
            bars_incl.len() >= bars_strict.len(),
            "inclusive ({}) must be >= strict ({})", bars_incl.len(), bars_strict.len()
        );
    }

    // === Issue #96: Untested public method coverage ===

    #[test]
    fn test_anomaly_summary_default_no_anomalies() {
        let processor = OpenDeviationBarProcessor::new(250).unwrap();
        let summary = processor.anomaly_summary();
        assert_eq!(summary.gaps_detected, 0);
        assert_eq!(summary.overlaps_detected, 0);
        assert_eq!(summary.timestamp_anomalies, 0);
        assert!(!summary.has_anomalies());
        assert_eq!(summary.total(), 0);
    }

    #[test]
    fn test_anomaly_summary_preserved_through_checkpoint() {
        // Process some trades, create checkpoint, restore, verify anomaly state
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trades = scenarios::single_breach_sequence(250);
        processor.process_agg_trade_records(&trades).unwrap();

        let checkpoint = processor.create_checkpoint("TEST");
        let restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        let summary = restored.anomaly_summary();
        // Default processor has no anomalies; checkpoint preserves that
        assert_eq!(summary.total(), 0);
    }

    #[test]
    fn test_anomaly_summary_from_checkpoint_with_anomalies() {
        // Deserialize a checkpoint that has anomaly data
        let json = r#"{
            "version": 3,
            "symbol": "TESTUSDT",
            "threshold_decimal_bps": 250,
            "prevent_same_timestamp_close": true,
            "defer_open": false,
            "current_bar": null,
            "thresholds": null,
            "last_timestamp_us": 1000000,
            "last_trade_id": 5,
            "price_hash": 0,
            "anomaly_summary": {"gaps_detected": 3, "overlaps_detected": 1, "timestamp_anomalies": 2}
        }"#;
        let checkpoint: crate::checkpoint::Checkpoint = serde_json::from_str(json).unwrap();
        let processor = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        let summary = processor.anomaly_summary();
        assert_eq!(summary.gaps_detected, 3);
        assert_eq!(summary.overlaps_detected, 1);
        assert_eq!(summary.timestamp_anomalies, 2);
        assert!(summary.has_anomalies());
        assert_eq!(summary.total(), 6);
    }

    #[test]
    fn test_with_inter_bar_config_and_cache_shared() {
        use crate::entropy_cache_global::get_global_entropy_cache;
        use crate::interbar::LookbackMode;

        let global_cache = get_global_entropy_cache();
        let config = InterBarConfig {
            lookback_mode: LookbackMode::FixedCount(100),
            compute_tier2: true,
            compute_tier3: true,
            ..Default::default()
        };

        // Two processors sharing the same global cache
        let p1 = OpenDeviationBarProcessor::new(250).unwrap()
            .with_inter_bar_config_and_cache(config.clone(), Some(global_cache.clone()));
        let p2 = OpenDeviationBarProcessor::new(500).unwrap()
            .with_inter_bar_config_and_cache(config, Some(global_cache));

        assert!(p1.inter_bar_enabled());
        assert!(p2.inter_bar_enabled());
    }

    #[test]
    fn test_set_inter_bar_config_with_cache_after_checkpoint() {
        use crate::entropy_cache_global::get_global_entropy_cache;
        use crate::interbar::LookbackMode;

        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let trades = scenarios::single_breach_sequence(250);
        processor.process_agg_trade_records(&trades).unwrap();

        // Simulate checkpoint round-trip (inter-bar config not preserved)
        let checkpoint = processor.create_checkpoint("TEST");
        let mut restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();
        assert!(!restored.inter_bar_enabled(), "Checkpoint does not preserve inter-bar config");

        // Re-enable with shared cache
        let global_cache = get_global_entropy_cache();
        restored.set_inter_bar_config_with_cache(
            InterBarConfig {
                lookback_mode: LookbackMode::FixedCount(100),
                compute_tier2: false,
                compute_tier3: false,
                ..Default::default()
            },
            Some(global_cache),
        );
        assert!(restored.inter_bar_enabled(), "set_inter_bar_config_with_cache should re-enable");
    }

    #[test]
    fn test_threshold_decimal_bps_getter() {
        let p250 = OpenDeviationBarProcessor::new(250).unwrap();
        assert_eq!(p250.threshold_decimal_bps(), 250);

        let p1000 = OpenDeviationBarProcessor::new(1000).unwrap();
        assert_eq!(p1000.threshold_decimal_bps(), 1000);
    }

    // =========================================================================
    // Issue #112: Gap-aware checkpoint recovery tests
    // =========================================================================

    #[test]
    fn test_checkpoint_gap_discards_forming_bar() {
        // Process some trades, checkpoint, then resume with a 2-hour gap
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Create trades that don't breach (stay within 0.25%)
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200_000_000), // t=0
            test_utils::create_test_agg_trade(2, "50010.0", "1.0", 1640995201_000_000), // t=+1s
            test_utils::create_test_agg_trade(3, "50020.0", "1.0", 1640995202_000_000), // t=+2s
        ];

        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 0, "No breach = no completed bars");

        // Create checkpoint (should have forming bar)
        let checkpoint = processor.create_checkpoint("BTCUSDT");
        assert!(checkpoint.has_incomplete_bar(), "Should have forming bar");

        // Resume from checkpoint
        let mut restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();

        // Feed trades with 2-hour gap (7,200,000,000 μs > 1-hour default max_gap)
        let gap_trades = vec![
            test_utils::create_test_agg_trade(4, "50030.0", "1.0", 1641002402_000_000), // +2h gap
            test_utils::create_test_agg_trade(5, "50040.0", "1.0", 1641002403_000_000),
        ];

        let bars = restored.process_agg_trade_records(&gap_trades).unwrap();
        // Forming bar should have been discarded, no oversized bar emitted
        assert_eq!(bars.len(), 0, "No bars should complete — forming bar was discarded");
        assert_eq!(
            restored.anomaly_summary().gaps_detected, 1,
            "Gap should be recorded in anomaly summary"
        );
    }

    #[test]
    fn test_checkpoint_small_gap_continues_bar() {
        // Resume with a gap UNDER the threshold — bar should continue
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200_000_000),
            test_utils::create_test_agg_trade(2, "50010.0", "1.0", 1640995201_000_000),
        ];

        let _ = processor.process_agg_trade_records(&trades).unwrap();
        let checkpoint = processor.create_checkpoint("BTCUSDT");
        let mut restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();

        // Feed trades with 30-minute gap (1,800,000,000 μs < 1-hour max_gap)
        let small_gap_trades = vec![
            test_utils::create_test_agg_trade(3, "50020.0", "1.0", 1640997001_000_000), // +30m
            test_utils::create_test_agg_trade(4, "50125.01", "1.0", 1640997002_000_000), // breach
        ];

        let bars = restored.process_agg_trade_records(&small_gap_trades).unwrap();
        assert_eq!(bars.len(), 1, "Bar should complete normally with small gap");
        // Bar should span from original open (trade 1) through gap trades
        assert_eq!(bars[0].open_time, 1640995200_000_000);
        assert_eq!(
            restored.anomaly_summary().gaps_detected, 0,
            "No gap anomaly for small gap"
        );
    }

    #[test]
    fn test_checkpoint_gap_custom_max_gap() {
        // Test with custom max_gap_us set to 30 minutes
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200_000_000),
        ];
        let _ = processor.process_agg_trade_records(&trades).unwrap();
        let checkpoint = processor.create_checkpoint("BTCUSDT");

        // Restore with custom 30-minute max gap
        let mut restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint)
            .unwrap()
            .with_max_gap(1_800_000_000); // 30 minutes

        // 45-minute gap should discard with 30-min threshold
        let gap_trades = vec![
            test_utils::create_test_agg_trade(2, "50010.0", "1.0", 1640997900_000_000), // +45min
        ];

        let _ = restored.process_agg_trade_records(&gap_trades).unwrap();
        assert_eq!(
            restored.anomaly_summary().gaps_detected, 1,
            "45-min gap should be detected with 30-min threshold"
        );
    }

    #[test]
    fn test_is_valid_range_rejects_oversized() {
        use crate::fixed_point::FixedPoint;

        let threshold_decimal_bps: u32 = 250; // 0.25%
        let threshold_ratio = ((threshold_decimal_bps as i64) * crate::fixed_point::SCALE)
            / (crate::fixed_point::BASIS_POINTS_SCALE as i64);

        // Bar with 0.50% range — exceeds 2x threshold (2 * 0.25% = 0.50%)
        // open=50000.0, high=50250.01, low=50000.0 → range=250.01/50000 ≈ 0.5%+
        let mut oversized = OpenDeviationBar::default();
        oversized.open = FixedPoint::from_str("50000.0").unwrap();
        oversized.high = FixedPoint::from_str("50250.01").unwrap();
        oversized.low = FixedPoint::from_str("50000.0").unwrap();
        assert!(
            !oversized.is_valid_range(threshold_ratio, 2),
            "Bar exceeding 2x threshold should be invalid"
        );

        // Bar with 0.20% range — within threshold
        let mut valid = OpenDeviationBar::default();
        valid.open = FixedPoint::from_str("50000.0").unwrap();
        valid.high = FixedPoint::from_str("50100.0").unwrap();
        valid.low = FixedPoint::from_str("50000.0").unwrap();
        assert!(
            valid.is_valid_range(threshold_ratio, 2),
            "Bar within threshold should be valid"
        );

        // Bar at exact breach boundary (0.25%) — should be valid (breach triggers AT threshold)
        let mut exact = OpenDeviationBar::default();
        exact.open = FixedPoint::from_str("50000.0").unwrap();
        exact.high = FixedPoint::from_str("50125.0").unwrap();
        exact.low = FixedPoint::from_str("50000.0").unwrap();
        assert!(
            exact.is_valid_range(threshold_ratio, 2),
            "Bar at exact threshold should be valid"
        );
    }

    #[test]
    fn test_checkpoint_no_incomplete_bar_gap_is_noop() {
        // If checkpoint has no forming bar, gap detection is irrelevant
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Process trades that complete a bar (produce breach)
        let trades = vec![
            test_utils::create_test_agg_trade(1, "50000.0", "1.0", 1640995200_000_000),
            test_utils::create_test_agg_trade(2, "50200.0", "1.0", 1640995201_000_000), // breach
        ];
        let bars = processor.process_agg_trade_records(&trades).unwrap();
        assert_eq!(bars.len(), 1);

        let checkpoint = processor.create_checkpoint("BTCUSDT");
        assert!(!checkpoint.has_incomplete_bar());

        let mut restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();

        // Large gap doesn't matter — no forming bar to discard
        let gap_trades = vec![
            test_utils::create_test_agg_trade(3, "50010.0", "1.0", 1641081600_000_000), // +24h
        ];
        let _ = restored.process_agg_trade_records(&gap_trades).unwrap();
        assert_eq!(
            restored.anomaly_summary().gaps_detected, 0,
            "No gap anomaly when no forming bar exists"
        );
    }
}