hypercall/observability/metrics_collector/

market_quality.rs

use super::*;

type BookQuote = hypercall_runtime_api::SnapshotBookQuote;
type NoArbGroupKey = (String, String, char);
type NoArbPoint = (f64, Option<f64>, Option<f64>);

#[derive(Default)]
struct MarketQualitySnapshot {
    total_spread_bps: f64,
    markets_with_spread: i64,
    total_bid_depth: f64,
    total_ask_depth: f64,
    crossed_orderbooks: i64,
}

impl MarketQualitySnapshot {
    fn avg_spread_bps(&self) -> f64 {
        if self.markets_with_spread > 0 {
            self.total_spread_bps / self.markets_with_spread as f64
        } else {
            0.0
        }
    }

    fn record_quote(&mut self, symbol: &str, quote: &BookQuote, snapshot_staleness: Duration) {
        if let (Some(bid), Some(ask)) = (quote.best_bid, quote.best_ask) {
            self.record_spread(symbol, quote, bid, ask, snapshot_staleness);
        }

        if let Some(size) = quote.best_bid_size {
            self.total_bid_depth += size;
        }
        if let Some(size) = quote.best_ask_size {
            self.total_ask_depth += size;
        }
    }

    fn record_spread(
        &mut self,
        symbol: &str,
        quote: &BookQuote,
        bid: f64,
        ask: f64,
        snapshot_staleness: Duration,
    ) {
        if bid <= 0.0 || ask <= 0.0 {
            return;
        }

        if bid >= ask {
            self.crossed_orderbooks += 1;
            log_crossed_orderbook(symbol, quote, bid, ask, snapshot_staleness);
        }

        let mid = (bid + ask) / 2.0;
        let spread_bps = ((ask - bid) / mid) * 10_000.0;
        self.total_spread_bps += spread_bps;
        self.markets_with_spread += 1;

        let underlying = symbol.split('-').next().unwrap_or("UNKNOWN");
        gauge!("ht_spread_bps", "underlying" => underlying.to_string()).set(spread_bps);
    }
}

impl MetricsCollector {
    // ===== Market Quality Metrics =====

    pub(super) async fn collect_market_quality_metrics(&self) {
        let quotes = self.quote_provider.all_quotes();
        let snapshot_staleness = self.quote_provider.staleness();
        let stale_markets: i64 = if snapshot_staleness > Duration::from_secs(60) {
            quotes.len() as i64
        } else {
            0
        };
        let mut snapshot = MarketQualitySnapshot::default();

        for (symbol, quote) in &quotes {
            snapshot.record_quote(symbol, quote, snapshot_staleness);
        }

        let avg_spread = snapshot.avg_spread_bps();
        gauge!("ht_avg_spread_bps").set(avg_spread);
        gauge!("ht_orderbook_bid_depth").set(snapshot.total_bid_depth);
        gauge!("ht_orderbook_ask_depth").set(snapshot.total_ask_depth);
        gauge!("ht_stale_markets").set(stale_markets as f64);
        gauge!("ht_active_markets").set(quotes.len() as f64);

        // CRITICAL INVARIANT: crossed orderbooks should ALWAYS be 0
        gauge!("ht_crossed_orderbooks").set(snapshot.crossed_orderbooks as f64);

        if snapshot.crossed_orderbooks > 0 {
            warn!(
                crossed_count = snapshot.crossed_orderbooks,
                total_markets = quotes.len(),
                "Market quality check found crossed orderbooks - see error logs for details"
            );
        }

        // ===== No-Arbitrage Monotonicity Check =====
        // Call asks must be non-increasing as strike increases.
        // Call bids must be non-increasing as strike increases.
        // Put asks must be non-decreasing as strike increases.
        // Put bids must be non-decreasing as strike increases.
        // Violations indicate free-money arbitrage on the book.
        let arbitrage_violations = Self::check_no_arbitrage_violations(&quotes);

        gauge!("ht_no_arbitrage_violations").set(arbitrage_violations as f64);

        if arbitrage_violations > 0 {
            error!(
                violation_count = arbitrage_violations,
                total_markets = quotes.len(),
                "🚨 NO-ARBITRAGE VIOLATION: option prices violate monotonicity across strikes"
            );
        }

        debug!(
            "Market quality: {:.1} avg spread bps, {} stale markets, {} crossed, {} arb violations (snapshot age={}s)",
            avg_spread,
            stale_markets,
            snapshot.crossed_orderbooks,
            arbitrage_violations,
            snapshot_staleness.as_secs()
        );
    }

    /// Check for no-arbitrage monotonicity violations across the book's best prices.
    ///
    /// Groups instruments by (underlying, expiry, option_type) and verifies:
    /// - Call best asks are non-increasing as strike increases
    /// - Put best asks are non-decreasing as strike increases
    ///
    /// Returns the number of (underlying, expiry, type, side) groups that have violations.
    fn check_no_arbitrage_violations(quotes: &HashMap<String, BookQuote>) -> i64 {
        let mut groups: HashMap<NoArbGroupKey, Vec<NoArbPoint>> = HashMap::new();
        for (symbol, quote) in quotes {
            if let Some((group, point)) = parse_no_arb_point(symbol, quote) {
                groups.entry(group).or_default().push(point);
            }
        }

        groups
            .into_iter()
            .map(|(group, strikes)| count_group_violations(group, strikes))
            .sum()
    }
}

fn log_crossed_orderbook(
    symbol: &str,
    quote: &BookQuote,
    bid: f64,
    ask: f64,
    snapshot_staleness: Duration,
) {
    let bids_preview: Vec<_> = quote.bids.iter().take(5).collect();
    let asks_preview: Vec<_> = quote.asks.iter().take(5).collect();
    error!(
        symbol = %symbol,
        best_bid = %bid,
        best_ask = %ask,
        spread = %(ask - bid),
        snapshot_staleness_secs = snapshot_staleness.as_secs(),
        bid_levels = quote.bids.len(),
        ask_levels = quote.asks.len(),
        bids_top5 = ?bids_preview,
        asks_top5 = ?asks_preview,
        "🚨 CROSSED ORDERBOOK: bid >= ask detected in snapshot"
    );
}

fn parse_no_arb_point(symbol: &str, quote: &BookQuote) -> Option<(NoArbGroupKey, NoArbPoint)> {
    // Parse symbol: "HYPE-20260327-683-C" -> (HYPE, 20260327, 683, C)
    let mut parts = symbol.split('-');
    let underlying = parts.next()?.to_string();
    let expiry = parts.next()?.to_string();
    let strike = parts.next()?.parse().ok()?;
    let opt_type = match parts.next()?.chars().next()? {
        c @ ('C' | 'P') => c,
        _ => return None,
    };
    if parts.next().is_some() {
        return None;
    }

    let bid = quote.best_bid.filter(|&b| b > 0.0);
    let ask = quote.best_ask.filter(|&a| a > 0.0);
    if bid.is_none() && ask.is_none() {
        return None;
    }

    Some(((underlying, expiry, opt_type), (strike, bid, ask)))
}

fn count_group_violations(
    (underlying, expiry, opt_type): NoArbGroupKey,
    mut strikes: Vec<NoArbPoint>,
) -> i64 {
    if strikes.len() < 2 {
        return 0;
    }

    strikes.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal));

    let mut has_ask_violation = false;
    let mut has_bid_violation = false;
    for pair in strikes.windows(2) {
        let (prev_strike, prev_bid, prev_ask) = pair[0];
        let (curr_strike, curr_bid, curr_ask) = pair[1];

        has_ask_violation |= check_side_violation(
            SideCheck {
                previous: prev_ask,
                current: curr_ask,
                previous_strike: prev_strike,
                current_strike: curr_strike,
                opt_type,
                side: "ask",
            },
            &underlying,
            &expiry,
            has_ask_violation,
        );
        has_bid_violation |= check_side_violation(
            SideCheck {
                previous: prev_bid,
                current: curr_bid,
                previous_strike: prev_strike,
                current_strike: curr_strike,
                opt_type,
                side: "bid",
            },
            &underlying,
            &expiry,
            has_bid_violation,
        );
    }

    i64::from(has_ask_violation) + i64::from(has_bid_violation)
}

struct SideCheck {
    previous: Option<f64>,
    current: Option<f64>,
    previous_strike: f64,
    current_strike: f64,
    opt_type: char,
    side: &'static str,
}

fn check_side_violation(
    check: SideCheck,
    underlying: &str,
    expiry: &str,
    already_logged: bool,
) -> bool {
    let Some(previous) = check.previous else {
        return false;
    };
    let Some(current) = check.current else {
        return false;
    };

    let violates = match check.opt_type {
        'C' => current > previous,
        'P' => current < previous,
        _ => false,
    };
    if !violates {
        return false;
    }

    if !already_logged {
        let direction = if check.opt_type == 'C' {
            "increases"
        } else {
            "decreases"
        };
        error!(
            underlying = %underlying,
            expiry = %expiry,
            prev_strike = check.previous_strike,
            curr_strike = check.current_strike,
            previous_price = previous,
            current_price = current,
            side = check.side,
            "No-arbitrage violation: option price {} with strike",
            direction
        );
    }

    true
}