opendeviationbar-streaming 13.78.1

//! Type definitions for the live bar engine.

use opendeviationbar_core::OpenDeviationBar;
use opendeviationbar_providers::binance::{AdaptiveRateLimiter, ReconnectionPolicy};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use tokio::sync::watch;

use opendeviationbar_core::checkpoint::Checkpoint;

/// Microseconds per UTC day -- used for midnight boundary detection.
pub(crate) const DAY_US: i64 = 86_400_000_000;

/// Ouroboros reset mode for bar construction.
///
/// Controls whether processors reset at UTC midnight boundaries:
/// - `Day` (default): Reset at midnight, producing orphan bars. Legacy path
///   retained because `#[default]` callers and tests rely on it — no
///   crypto symbol selects this mode in production.
/// - `Aion`: Continuous mode, no midnight reset — bars span across day
///   boundaries (24/7 crypto). Active mode for every symbol in the registry.
/// - `Week`: Gap-based reset, originally added for non-crypto pipelines.
///   Dormant in the crypto-only build (no symbol requests it), but kept as
///   dead code so the extensive test coverage and dispatch paths remain
///   compilable and can be revived by an external consumer if needed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum OuroborosMode {
    /// Day mode: reset at UTC midnight (legacy default).
    #[default]
    Day,
    /// Aion mode: continuous, no midnight reset (24/7 crypto).
    Aion,
    /// Week mode: gap-based reset. `max_gap_us` is the inter-trade gap
    /// threshold in microseconds. Default: 14_400_000_000 (4 hours).
    Week { max_gap_us: i64 },
}

/// A completed bar with metadata identifying its source.
/// Issue #96: `symbol` is `Arc<str>` -- created once per symbol task, cheap clone per bar.
#[derive(Debug, Clone)]
pub struct CompletedBar {
    pub symbol: Arc<str>,
    pub threshold_decimal_bps: u32,
    pub bar: OpenDeviationBar,
}

/// Issue #214: A forming (incomplete) bar snapshot for SSE push to consumers.
///
/// Published at 1Hz via `tokio::sync::watch` channels. Each (symbol, threshold)
/// pair has its own watch channel. The daemon thread reads snapshots without
/// blocking the trade-processing hot path.
#[derive(Debug, Clone)]
pub struct FormingBar {
    pub symbol: Arc<str>,
    pub threshold_decimal_bps: u32,
    pub bar: OpenDeviationBar,
    /// Microsecond timestamp of the last trade that updated this forming bar.
    /// Used by the daemon thread to skip broadcasting stale forming bars
    /// (e.g., when markets are closed and no trades arrive).
    pub last_trade_timestamp_us: i64,
}

/// Key type for forming bar watch channels: (symbol, threshold).
pub(crate) type FormingBarKey = (Arc<str>, u32);

/// Map of watch receivers for forming bar snapshots (Issue #214).
/// Extracted from the engine before `start()` (like `take_checkpoint_receiver()`).
pub type FormingBarWatches = HashMap<FormingBarKey, watch::Receiver<Option<FormingBar>>>;

/// Metrics for the live bar engine.
/// Issue #96 Task #6: Added backpressure metrics for monitoring queue behavior
/// Issue #96 Task #12: Expose ring buffer metrics (dropped bars, queue depth)
#[derive(Debug)]
pub struct LiveEngineMetrics {
    pub trades_received: AtomicU64,
    pub bars_emitted: AtomicU64,
    pub reconnections: AtomicU64,
    pub backpressure_events: AtomicU64, // Times ring buffer was full and bar dropped
    pub dropped_bars: AtomicU64,        // Total bars dropped due to full ring buffer
    pub max_queue_depth: AtomicU64,     // Maximum observed queue depth
    pub total_block_time_ms: AtomicU64, // Accumulated time producer waited for queue space
    pub gap_fills: AtomicU64,           // Number of reconnection gap fills performed
    pub gap_trades_recovered: AtomicU64, // Total trades recovered via REST gap fill
    pub bars_suppressed: AtomicU64,     // Bars suppressed by committed_floors dedup (#345)
    pub trades_skipped_monotonicity: AtomicU64, // #345 MONO-GUARD: WS trades skipped (already processed by REST fill)
}

impl Default for LiveEngineMetrics {
    fn default() -> Self {
        Self {
            trades_received: AtomicU64::new(0),
            bars_emitted: AtomicU64::new(0),
            reconnections: AtomicU64::new(0),
            backpressure_events: AtomicU64::new(0),
            dropped_bars: AtomicU64::new(0),
            max_queue_depth: AtomicU64::new(0),
            total_block_time_ms: AtomicU64::new(0),
            gap_fills: AtomicU64::new(0),
            gap_trades_recovered: AtomicU64::new(0),
            bars_suppressed: AtomicU64::new(0),
            trades_skipped_monotonicity: AtomicU64::new(0),
        }
    }
}

/// WebSocket connection mode for the live bar engine.
///
/// Controls how the sidecar connects to Binance WebSocket streams:
/// - `Combined` (default): 2 combined connections, symbols split by volume for fault isolation
/// - `PerSymbol`: N individual connections, one per symbol (rollback path per D-01)
///
/// Set via `OPENDEVIATIONBAR_STREAMING_WS_MODE` env var ("combined" or "per_symbol").
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum WsMode {
    #[default]
    Combined,
    PerSymbol,
}

/// Split symbols into 2 groups by known Binance spot volume ranking.
///
/// High-volume symbols are assigned to group A, lower-volume to group B,
/// alternating to balance group sizes. Symbols not in the ranking go to
/// the smaller group.
///
/// Returns (group_a, group_b) where group_a contains the first half of
/// ranked symbols and group_b contains the second half.
pub fn split_symbols_by_volume(symbols: &[String]) -> (Vec<String>, Vec<String>) {
    // Hardcoded volume ranking order (top 13 by Binance spot 24h volume, stable)
    const VOLUME_RANKING: &[&str] = &[
        "BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "BNBUSDT", "DOGEUSDT", "ADAUSDT", "AVAXUSDT",
        "LINKUSDT", "TRXUSDT", "SUIUSDT", "LTCUSDT", "NEARUSDT",
    ];

    // Single-connection short-circuit for small N.
    // Binance combined-stream connections support up to 1024 streams each, so
    // any symbol count <= 9 fits comfortably on ONE connection. Spawning two
    // connections for 2-9 symbols doubles the reconnect-attempt rate against
    // the 300 conn/5min per-IP budget for no fault-isolation benefit. The
    // original "2x9 hybrid" design targeted 18 symbols where splitting gave
    // real redundancy; below that threshold, single-connection is safer.
    // Threshold 9 is the boundary: at 10+ symbols, fault-isolation starts
    // mattering more than reconnect-budget conservation.
    if symbols.len() <= 9 {
        return (symbols.to_vec(), Vec::new());
    }

    // Score each symbol: ranked symbols get their rank index, unranked get a high value
    let mut scored: Vec<(usize, &String)> = symbols
        .iter()
        .map(|s| {
            let rank = VOLUME_RANKING
                .iter()
                .position(|&r| r == s.as_str())
                .unwrap_or(VOLUME_RANKING.len() + 1000);
            (rank, s)
        })
        .collect();

    // Sort by rank (high-volume first)
    scored.sort_by_key(|(rank, _)| *rank);

    let half = scored.len().div_ceil(2); // 5 -> 3, 26 -> 13
    let mut group_a = Vec::with_capacity(half);
    let mut group_b = Vec::with_capacity(scored.len() - half);

    for (i, (_, sym)) in scored.into_iter().enumerate() {
        if i < half {
            group_a.push(sym.clone());
        } else {
            group_b.push(sym.clone());
        }
    }

    (group_a, group_b)
}

/// Configuration for the live bar engine.
#[derive(Debug, Clone)]
pub struct LiveEngineConfig {
    /// Per-symbol threshold map (e.g., {"BTCUSDT": [250, 500], "ETHUSDT": [250, 500]}).
    ///
    /// Phase 59 (Bug B fix): Replaces the old flat `symbols: Vec<String>` + `thresholds: Vec<u32>`
    /// pair which produced a Cartesian-product global-union state. The HashMap enforces
    /// per-symbol pairing at the type level -- it is structurally impossible to apply a
    /// wrong thresholds to symbols.
    pub symbol_thresholds: HashMap<String, Vec<u32>>,
    /// Whether to compute inter-bar + intra-bar microstructure features
    pub include_microstructure: bool,
    /// Whether to compute the 3 cheap bar-close features (Petrosian FD, Katz FD,
    /// dispersion entropy). Decoupled from `include_microstructure` (2026-06): these
    /// are flat O(200)-per-bar close-path features, independent of the expensive
    /// per-trade intra-bar microstructure. Gated by
    /// `OPENDEVIATIONBAR_STREAMING_BAR_CLOSE_FEATURES` (default true).
    pub include_bar_close_features: bool,
    /// Channel capacity for completed bars (backpressure)
    pub bar_channel_capacity: usize,
    /// Reconnection policy for WebSocket connections
    pub reconnection_policy: ReconnectionPolicy,
    /// Initial checkpoints keyed by (symbol, threshold) for resuming incomplete bars
    pub initial_checkpoints: HashMap<(String, u32), Checkpoint>,
    /// Optional rate limiter for REST API gap-fill pacing (Issue #162).
    /// When set, replaces fixed 100ms sleep with budget-aware acquire().
    pub rate_limiter: Option<Arc<AdaptiveRateLimiter>>,
    /// Issue #128: Per-feature computation toggles
    pub compute_tier2: bool,
    pub compute_tier3: bool,
    pub compute_hurst: Option<bool>,
    pub compute_permutation_entropy: Option<bool>,
    /// Per-symbol gap detector seeds from ClickHouse (max trade ID across thresholds).
    /// Propagated from StreamManager's `trade_id_state` before `start()`.
    /// Used to seed `TradeIdGapDetector` so the first WS trade triggers gap
    /// detection even when no checkpoint files exist.
    pub gap_detector_seeds: HashMap<String, i64>,
    /// WebSocket connection mode: Combined (2 connections) or PerSymbol (N connections).
    /// Default: Combined. Set via OPENDEVIATIONBAR_STREAMING_WS_MODE env var.
    pub ws_mode: WsMode,
    /// Per-symbol ouroboros mode. Symbols not in this map default to Day.
    /// Used by Phase 16 (Sidecar Aion Startup) to thread per-symbol mode
    /// through start(), fill_from_rest(), symbol_task(), and puck_fill().
    pub symbol_modes: HashMap<String, OuroborosMode>,
    /// WebSocket base URL (e.g., "wss://stream.binance.com:9443").
    /// Injected from registry MarketConfig.ws_base. Defaults to DEFAULT_WS_BASE_URL.
    pub ws_base_url: String,
    /// Extra query params appended to WS URL (e.g., "timeUnit=MICROSECOND").
    /// Injected from registry MarketConfig.ws_params. Defaults to empty string.
    pub ws_params: String,
    /// REST API base URL for fill_from_rest (e.g., "https://api.binance.com/api/v3/aggTrades").
    /// Injected from registry MarketConfig.rest_url. Defaults to Binance Spot.
    pub rest_base_url: String,
    /// Extra HTTP headers for REST requests (e.g., [("X-MBX-TIME-UNIT", "MICROSECOND")]).
    /// Injected from registry MarketConfig.rest_headers. Defaults to empty vec.
    pub rest_headers: Vec<(String, String)>,
}

impl LiveEngineConfig {
    /// Get bar channel capacity from environment or use default (10K)
    /// Issue #96 Task #6: OPENDEVIATIONBAR_MAX_PENDING_BARS env var support
    fn get_bar_channel_capacity() -> usize {
        std::env::var("OPENDEVIATIONBAR_MAX_PENDING_BARS")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .unwrap_or(10_000)
    }

    /// Read the bar-close feature toggle from the environment.
    ///
    /// Decoupled from the microstructure gate (2026-06): the 3 cheap close-path
    /// features (Petrosian FD, Katz FD, dispersion entropy) are governed by their
    /// own flag so they populate live + historical while the expensive per-trade
    /// intra-bar microstructure (VPIN/Kyle/OFI) stays off (06-02 cost scale-down).
    /// Defaults to true (on). Accepts 1/true/yes/on (case-insensitive) to enable,
    /// anything else to disable.
    fn get_include_bar_close_features() -> bool {
        std::env::var("OPENDEVIATIONBAR_STREAMING_BAR_CLOSE_FEATURES")
            .ok()
            .map(|v| Self::parse_bar_close_flag(&v))
            .unwrap_or(true)
    }

    /// Pure parse of the bar-close enable flag (F4: testable without env-var state).
    /// `1`/`true`/`yes`/`on` (case-insensitive, trimmed) enable; anything else disables.
    fn parse_bar_close_flag(raw: &str) -> bool {
        matches!(
            raw.trim().to_ascii_lowercase().as_str(),
            "1" | "true" | "yes" | "on"
        )
    }

    /// Create config with sensible defaults.
    ///
    /// Phase 59 (Bug B fix): Takes a `HashMap<String, Vec<u32>>` to enforce per-symbol
    /// threshold pairing at the type level. The old `new(symbols, thresholds)` signature
    /// that produced a Cartesian product is gone -- there is no migration shim.
    pub fn new(symbol_thresholds: HashMap<String, Vec<u32>>) -> Self {
        Self {
            symbol_thresholds,
            include_microstructure: true,
            include_bar_close_features: Self::get_include_bar_close_features(),
            bar_channel_capacity: Self::get_bar_channel_capacity(),
            reconnection_policy: ReconnectionPolicy::default(),
            initial_checkpoints: HashMap::new(),
            compute_tier2: true,
            compute_tier3: false,
            compute_hurst: None,
            compute_permutation_entropy: None,
            rate_limiter: None,
            gap_detector_seeds: HashMap::new(),
            ws_mode: WsMode::default(),
            symbol_modes: HashMap::new(),
            ws_base_url: opendeviationbar_providers::binance::DEFAULT_WS_BASE_URL.to_string(),
            ws_params: String::new(),
            rest_base_url: "https://api.binance.com/api/v3/aggTrades".to_string(),
            rest_headers: Vec::new(),
        }
    }

    /// Derive symbols list from `symbol_thresholds` keys (sorted, for callers that need a flat list).
    ///
    /// Phase 59: Replaces the removed `symbols: Vec<String>` field. Sorted output provides
    /// deterministic iteration order for logging, tests, and any caller that relied on
    /// the old field being a stable `Vec`.
    pub fn symbols(&self) -> Vec<String> {
        let mut syms: Vec<String> = self.symbol_thresholds.keys().cloned().collect();
        syms.sort();
        syms
    }

    /// Set rate limiter for REST API gap-fill pacing (Issue #162).
    pub fn with_rate_limiter(mut self, limiter: Arc<AdaptiveRateLimiter>) -> Self {
        self.rate_limiter = Some(limiter);
        self
    }

    /// Inject a checkpoint for a specific (symbol, threshold) pair.
    /// Must be called before `LiveBarEngine::start()`.
    pub fn with_checkpoint(
        mut self,
        symbol: String,
        threshold: u32,
        checkpoint: Checkpoint,
    ) -> Self {
        self.initial_checkpoints
            .insert((symbol, threshold), checkpoint);
        self
    }

    /// Set bar channel capacity explicitly (overrides env var)
    pub fn with_bar_channel_capacity(mut self, capacity: usize) -> Self {
        self.bar_channel_capacity = capacity;
        self
    }
}

impl LiveEngineMetrics {
    /// Snapshot of current metrics.
    pub fn snapshot(&self) -> LiveEngineMetricsSnapshot {
        LiveEngineMetricsSnapshot {
            trades_received: self.trades_received.load(Ordering::Relaxed),
            bars_emitted: self.bars_emitted.load(Ordering::Relaxed),
            reconnections: self.reconnections.load(Ordering::Relaxed),
            dropped_bars: self.dropped_bars.load(Ordering::Relaxed),
            max_queue_depth: self.max_queue_depth.load(Ordering::Relaxed),
            backpressure_events: self.backpressure_events.load(Ordering::Relaxed),
            gap_fills: self.gap_fills.load(Ordering::Relaxed),
            gap_trades_recovered: self.gap_trades_recovered.load(Ordering::Relaxed),
            bars_suppressed: self.bars_suppressed.load(Ordering::Relaxed),
            trades_skipped_monotonicity: self.trades_skipped_monotonicity.load(Ordering::Relaxed),
        }
    }

    /// Get ring buffer queue depth estimate (from snapshot timing).
    /// Note: This is approximate due to concurrent updates.
    pub fn estimate_queue_depth(&self) -> u64 {
        let emitted = self.bars_emitted.load(Ordering::Relaxed);
        let dropped = self.dropped_bars.load(Ordering::Relaxed);
        let received = self.trades_received.load(Ordering::Relaxed);
        // Rough estimate: bars emitted + dropped vs trades received
        // This is not exact but gives a sense of queue pressure
        if emitted + dropped > received {
            0
        } else {
            received - emitted - dropped
        }
    }
}

/// Immutable metrics snapshot for reporting.
#[derive(Debug, Clone)]
pub struct LiveEngineMetricsSnapshot {
    pub trades_received: u64,
    pub bars_emitted: u64,
    pub reconnections: u64,
    pub dropped_bars: u64,
    pub max_queue_depth: u64,
    pub backpressure_events: u64,
    pub gap_fills: u64,
    pub gap_trades_recovered: u64,
    pub bars_suppressed: u64,
    pub trades_skipped_monotonicity: u64,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parse_bar_close_flag_truth_table() {
        // F4: the env reader's parse table is pure + tested (no env-var manipulation).
        for on in ["1", "true", "TRUE", " yes ", "On", "YeS", "  on  "] {
            assert!(
                LiveEngineConfig::parse_bar_close_flag(on),
                "{on:?} should enable bar_close"
            );
        }
        for off in ["0", "false", "no", "off", "", "maybe", "2", "true!"] {
            assert!(
                !LiveEngineConfig::parse_bar_close_flag(off),
                "{off:?} should disable bar_close"
            );
        }
    }

    fn sym_thresh(pairs: &[(&str, &[u32])]) -> HashMap<String, Vec<u32>> {
        let mut m = HashMap::new();
        for (sym, thrs) in pairs {
            m.insert((*sym).to_string(), thrs.to_vec());
        }
        m
    }

    #[test]
    fn test_ws_mode_combined_is_default() {
        let config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250]), ("ETHUSDT", &[250])]));
        assert_eq!(config.ws_mode, WsMode::Combined);
    }

    #[test]
    fn test_ws_mode_per_symbol_can_be_set() {
        let mut config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        config.ws_mode = WsMode::PerSymbol;
        assert_eq!(config.ws_mode, WsMode::PerSymbol);
    }

    #[test]
    fn test_split_symbols_by_volume_26_symbols() {
        // 26 symbols should split into 2 groups of 13
        let symbols: Vec<String> = vec![
            "BTCUSDT",
            "ETHUSDT",
            "SOLUSDT",
            "XRPUSDT",
            "BNBUSDT",
            "DOGEUSDT",
            "ADAUSDT",
            "AVAXUSDT",
            "LINKUSDT",
            "TRXUSDT",
            "SUIUSDT",
            "LTCUSDT",
            "NEARUSDT",
            "DOTUSDT",
            "MATICUSDT",
            "SHIBUSDT",
            "BCHUSDT",
            "UNIUSDT",
            "APTUSDT",
            "ICPUSDT",
            "ETCUSDT",
            "FILUSDT",
            "XLMUSDT",
            "ATOMUSDT",
            "VETUSDT",
            "AAVEUSDT",
        ]
        .into_iter()
        .map(String::from)
        .collect();

        let (a, b) = split_symbols_by_volume(&symbols);
        assert_eq!(a.len(), 13);
        assert_eq!(b.len(), 13);
        // High-volume symbols should be in group A
        assert!(a.contains(&"BTCUSDT".to_string()));
        assert!(a.contains(&"ETHUSDT".to_string()));
        assert!(a.contains(&"SOLUSDT".to_string()));
    }

    #[test]
    fn test_split_symbols_by_volume_nine_short_circuits() {
        // 9 symbols should short-circuit to a single connection (N <= 9).
        // Reconnect-budget conservation: two connections for 9 symbols doubles
        // handshake rate vs the 300-conn/5min per-IP budget for zero benefit,
        // since Binance combined streams support 1024 streams per connection.
        let symbols: Vec<String> = vec![
            "BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "BNBUSDT", "DOGEUSDT", "ADAUSDT",
            "AVAXUSDT", "LINKUSDT",
        ]
        .into_iter()
        .map(String::from)
        .collect();

        let (a, b) = split_symbols_by_volume(&symbols);
        assert_eq!(a.len(), 9);
        assert_eq!(b.len(), 0);
    }

    #[test]
    fn test_split_symbols_by_volume_ten_splits() {
        // 10 symbols should split normally (fault isolation beats reconnect
        // conservation at this count).
        let symbols: Vec<String> = vec![
            "BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "BNBUSDT", "DOGEUSDT", "ADAUSDT",
            "AVAXUSDT", "LINKUSDT", "TRXUSDT",
        ]
        .into_iter()
        .map(String::from)
        .collect();

        let (a, b) = split_symbols_by_volume(&symbols);
        assert_eq!(a.len() + b.len(), 10);
        assert!(a.len() >= 1 && b.len() >= 1);
    }

    #[test]
    fn test_split_symbols_by_volume_two_short_circuits() {
        // The production BTCUSDT + ETHUSDT case — must be one connection, not two.
        let symbols: Vec<String> = vec!["BTCUSDT".to_string(), "ETHUSDT".to_string()];
        let (a, b) = split_symbols_by_volume(&symbols);
        assert_eq!(a.len(), 2, "N=2 must use a single connection");
        assert_eq!(b.len(), 0);
    }

    #[test]
    fn test_split_symbols_by_volume_single_symbol() {
        let symbols: Vec<String> = vec!["BTCUSDT".to_string()];
        let (a, b) = split_symbols_by_volume(&symbols);
        assert_eq!(a.len(), 1);
        assert_eq!(b.len(), 0);
    }
}