opendeviationbar-streaming 13.78.0

//! Pre-write guard validation for completed bars (INTEG-05).
//!
//! Ports Python guards from `bulk_operations.py`:
//! - `_guard_close_time_us_scale()` -> timestamp range check
//! - `_guard_volume_non_negative()` -> volume overflow check
//!
//! Guards run in the fan-out dispatch path (trade_dispatch.rs) BEFORE
//! bars are sent to BarSink implementations. Invalid bars are logged
//! and silently dropped -- they never reach ClickHouse.

use crate::engine::traits::{BarSink, SinkError};
use crate::live_engine::CompletedBar;
use opendeviationbar_core::fixed_point::SCALE;

/// Minimum valid microsecond timestamp (~2001-09-09).
const MIN_US: i64 = 1_000_000_000_000_000;
/// Maximum valid microsecond timestamp (~2049-03-22).
const MAX_US: i64 = 2_500_000_000_000_000;

/// Scale divisor for i128 volume fields (same as row.rs VOLUME_SCALE).
const VOLUME_SCALE: f64 = SCALE as f64;

/// Validate a completed bar for ClickHouse write eligibility.
///
/// Returns `true` if the bar passes all guards, `false` if it should be rejected.
///
/// # Guards
///
/// 1. **Timestamp range (USEC-03)**: `open_time` and `close_time` must be in
///    `[1e15, 2.5e15]` microseconds. Values below 1e15 indicate millisecond or
///    second scale timestamps. Values above 2.5e15 exceed year 2049.
///
/// 2. **Volume non-negative (Issue #88)**: Volume (`i128 as f64 / SCALE`) must
///    be >= 0. Negative volume indicates integer overflow.
pub fn validate_bar_for_write(bar: &CompletedBar) -> bool {
    let b = &bar.bar;

    // Guard 1: close_time scale (USEC-03)
    if b.close_time < MIN_US || b.close_time > MAX_US {
        tracing::error!(
            symbol = %bar.symbol,
            threshold = bar.threshold_decimal_bps,
            close_time_us = b.close_time,
            "pre-write guard: close_time_us out of microsecond range"
        );
        return false;
    }
    if b.open_time < MIN_US || b.open_time > MAX_US {
        tracing::error!(
            symbol = %bar.symbol,
            threshold = bar.threshold_decimal_bps,
            open_time_us = b.open_time,
            "pre-write guard: open_time_us out of microsecond range"
        );
        return false;
    }

    // Guard 2: volume non-negative (Issue #88)
    let vol = b.volume as f64 / VOLUME_SCALE;
    if vol < 0.0 {
        tracing::error!(
            symbol = %bar.symbol,
            threshold = bar.threshold_decimal_bps,
            volume = vol,
            "pre-write guard: negative volume (overflow)"
        );
        return false;
    }

    true
}

/// Dispatch a completed bar to all sinks behind a shared mutex.
///
/// Called from trade_dispatch, puck_fill, and fill_from_rest AFTER the bar
/// has been pushed to the ring buffer. Runs validate_bar_for_write first --
/// invalid bars are silently skipped (already logged by the guard).
///
/// Sink errors are logged but NEVER propagated -- ring buffer delivery to
/// Python is the primary path and must not be affected by sink failures.
pub fn dispatch_to_sinks(
    bar: &CompletedBar,
    sinks: &std::sync::Arc<std::sync::Mutex<Vec<Box<dyn BarSink>>>>,
) {
    if !validate_bar_for_write(bar) {
        return;
    }
    if let Ok(mut locked) = sinks.lock() {
        for sink in locked.iter_mut() {
            match sink.on_bar(bar) {
                Ok(()) => {}
                Err(SinkError::Recoverable(msg)) => {
                    tracing::warn!(
                        sink = sink.name(),
                        symbol = %bar.symbol,
                        threshold = bar.threshold_decimal_bps,
                        %msg,
                        "sink recoverable error"
                    );
                }
                Err(SinkError::Unrecoverable(msg)) => {
                    tracing::error!(
                        sink = sink.name(),
                        symbol = %bar.symbol,
                        threshold = bar.threshold_decimal_bps,
                        %msg,
                        "sink unrecoverable error"
                    );
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::{FixedPoint, OpenDeviationBar, Tick};
    use std::sync::Arc;

    /// Helper: create a CompletedBar with configurable timestamps and volume.
    fn make_completed_bar(open_time: i64, close_time: i64, volume: i128) -> CompletedBar {
        let trade = Tick {
            ref_id: 1,
            price: FixedPoint::from_str("50000.0").unwrap(),
            volume: FixedPoint::from_str("1.0").unwrap(),
            first_sub_id: 1,
            last_sub_id: 1,
            timestamp: 1_700_000_000_000_000, // valid us
            is_buyer_maker: false,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        };
        let mut bar = OpenDeviationBar::new(&trade);
        bar.open_time = open_time;
        bar.close_time = close_time;
        bar.volume = volume;

        CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar,
        }
    }

    #[test]
    fn test_guard_valid_bar() {
        let bar = make_completed_bar(
            1_700_000_000_000_000, // valid us range
            1_700_000_100_000_000,
            5_000_000_000, // 50.0 in f64
        );
        assert!(validate_bar_for_write(&bar));
    }

    #[test]
    fn test_guard_close_time_millisecond_scale() {
        // 1_700_000_000_000 is millisecond scale (below 1e15)
        let bar = make_completed_bar(
            1_700_000_000_000_000,
            1_700_000_000_000, // milliseconds, not microseconds
            5_000_000_000,
        );
        assert!(!validate_bar_for_write(&bar));
    }

    #[test]
    fn test_guard_open_time_beyond_2049() {
        // 3_000_000_000_000_000 exceeds year 2049
        let bar = make_completed_bar(
            3_000_000_000_000_000, // beyond 2049
            1_700_000_100_000_000,
            5_000_000_000,
        );
        assert!(!validate_bar_for_write(&bar));
    }

    #[test]
    fn test_guard_zero_volume_valid() {
        // Zero volume is valid (only negative rejected)
        let bar = make_completed_bar(1_700_000_000_000_000, 1_700_000_100_000_000, 0);
        assert!(validate_bar_for_write(&bar));
    }

    #[test]
    fn test_guard_negative_volume_rejected() {
        // Negative volume indicates overflow
        let bar = make_completed_bar(
            1_700_000_000_000_000,
            1_700_000_100_000_000,
            -1_000_000_000, // negative
        );
        assert!(!validate_bar_for_write(&bar));
    }

    #[test]
    fn test_guard_normal_bar_all_valid() {
        // A completely normal bar with all fields valid
        let bar = make_completed_bar(
            1_700_000_000_000_000,
            1_700_000_100_000_000,
            10_000_000_000, // 100.0 in f64
        );
        assert!(validate_bar_for_write(&bar));
    }
}