opendeviationbar-streaming 13.78.0

//! MD5 dedup token generation with Python hashlib parity.
//!
//! Issue #318: Rust-native ClickHouse writer — dedup tokens.
//! Issue #465: batch dedup token now includes a content discriminator
//! (row_count + first_tid + last_tid) so two distinct batches that share a
//! min/max time bracket no longer collide and silently drop on INSERT.
//!
//! Two formulas live here:
//!
//! 1. The 5-tuple `compute_dedup_token` — kept for parity with the per-row
//!    `cache_key` column written by Python's `CacheKey.hash_key`
//!    (`python/opendeviationbar/clickhouse/cache.py::CacheKey.hash_key`).
//!    Used as a per-key audit string, **not** as the INSERT dedup token.
//!
//! 2. The 8-tuple `compute_batch_dedup_token` — the value passed to
//!    ClickHouse's `insert_deduplication_token` setting. Mirrors the
//!    Python helper in
//!    `python/opendeviationbar/clickhouse/bulk_operations.py::compute_batch_dedup_token`,
//!    so byte-identical input rows (Rust streaming or Python bulk) produce
//!    byte-identical tokens.

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. Used by Python's `CacheKey.hash_key` for the per-row
/// `cache_key` column, **not** for ClickHouse `insert_deduplication_token`.
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 batch dedup token used as ClickHouse's
/// `insert_deduplication_token`.
///
/// Formula (Issue #465):
///
/// ```text
/// MD5("{symbol}_{threshold}_{min_open}_{max_close}_aion_{row_count}_{first_tid}_{last_tid}")
/// ```
///
/// Where:
///
/// - `min_open` = `min(open_time_us)` across the batch
/// - `max_close` = `max(close_time_us)` across the batch
/// - `row_count` = `rows.len()`
/// - `first_tid` = `min(first_agg_trade_id)` across the batch
/// - `last_tid` = `max(last_agg_trade_id)` across the batch
///
/// The trailing three components (row count + TID bounds) are the content
/// discriminator that prevents the pre-#465 collision where two distinct
/// batches sharing a min/max time bracket produced identical tokens and
/// caused ClickHouse to silently drop the second INSERT.
///
/// Returns an empty string if the batch is empty.
///
/// MUST match Python's
/// `python/opendeviationbar/clickhouse/bulk_operations.py::compute_batch_dedup_token`
/// byte for byte.
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);
    let row_count = rows.len();
    let first_tid = rows.iter().map(|r| r.first_agg_trade_id).min().unwrap_or(0);
    let last_tid = rows.iter().map(|r| r.last_agg_trade_id).max().unwrap_or(0);

    let key_str = format!(
        "{}_{}_{}_{}_{}_{}_{}_{}",
        symbol, threshold, min_open, max_close, "aion", row_count, first_tid, last_tid,
    );
    let digest = md5::compute(key_str.as_bytes());
    format!("{digest:x}")
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::OpenDeviationBar;
    use opendeviationbar_core::fixed_point::FixedPoint;
    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() {
        // Issue #465: batch token formula now includes row_count + first_tid + last_tid.
        let mut bar1 = 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()
        };
        bar1.first_agg_trade_id = 100;
        bar1.last_agg_trade_id = 199;
        let completed1 = CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: bar1.clone(),
        };

        let mut bar2 = bar1;
        bar2.open_time = 1_700_000_050_000_001;
        bar2.close_time = 1_700_000_100_000_000;
        bar2.first_agg_trade_id = 200;
        bar2.last_agg_trade_id = 299;
        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);

        // Recomputed by hand against the 8-tuple format to anchor the formula.
        // key_str = "BTCUSDT_250_1700000000000000_1700000100000000_aion_2_100_299"
        let key_str = "BTCUSDT_250_1700000000000000_1700000100000000_aion_2_100_299";
        let expected = format!("{:x}", md5::compute(key_str.as_bytes()));
        assert_eq!(batch_token, expected);
    }

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

    /// Issue #465 regression: distinct batches sharing the same
    /// `(symbol, threshold, min(open_time_us), max(close_time_us))` but
    /// different content (row count and/or trade-ID range) MUST produce
    /// different tokens. Pre-#465 they collided and the second INSERT was
    /// silently dropped by ClickHouse `insert_deduplication_token`.
    #[test]
    fn test_batch_dedup_token_distinct_content_yields_distinct_tokens() {
        fn make_row(
            open_time: i64,
            close_time: i64,
            first_tid: i64,
            last_tid: i64,
        ) -> ClickHouseBarRow {
            let mut bar = OpenDeviationBar {
                open_time,
                close_time,
                open: FixedPoint(5_000_000_000_000),
                high: FixedPoint(5_000_000_000_000),
                low: FixedPoint(5_000_000_000_000),
                close: FixedPoint(5_000_000_000_000),
                ..Default::default()
            };
            bar.first_agg_trade_id = first_tid;
            bar.last_agg_trade_id = last_tid;
            ClickHouseBarRow::from_completed_bar(&CompletedBar {
                symbol: Arc::from("BTCUSDT"),
                threshold_decimal_bps: 250,
                bar,
            })
        }

        // Both batches span the same min(open_time_us) and max(close_time_us).
        // Batch A: 3 rows, TIDs 1..=300.
        let batch_a = vec![
            make_row(1_000_000, 1_500_000, 1, 100),
            make_row(1_300_000, 1_700_000, 101, 200),
            make_row(1_600_000, 2_000_000, 201, 300),
        ];

        // Batch B: 5 rows, different TID range.
        let batch_b = vec![
            make_row(1_000_000, 1_200_000, 500, 599),
            make_row(1_100_000, 1_400_000, 600, 699),
            make_row(1_400_000, 1_600_000, 700, 799),
            make_row(1_700_000, 1_800_000, 800, 899),
            make_row(1_800_000, 2_000_000, 900, 999),
        ];

        let token_a = compute_batch_dedup_token(&batch_a);
        let token_b = compute_batch_dedup_token(&batch_b);

        assert_ne!(
            token_a, token_b,
            "Issue #465: distinct batches sharing min/max time MUST produce \
             distinct dedup tokens; got identical {token_a}",
        );
    }

    /// Idempotency: the same batch produces the same token across calls
    /// (so transient ClickHouse failures can be retried safely without
    /// regenerating a fresh token).
    #[test]
    fn test_batch_dedup_token_idempotent() {
        let mut 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()
        };
        bar.first_agg_trade_id = 42;
        bar.last_agg_trade_id = 99;
        let row = ClickHouseBarRow::from_completed_bar(&CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar,
        });
        let rows = vec![row];

        assert_eq!(
            compute_batch_dedup_token(&rows),
            compute_batch_dedup_token(&rows),
            "retrying the same batch must yield the same token"
        );
    }
}