opendeviationbar-core 13.75.0

//! Streaming spread statistics accumulator for bid/ask tick data.
//!
//! Computes up to 37 spread microstructure features in O(1) per tick with no heap allocation.
//! All output statistics are `Option<f64>` with mathematically justified N-guards.
//!
//! - Phase 53 (SA-01, SA-03): 26 base features, always computed.
//! - Phase 54 (SA-02, SA-04): 11 advanced features behind `compute_advanced` toggle.
//!
//! ## Algorithm
//!
//! - Welford M1-M4 online algorithm (Cook 2022) for mean, variance, skewness, kurtosis
//! - TWAS: time-weighted average spread using previous-value (step function) convention
//! - Kaufman Efficiency Ratio: |net change| / sum(|tick changes|)
//! - Time-at-wide/tight: cumulative dt classified against running Welford mean
//! - Online co-moment for spread-price correlation and spread autocorrelation

use serde::{Deserialize, Serialize};

/// Output: up to 37 spread statistics extracted from SpreadAccumulator at bar finalization.
/// All fields are `Option<f64>` -- `None` when the mathematical precondition is not met.
/// Base features (Groups 1-7, 26 fields) are always computed.
/// Advanced features (Group 8, 11 fields) are None when `compute_advanced` is false.
#[derive(Debug, Clone, Default)]
pub struct SpreadFeatures {
    // Group 1: Spread OHLC (N >= 1)
    pub spread_open: Option<f64>,
    pub spread_high: Option<f64>,
    pub spread_low: Option<f64>,
    pub spread_close: Option<f64>,

    // Group 2: TWAS (sum_dt > 0)
    pub twas: Option<f64>,

    // Group 3: Welford M1-M4
    pub spread_mean: Option<f64>,         // N >= 1
    pub spread_variance: Option<f64>,     // N >= 2
    pub spread_skewness: Option<f64>,     // N >= 3 AND M2 > eps
    pub spread_kurtosis: Option<f64>,     // N >= 4 AND M2 > eps

    // Group 4: Quote asymmetry and imbalance
    pub quote_asymmetry: Option<f64>,     // asymmetry_count > 0
    pub bid_ask_imbalance: Option<f64>,   // N >= 2

    // Group 5: Time-at-wide/tight (N >= 2 AND sum_dt > 0)
    pub time_at_wide_ratio: Option<f64>,
    pub time_at_tight_ratio: Option<f64>,

    // Group 6: Bid/Ask OHLC (N >= 1)
    pub bid_open: Option<f64>,
    pub bid_high: Option<f64>,
    pub bid_low: Option<f64>,
    pub bid_close: Option<f64>,
    pub ask_open: Option<f64>,
    pub ask_high: Option<f64>,
    pub ask_low: Option<f64>,
    pub ask_close: Option<f64>,

    // Group 7: Kaufman ER and derived
    pub spread_kaufman_er: Option<f64>,       // N >= 2
    pub spread_range: Option<f64>,            // N >= 1 AND mean > eps
    pub spread_open_close_ratio: Option<f64>, // N >= 1 AND open > eps
    pub tick_count_with_quotes: Option<f64>,  // Always Some
    pub total_quote_duration_us: Option<f64>, // N >= 1

    // Group 8: Advanced Features (Phase 54, SA-02)
    // All None when compute_advanced=false (SA-04 toggle)
    pub spread_at_breach: Option<f64>,         // N>=2, mean>eps
    pub quote_side_imbalance: Option<f64>,     // N>=2, denominator>eps
    pub quote_staleness_ratio: Option<f64>,    // N>=2
    pub signed_tick_ratio: Option<f64>,        // N>=2, total>0
    pub spread_price_correlation: Option<f64>, // N>=3, both stds>eps
    pub tick_arrival_cv: Option<f64>,          // N>=3 (>=2 dt values)
    pub spread_trajectory_shape: Option<f64>,  // N>=1, range>eps
    pub executable_spread_ratio: Option<f64>,  // N>=2
    pub spread_autocorrelation: Option<f64>,   // N>=3, M2>eps
    pub mid_momentum_ratio: Option<f64>,       // N>=1, range>eps
    pub worst_spread: Option<f64>,             // N>=1, mean>eps
}

/// Streaming spread statistics accumulator.
/// Updated O(1) per tick. No heap allocation.
/// All output statistics are `Option<f64>` with mathematically justified N-guards.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SpreadAccumulator {
    // === Tick counter (8 bytes) ===
    n: u64,

    // === Spread OHLC (32 bytes) ===
    spread_open: f64,
    spread_high: f64,
    spread_low: f64,
    spread_close: f64,

    // === TWAS state (16 bytes) ===
    twas_sum_spread_dt: f64,
    twas_sum_dt: f64,

    // === Welford M1-M4 (40 bytes) ===
    welford_mean: f64,
    welford_m2: f64,
    welford_m3: f64,
    welford_m4: f64,
    welford_n: f64, // as f64 to avoid casts in hot path

    // === Quote asymmetry (16 bytes) ===
    asymmetry_sum: f64,
    asymmetry_count: u64,

    // === Bid-ask imbalance (24 bytes) ===
    prev_bid: f64,
    prev_ask: f64,
    imbalance_sum: f64,

    // === Time-at-wide/tight (24 bytes) ===
    time_wide_us: f64,
    time_tight_us: f64,
    prev_spread: f64,

    // === Bid/Ask OHLC (64 bytes) ===
    bid_open: f64,
    bid_high: f64,
    bid_low: f64,
    bid_close: f64,
    ask_open: f64,
    ask_high: f64,
    ask_low: f64,
    ask_close: f64,

    // === Kaufman ER (24 bytes) ===
    kaufman_first_spread: f64,
    kaufman_sum_abs_changes: f64,
    kaufman_prev_spread: f64,

    // === Timestamps (24 bytes) ===
    first_timestamp_us: i64,
    last_timestamp_us: i64,
    prev_timestamp_us: i64,

    // === Advanced toggle (Phase 54, SA-04) ===
    // When false, all advanced state updates are skipped (zero overhead).
    #[serde(default)]
    compute_advanced: bool,

    // === Tier 2 Advanced State (Phase 54) ===

    // quote_side_imbalance: bid/ask move accumulators (16 bytes)
    #[serde(default)]
    adv_bid_move_sum: f64,
    #[serde(default)]
    adv_ask_move_sum: f64,

    // quote_staleness_ratio: stale tick counter (8 bytes)
    #[serde(default)]
    adv_stale_count: u64,

    // signed_tick_ratio: up/down mid-price tick counters (16 bytes)
    #[serde(default)]
    adv_up_ticks: u64,
    #[serde(default)]
    adv_down_ticks: u64,

    // spread_price_correlation: mid Welford + co-moment (32 bytes)
    #[serde(default)]
    adv_mid_mean: f64,
    #[serde(default)]
    adv_mid_m2: f64,
    #[serde(default)]
    adv_co_moment_sxy: f64,
    #[serde(default)]
    adv_prev_mid: f64,

    // tick_arrival_cv: inter-arrival Welford (24 bytes)
    #[serde(default)]
    adv_dt_mean: f64,
    #[serde(default)]
    adv_dt_m2: f64,
    #[serde(default)]
    adv_dt_n: f64,

    // executable_spread_ratio: count of spreads <= 2*mean (8 bytes)
    #[serde(default)]
    adv_executable_count: u64,

    // === Tier 3 Advanced State (Phase 54) ===

    // spread_autocorrelation: lag-1 co-moment (16 bytes -- reuses prev_spread from base)
    #[serde(default)]
    adv_autocorr_sxy: f64,
    #[serde(default)]
    adv_autocorr_n: f64,

    // mid_momentum_ratio: mid OHLC (32 bytes)
    #[serde(default)]
    adv_mid_open: f64,
    #[serde(default)]
    adv_mid_high: f64,
    #[serde(default)]
    adv_mid_low: f64,
    #[serde(default)]
    adv_mid_close: f64,
}

impl SpreadAccumulator {
    /// Create a new SpreadAccumulator with the advanced feature toggle (Phase 54, SA-04).
    /// When `compute_advanced` is true, 11 additional features are computed.
    /// When false, advanced features return None with zero overhead.
    pub fn new(compute_advanced: bool) -> Self {
        Self {
            compute_advanced,
            ..Default::default()
        }
    }

    /// Get the compute_advanced setting (Phase 54, SA-04).
    pub fn compute_advanced(&self) -> bool {
        self.compute_advanced
    }

    /// O(1) per-tick update. Call for every tick with valid bid/ask data.
    #[inline]
    pub fn update(&mut self, bid: f64, ask: f64, timestamp_us: i64) {
        let spread = ask - bid;
        self.n += 1;
        let n = self.n;

        if n == 1 {
            // First tick: initialize all OHLC opens
            self.spread_open = spread;
            self.spread_high = spread;
            self.spread_low = spread;
            self.bid_open = bid;
            self.bid_high = bid;
            self.bid_low = bid;
            self.ask_open = ask;
            self.ask_high = ask;
            self.ask_low = ask;
            self.kaufman_first_spread = spread;
            self.first_timestamp_us = timestamp_us;
        } else {
            // Update OHLC highs/lows
            if spread > self.spread_high {
                self.spread_high = spread;
            }
            if spread < self.spread_low {
                self.spread_low = spread;
            }
            if bid > self.bid_high {
                self.bid_high = bid;
            }
            if bid < self.bid_low {
                self.bid_low = bid;
            }
            if ask > self.ask_high {
                self.ask_high = ask;
            }
            if ask < self.ask_low {
                self.ask_low = ask;
            }

            // TWAS: accumulate prev_spread * dt (step function convention)
            let dt = (timestamp_us - self.prev_timestamp_us) as f64;
            if dt > 0.0 {
                self.twas_sum_spread_dt += self.prev_spread * dt;
                self.twas_sum_dt += dt;
            }

            // Bid-ask imbalance: (bid_move - ask_move) / spread
            if spread.abs() > f64::EPSILON {
                let bid_move = bid - self.prev_bid;
                let ask_move = ask - self.prev_ask;
                self.imbalance_sum += (bid_move - ask_move) / spread;
            }

            // Kaufman ER: accumulate |spread change|
            self.kaufman_sum_abs_changes += (spread - self.kaufman_prev_spread).abs();

            // Time-at-wide/tight using running Welford mean as threshold
            if dt > 0.0 {
                if self.prev_spread > self.welford_mean {
                    self.time_wide_us += dt;
                } else {
                    self.time_tight_us += dt;
                }
            }
        }

        // Always update closes
        self.spread_close = spread;
        self.bid_close = bid;
        self.ask_close = ask;

        // Quote asymmetry: count ticks with nonzero spread
        if spread.abs() > f64::EPSILON {
            // For symmetric bid/ask quotes, asymmetry is 0 by definition.
            // The accumulator tracks count for the N-guard.
            self.asymmetry_count += 1;
            // asymmetry_sum stays 0 for symmetric quotes (mid == (bid+ask)/2)
        }

        // Welford M1-M4 (must be after time-at-wide/tight which uses welford_mean)
        self.welford_update(spread);

        // === Advanced feature updates (Phase 54, SA-02) ===
        // Runs AFTER welford_update (welford_mean is current) but BEFORE
        // prev_spread/prev_bid/prev_ask update (they still hold previous values).
        if self.compute_advanced {
            let mid = (bid + ask) * 0.5;

            if n == 1 {
                // First tick: initialize mid OHLC
                self.adv_mid_open = mid;
                self.adv_mid_high = mid;
                self.adv_mid_low = mid;
                self.adv_prev_mid = mid;
            } else {
                // quote_side_imbalance: accumulate bid/ask moves
                self.adv_bid_move_sum += bid - self.prev_bid;
                self.adv_ask_move_sum += ask - self.prev_ask;

                // quote_staleness_ratio: detect unchanged quotes
                if (bid - self.prev_bid).abs() <= f64::EPSILON
                    && (ask - self.prev_ask).abs() <= f64::EPSILON
                {
                    self.adv_stale_count += 1;
                }

                // signed_tick_ratio: mid-price direction
                if mid > self.adv_prev_mid + f64::EPSILON {
                    self.adv_up_ticks += 1;
                } else if mid < self.adv_prev_mid - f64::EPSILON {
                    self.adv_down_ticks += 1;
                }

                // spread_price_correlation: Welford co-moment for (spread, mid)
                // dx = spread - welford_mean (delta AFTER mean update)
                let dx = spread - self.welford_mean;
                let old_mid_mean = self.adv_mid_mean;
                let mid_delta = mid - self.adv_mid_mean;
                self.adv_mid_mean += mid_delta / self.welford_n;
                let mid_delta2 = mid - self.adv_mid_mean;
                self.adv_mid_m2 += mid_delta * mid_delta2;
                self.adv_co_moment_sxy += dx * (mid - old_mid_mean);

                // tick_arrival_cv: Welford on inter-arrival times
                let dt = (timestamp_us - self.prev_timestamp_us) as f64;
                self.adv_dt_n += 1.0;
                let dt_delta = dt - self.adv_dt_mean;
                self.adv_dt_mean += dt_delta / self.adv_dt_n;
                self.adv_dt_m2 += dt_delta * (dt - self.adv_dt_mean);

                // spread_autocorrelation: lag-1 co-moment
                // Uses self.prev_spread which still holds previous spread
                self.adv_autocorr_sxy +=
                    (spread - self.welford_mean) * (self.prev_spread - self.welford_mean);
                self.adv_autocorr_n += 1.0;

                // mid OHLC: update high/low
                if mid > self.adv_mid_high {
                    self.adv_mid_high = mid;
                }
                if mid < self.adv_mid_low {
                    self.adv_mid_low = mid;
                }

                self.adv_prev_mid = mid;
            }

            // Always update mid close
            self.adv_mid_close = mid;

            // executable_spread_ratio: count spreads <= 2*mean
            // welford_mean is already updated by welford_update above
            if spread <= 2.0 * self.welford_mean {
                self.adv_executable_count += 1;
            }
        }

        // Update previous values (AFTER advanced block so prev_spread is still lag-1)
        self.prev_bid = bid;
        self.prev_ask = ask;
        self.prev_spread = spread;
        self.prev_timestamp_us = timestamp_us;
        self.kaufman_prev_spread = spread;
        self.last_timestamp_us = timestamp_us;
    }

    /// O(1) Welford online update for M1 (mean), M2, M3, M4.
    /// Source: Cook (2022), extending Knuth-Welford.
    /// CRITICAL: M4 updated BEFORE M3, M3 before M2 (each uses pre-update values).
    #[inline]
    fn welford_update(&mut self, spread: f64) {
        let n1 = self.welford_n; // n before increment
        self.welford_n += 1.0;
        let n = self.welford_n;

        let delta = spread - self.welford_mean;
        let delta_n = delta / n;
        let delta_n2 = delta_n * delta_n;
        let term1 = delta * delta_n * n1;

        self.welford_mean += delta_n;

        // M4 must be updated BEFORE M3, M3 before M2
        self.welford_m4 += term1 * delta_n2 * (n * n - 3.0 * n + 3.0)
            + 6.0 * delta_n2 * self.welford_m2
            - 4.0 * delta_n * self.welford_m3;
        self.welford_m3 +=
            term1 * delta_n * (n - 2.0) - 3.0 * delta_n * self.welford_m2;
        self.welford_m2 += term1;
    }

    /// Extract all statistics with mathematically justified N-guards.
    /// Returns `SpreadFeatures` with `Option<f64>` fields.
    /// Base features (26) are always computed. Advanced features (11) are None
    /// when `compute_advanced` is false (SA-04 toggle).
    pub fn finalize(&self) -> SpreadFeatures {
        let n = self.n;
        let has_data = n >= 1;
        let has_two = n >= 2;
        let sum_dt_positive = self.twas_sum_dt > 0.0;

        SpreadFeatures {
            // Group 1: Spread OHLC (N >= 1)
            spread_open: if has_data {
                Some(self.spread_open)
            } else {
                None
            },
            spread_high: if has_data {
                Some(self.spread_high)
            } else {
                None
            },
            spread_low: if has_data {
                Some(self.spread_low)
            } else {
                None
            },
            spread_close: if has_data {
                Some(self.spread_close)
            } else {
                None
            },

            // Group 2: TWAS (sum_dt > 0)
            twas: if sum_dt_positive {
                Some(self.twas_sum_spread_dt / self.twas_sum_dt)
            } else {
                None
            },

            // Group 3: Welford moments
            spread_mean: if has_data {
                Some(self.welford_mean)
            } else {
                None
            },
            spread_variance: if has_two {
                Some(self.welford_m2 / (n as f64 - 1.0))
            } else {
                None
            },
            spread_skewness: if n >= 3 && self.welford_m2 > f64::EPSILON {
                let n_f = n as f64;
                Some(n_f.sqrt() * self.welford_m3 / self.welford_m2.powf(1.5))
            } else {
                None
            },
            spread_kurtosis: if n >= 4 && self.welford_m2 > f64::EPSILON {
                let n_f = n as f64;
                Some(n_f * self.welford_m4 / (self.welford_m2 * self.welford_m2) - 3.0)
            } else {
                None
            },

            // Group 4: Quote asymmetry and imbalance
            quote_asymmetry: if self.asymmetry_count > 0 {
                Some(self.asymmetry_sum / self.asymmetry_count as f64)
            } else {
                None
            },
            bid_ask_imbalance: if has_two {
                Some(self.imbalance_sum / (n as f64 - 1.0))
            } else {
                None
            },

            // Group 5: Time-at-wide/tight (N >= 2 AND sum_dt > 0)
            time_at_wide_ratio: if has_two && sum_dt_positive {
                Some(self.time_wide_us / self.twas_sum_dt)
            } else {
                None
            },
            time_at_tight_ratio: if has_two && sum_dt_positive {
                Some(self.time_tight_us / self.twas_sum_dt)
            } else {
                None
            },

            // Group 6: Bid/Ask OHLC (N >= 1)
            bid_open: if has_data { Some(self.bid_open) } else { None },
            bid_high: if has_data { Some(self.bid_high) } else { None },
            bid_low: if has_data { Some(self.bid_low) } else { None },
            bid_close: if has_data {
                Some(self.bid_close)
            } else {
                None
            },
            ask_open: if has_data { Some(self.ask_open) } else { None },
            ask_high: if has_data { Some(self.ask_high) } else { None },
            ask_low: if has_data { Some(self.ask_low) } else { None },
            ask_close: if has_data {
                Some(self.ask_close)
            } else {
                None
            },

            // Group 7: Kaufman ER and derived
            spread_kaufman_er: if has_two {
                let net_change = (self.spread_close - self.kaufman_first_spread).abs();
                if self.kaufman_sum_abs_changes > f64::EPSILON {
                    Some(net_change / self.kaufman_sum_abs_changes)
                } else {
                    // All spreads identical -> no changes -> ER undefined but
                    // net_change is also 0, so 0/0. Return Some(0.0) since
                    // no movement = no efficiency.
                    Some(0.0)
                }
            } else {
                None
            },
            spread_range: if has_data && self.welford_mean > f64::EPSILON {
                Some((self.spread_high - self.spread_low) / self.welford_mean)
            } else {
                None
            },
            spread_open_close_ratio: if has_data && self.spread_open > f64::EPSILON {
                Some(self.spread_close / self.spread_open)
            } else {
                None
            },
            tick_count_with_quotes: Some(n as f64),
            total_quote_duration_us: if has_data {
                Some((self.last_timestamp_us - self.first_timestamp_us) as f64)
            } else {
                None
            },

            // Group 8: Advanced (Phase 54, SA-02)
            // All None when compute_advanced is false (SA-04)
            spread_at_breach: if self.compute_advanced && has_two && self.welford_mean > f64::EPSILON {
                Some(self.spread_close / self.welford_mean)
            } else {
                None
            },

            quote_side_imbalance: if self.compute_advanced && has_two {
                let denom = self.adv_bid_move_sum.abs() + self.adv_ask_move_sum.abs();
                if denom > f64::EPSILON {
                    Some((self.adv_bid_move_sum - self.adv_ask_move_sum) / denom)
                } else {
                    None
                }
            } else {
                None
            },

            quote_staleness_ratio: if self.compute_advanced && has_two {
                Some(self.adv_stale_count as f64 / n as f64)
            } else {
                None
            },

            signed_tick_ratio: if self.compute_advanced && has_two {
                let total = self.adv_up_ticks + self.adv_down_ticks;
                if total > 0 {
                    Some((self.adv_up_ticks as f64 - self.adv_down_ticks as f64) / total as f64)
                } else {
                    None
                }
            } else {
                None
            },

            spread_price_correlation: if self.compute_advanced && n >= 3 {
                let std_spread = self.welford_m2.sqrt();
                let std_mid = self.adv_mid_m2.sqrt();
                if std_spread > f64::EPSILON && std_mid > f64::EPSILON {
                    Some(self.adv_co_moment_sxy / (std_spread * std_mid))
                } else {
                    None
                }
            } else {
                None
            },

            tick_arrival_cv: if self.compute_advanced && self.adv_dt_n >= 2.0 {
                let dt_var = self.adv_dt_m2 / (self.adv_dt_n - 1.0);
                if self.adv_dt_mean > f64::EPSILON {
                    Some(dt_var.sqrt() / self.adv_dt_mean)
                } else {
                    None
                }
            } else {
                None
            },

            spread_trajectory_shape: if self.compute_advanced && has_data {
                let range = self.spread_high - self.spread_low;
                if range > f64::EPSILON {
                    Some((self.spread_close - self.spread_open) / range)
                } else {
                    None
                }
            } else {
                None
            },

            executable_spread_ratio: if self.compute_advanced && has_two {
                Some(self.adv_executable_count as f64 / n as f64)
            } else {
                None
            },

            spread_autocorrelation: if self.compute_advanced && n >= 3 && self.welford_m2 > f64::EPSILON {
                Some(self.adv_autocorr_sxy / self.welford_m2)
            } else {
                None
            },

            mid_momentum_ratio: if self.compute_advanced && has_data {
                let range = self.adv_mid_high - self.adv_mid_low;
                if range > f64::EPSILON {
                    Some((self.adv_mid_close - self.adv_mid_open) / range)
                } else {
                    None
                }
            } else {
                None
            },

            worst_spread: if self.compute_advanced && has_data && self.welford_mean > f64::EPSILON {
                Some(self.spread_high / self.welford_mean)
            } else {
                None
            },
        }
    }
}