hypercall/observability/metrics_collector/

integrity.rs

use super::*;
use crate::shared::order_types::perp_underlying;
use rust_decimal::Decimal;

fn position_imbalance_alert_threshold() -> Decimal {
    Decimal::new(1, 4)
}

#[derive(Debug, Default, PartialEq)]
struct PositionImbalanceMetrics {
    total_imbalance: Decimal,
    imbalanced_symbols: i64,
    excluded_perp_symbols: i64,
}

fn evaluate_position_imbalance(
    net_positions_by_symbol: &HashMap<String, Decimal>,
) -> PositionImbalanceMetrics {
    let mut metrics = PositionImbalanceMetrics::default();

    for (symbol, net_position) in net_positions_by_symbol {
        if perp_underlying(symbol).is_some() {
            if *net_position != Decimal::ZERO {
                metrics.excluded_perp_symbols += 1;
            }
            continue;
        }

        let abs_net = net_position.abs();
        metrics.total_imbalance += abs_net;
        if abs_net > position_imbalance_alert_threshold() {
            metrics.imbalanced_symbols += 1;
        }
    }

    metrics
}

impl MetricsCollector {
    // ===== Position Integrity Metrics =====
    // Positions live in the in-memory portfolio cache, not the database.

    pub(super) async fn collect_position_metrics(&self) {
        let portfolios = self.portfolio_cache.get_all_portfolios().await;
        let mut net_positions_by_symbol: HashMap<String, Decimal> = HashMap::new();

        for portfolio in portfolios.values() {
            for position in portfolio.positions.values() {
                *net_positions_by_symbol
                    .entry(position.symbol.clone())
                    .or_insert(Decimal::ZERO) += position.amount;
            }
        }

        let metrics = evaluate_position_imbalance(&net_positions_by_symbol);
        match metrics.total_imbalance.to_f64() {
            Some(total_imbalance) => {
                gauge!("ht_integrity_position_imbalance").set(total_imbalance);
                gauge!("ht_integrity_position_imbalance_conversion_failures").set(0.0);
                gauge!("ht_integrity_position_imbalance_complete").set(1.0);
            }
            None => {
                error!(
                    imbalance = %metrics.total_imbalance,
                    "Failed to convert position imbalance to f64"
                );
                gauge!("ht_integrity_position_imbalance_conversion_failures").set(1.0);
                gauge!("ht_integrity_position_imbalance_complete").set(0.0);
            }
        }

        gauge!("ht_integrity_imbalanced_symbols").set(metrics.imbalanced_symbols as f64);
        gauge!("ht_integrity_excluded_perp_symbols").set(metrics.excluded_perp_symbols as f64);

        if metrics.imbalanced_symbols > 0 {
            error!(
                imbalance = %metrics.total_imbalance,
                symbols = metrics.imbalanced_symbols,
                "🚨 POSITION IMBALANCE: matching-engine option positions do not net to zero"
            );
        }
    }

    // ===== Integrity Metrics =====
    // These metrics help detect margin engine bugs by tracking system-wide invariants

    pub(super) async fn collect_integrity_metrics(&self) {
        if let Some(ref integrity_db) = self.integrity_db {
            let results = integrity_db.get_integrity_query_results().await;
            {
                let fill_volume_result = results.fill_volume;
                let settlement_result = results.settlement_stats;
                let oi_result = results.open_interest_by_underlying;

                // NOTE: Position imbalance metrics (ht_integrity_position_imbalance,
                // ht_integrity_imbalanced_symbols) are computed from the in-memory
                // portfolio cache in collect_position_metrics(), not from the DB.
                // Positions are not persisted to the database.

                // Fill volume stats
                if let Ok((fill_count, total_notional)) = fill_volume_result {
                    gauge!("ht_ledger_total_fills").set(fill_count as f64);
                    if let Some(v) = total_notional.to_f64() {
                        gauge!("ht_ledger_total_volume_usd").set(v);
                    } else {
                        error!(
                            "Failed to convert total_notional to f64: {}",
                            total_notional
                        );
                    }
                }

                // Settlement integrity
                if let Ok((total, applied, pending, total_value)) = settlement_result {
                    gauge!("ht_settlement_total").set(total as f64);
                    gauge!("ht_settlement_applied").set(applied as f64);
                    gauge!("ht_settlement_pending").set(pending as f64);
                    if let Some(v) = total_value.to_f64() {
                        gauge!("ht_settlement_total_value_usd").set(v);
                    } else {
                        error!(
                            "Failed to convert settlement total_value to f64: {}",
                            total_value
                        );
                    }

                    // Warn if there are pending settlements that might be stuck
                    if pending > 0 {
                        debug!(
                            "Settlements: {} total, {} applied, {} pending",
                            total, applied, pending
                        );
                    }
                }

                // Orphaned orders on settled instruments (CALL-553 canary)
                if let Ok(count) = results.orphaned_settlement_orders {
                    gauge!("ht_orphaned_settlement_orders").set(count as f64);
                }

                // Settlement events sum from ledger_events (drift detection)
                if let Ok(total) = results.ledger_events_settlement_total {
                    if let Some(v) = total.to_f64() {
                        gauge!("ht_ledger_events_settlement_sum").set(v);
                    } else {
                        error!(
                            "Failed to convert ledger_events_settlement_total to f64: {}",
                            total
                        );
                    }
                }

                // Open interest by underlying
                if let Ok(oi_list) = oi_result {
                    for (underlying, oi) in oi_list {
                        if let Some(v) = oi.to_f64() {
                            gauge!("ht_open_interest_by_underlying", "underlying" => underlying.clone())
                                    .set(v);
                        } else {
                            error!("Failed to convert OI for {} to f64: {}", underlying, oi);
                        }
                    }
                }
            }
        }
    }
}

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

    #[test]
    fn position_imbalance_sums_option_exposure() {
        let net_positions = HashMap::from([
            ("BTC-20260115-100000-C".to_string(), dec!(1.25)),
            ("ETH-20260115-5000-P".to_string(), dec!(-0.5)),
        ]);

        let metrics = evaluate_position_imbalance(&net_positions);

        assert_eq!(metrics.total_imbalance, dec!(1.75));
        assert_eq!(metrics.imbalanced_symbols, 2);
        assert_eq!(metrics.excluded_perp_symbols, 0);
    }

    #[test]
    fn position_imbalance_excludes_external_perps() {
        let net_positions = HashMap::from([
            ("BTC-PERP".to_string(), dec!(7)),
            ("BTC-20260115-100000-C".to_string(), dec!(0)),
        ]);

        let metrics = evaluate_position_imbalance(&net_positions);

        assert_eq!(metrics.total_imbalance, dec!(0));
        assert_eq!(metrics.imbalanced_symbols, 0);
        assert_eq!(metrics.excluded_perp_symbols, 1);
    }

    #[test]
    fn position_imbalance_ignores_dust_for_symbol_count() {
        let net_positions = HashMap::from([("BTC-20260115-100000-C".to_string(), dec!(0.0001))]);

        let metrics = evaluate_position_imbalance(&net_positions);

        assert_eq!(metrics.total_imbalance, dec!(0.0001));
        assert_eq!(metrics.imbalanced_symbols, 0);
    }
}