opendeviationbar-streaming 13.70.3

//! MD5 dedup token generation with Python hashlib parity.
//!
//! Issue #318: Rust-native ClickHouse writer — dedup tokens.
//!
//! The dedup token formula matches the Python implementation in
//! `python/opendeviationbar/clickhouse/bulk_operations.py` (line 557-560):
//!
//! ```python
//! key_str = f"{symbol}_{threshold_decimal_bps}_{start_ts}_{end_ts}_{ouroboros_mode}"
//! df["cache_key"] = hashlib.md5(key_str.encode()).hexdigest()
//! ```

use super::row::ClickHouseBarRow;

/// Compute an MD5 dedup token for a single bar's cache key.
///
/// Formula: `MD5("{symbol}_{threshold}_{start_ts}_{end_ts}_{ouroboros_mode}")`
///
/// This MUST produce output byte-identical to Python's `hashlib.md5().hexdigest()`
/// for the same input string.
pub fn compute_dedup_token(
    symbol: &str,
    threshold_decimal_bps: u32,
    start_ts: i64,
    end_ts: i64,
    ouroboros_mode: &str,
) -> String {
    let key_str = format!(
        "{}_{}_{}_{}_{}", symbol, threshold_decimal_bps, start_ts, end_ts, ouroboros_mode
    );
    let digest = md5::compute(key_str.as_bytes());
    format!("{digest:x}")
}

/// Compute an MD5 dedup token for a batch of rows.
///
/// Uses the batch's symbol, threshold, min(open_time_us), max(close_time_us),
/// and "aion" as the ouroboros mode.
///
/// Returns an empty string if the batch is empty.
pub fn compute_batch_dedup_token(rows: &[ClickHouseBarRow]) -> String {
    if rows.is_empty() {
        return String::new();
    }

    let symbol = &rows[0].symbol;
    let threshold = rows[0].threshold_decimal_bps;
    let min_open = rows.iter().map(|r| r.open_time_us).min().unwrap_or(0);
    let max_close = rows.iter().map(|r| r.close_time_us).max().unwrap_or(0);

    compute_dedup_token(symbol, threshold, min_open, max_close, "aion")
}

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

    use crate::live_engine::CompletedBar;

    #[test]
    fn test_dedup_token_parity_with_python() {
        // Python: hashlib.md5('BTCUSDT_250_1700000000000000_1700000100000000_aion'.encode()).hexdigest()
        // Result: 1fd03e7c7957ee24987e9c710f82a7ca
        let token = compute_dedup_token(
            "BTCUSDT",
            250,
            1_700_000_000_000_000,
            1_700_000_100_000_000,
            "aion",
        );
        assert_eq!(
            token, "1fd03e7c7957ee24987e9c710f82a7ca",
            "Dedup token must be byte-identical to Python hashlib.md5"
        );
    }

    #[test]
    fn test_batch_dedup_token() {
        let bar = OpenDeviationBar {
            open_time: 1_700_000_000_000_000,
            close_time: 1_700_000_050_000_000,
            open: FixedPoint(5_000_000_000_000),
            ..Default::default()
        };
        let completed1 = CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: bar.clone(),
        };

        let mut bar2 = bar;
        bar2.open_time = 1_700_000_050_000_001;
        bar2.close_time = 1_700_000_100_000_000;
        let completed2 = CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: bar2,
        };

        let row1 = ClickHouseBarRow::from_completed_bar(&completed1);
        let row2 = ClickHouseBarRow::from_completed_bar(&completed2);
        let rows = vec![row1, row2];

        let batch_token = compute_batch_dedup_token(&rows);

        // Batch token uses min(open_time_us) and max(close_time_us)
        let expected = compute_dedup_token(
            "BTCUSDT",
            250,
            1_700_000_000_000_000,  // min open
            1_700_000_100_000_000,  // max close
            "aion",
        );
        assert_eq!(batch_token, expected);
    }

    #[test]
    fn test_batch_dedup_token_empty() {
        let token = compute_batch_dedup_token(&[]);
        assert!(token.is_empty());
    }
}