hypercall_api/rfq/

indicative_quote_cache.rs

use dashmap::DashMap;
use hypercall_types::WalletAddress;
use rust_decimal::Decimal;
use std::sync::Arc;

/// Entry for a single QP's indicative quote on a single instrument.
/// Prices are the only source of truth — implied vol is derived
/// downstream by a canonical pricer when needed.
#[derive(Debug, Clone)]
pub struct IndicativeQuoteEntry {
    pub instrument: String,
    pub bid_price: Decimal,
    pub ask_price: Decimal,
    pub max_bid_size: Decimal,
    pub max_ask_size: Decimal,
    pub updated_at: u64,
}

/// Aggregated indicative quote across all QPs for a single instrument.
#[derive(Debug, Clone)]
pub struct AggregatedIndicativeQuote {
    pub instrument: String,
    pub best_bid: Decimal,
    pub best_ask: Decimal,
    pub indicative_bid_size: Decimal,
    pub indicative_ask_size: Decimal,
    pub num_providers: u32,
    pub updated_at: u64,
}

/// In-memory cache of indicative quotes from all QPs, with TTL eviction
/// and aggregation into best-bid/best-ask per instrument.
pub struct IndicativeQuoteCache {
    /// Per-QP per-instrument quotes: (qp_wallet, instrument) -> entry
    quotes: DashMap<(WalletAddress, String), IndicativeQuoteEntry>,
    /// Aggregated best quotes per instrument
    aggregates: DashMap<String, AggregatedIndicativeQuote>,
    /// TTL in milliseconds for stale quote eviction
    ttl_ms: u64,
}

impl IndicativeQuoteCache {
    pub fn new(ttl_ms: u64) -> Self {
        Self {
            quotes: DashMap::new(),
            aggregates: DashMap::new(),
            ttl_ms,
        }
    }

    /// Update quotes from a QP with a full snapshot. Upserts every
    /// instrument in `entries` and evicts any previously-seen instrument
    /// for this QP that is NOT present in the new batch, since `rfq-mm`
    /// streams periodic snapshots and silently omits instruments when it
    /// can no longer price them. Without the eviction, the aggregate
    /// would keep serving stale prices until TTL eviction fires much
    /// later, so `/options-summary` could show an old quote long after
    /// the QP stopped supporting that instrument.
    pub fn update(
        &self,
        qp_wallet: WalletAddress,
        entries: Vec<IndicativeQuoteEntry>,
    ) -> Vec<(String, Option<AggregatedIndicativeQuote>)> {
        use std::collections::HashSet;

        let incoming: HashSet<String> = entries.iter().map(|e| e.instrument.clone()).collect();
        let mut affected_instruments: Vec<String> = Vec::new();

        // Evict this QP's previous entries that are absent from the new snapshot.
        self.quotes.retain(|(wallet, instrument), _| {
            if wallet == &qp_wallet && !incoming.contains(instrument) {
                if !affected_instruments.contains(instrument) {
                    affected_instruments.push(instrument.clone());
                }
                false
            } else {
                true
            }
        });

        // Upsert the new snapshot.
        for entry in entries {
            let instrument = entry.instrument.clone();
            self.quotes.insert((qp_wallet, instrument.clone()), entry);
            if !affected_instruments.contains(&instrument) {
                affected_instruments.push(instrument);
            }
        }

        affected_instruments
            .into_iter()
            .map(|instrument| {
                let aggregate = self.recompute_aggregate(&instrument);
                (instrument, aggregate)
            })
            .collect()
    }

    /// Evict all quotes from a disconnected QP.
    pub fn evict_qp(
        &self,
        qp_wallet: &WalletAddress,
    ) -> Vec<(String, Option<AggregatedIndicativeQuote>)> {
        let mut affected_instruments = Vec::new();

        self.quotes.retain(|(wallet, instrument), _| {
            if wallet == qp_wallet {
                if !affected_instruments.contains(instrument) {
                    affected_instruments.push(instrument.clone());
                }
                false
            } else {
                true
            }
        });

        affected_instruments
            .into_iter()
            .map(|instrument| {
                let aggregate = self.recompute_aggregate(&instrument);
                (instrument, aggregate)
            })
            .collect()
    }

    /// Get the aggregated quote for an instrument.
    pub fn get_aggregate(&self, instrument: &str) -> Option<AggregatedIndicativeQuote> {
        self.aggregates.get(instrument).map(|r| r.clone())
    }

    /// Get all aggregated quotes.
    pub fn get_all_aggregates(&self) -> Vec<AggregatedIndicativeQuote> {
        self.aggregates.iter().map(|r| r.value().clone()).collect()
    }

    /// Start a background task that evicts stale quotes.
    pub fn start_ttl_eviction(
        self: &Arc<Self>,
        mut shutdown: tokio::sync::broadcast::Receiver<()>,
    ) {
        let cache = Arc::clone(self);
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
            loop {
                tokio::select! {
                    _ = shutdown.recv() => break,
                    _ = interval.tick() => {
                        cache.evict_stale();
                    }
                }
            }
        });
    }

    /// Evict entries older than TTL.
    fn evict_stale(&self) {
        let now = get_timestamp_millis();
        let mut affected_instruments = Vec::new();

        self.quotes.retain(|(_, instrument), entry| {
            if now.saturating_sub(entry.updated_at) > self.ttl_ms {
                if !affected_instruments.contains(instrument) {
                    affected_instruments.push(instrument.clone());
                }
                false
            } else {
                true
            }
        });

        for instrument in affected_instruments {
            self.recompute_aggregate(&instrument);
        }
    }

    /// Recompute the aggregate for a single instrument from all QP entries.
    fn recompute_aggregate(&self, instrument: &str) -> Option<AggregatedIndicativeQuote> {
        let mut best_bid = Decimal::ZERO;
        let mut best_ask = Decimal::MAX;
        let mut bid_size_at_best = Decimal::ZERO;
        let mut ask_size_at_best = Decimal::ZERO;
        let mut num_providers = 0u32;
        let mut latest_update = 0u64;
        let mut has_entries = false;

        for entry in self.quotes.iter() {
            let ((_, inst), quote) = entry.pair();
            if inst != instrument {
                continue;
            }
            has_entries = true;
            num_providers += 1;

            if quote.updated_at > latest_update {
                latest_update = quote.updated_at;
            }

            if quote.bid_price > best_bid {
                best_bid = quote.bid_price;
                bid_size_at_best = quote.max_bid_size;
            } else if quote.bid_price == best_bid {
                bid_size_at_best += quote.max_bid_size;
            }

            if quote.ask_price < best_ask {
                best_ask = quote.ask_price;
                ask_size_at_best = quote.max_ask_size;
            } else if quote.ask_price == best_ask {
                ask_size_at_best += quote.max_ask_size;
            }
        }

        if !has_entries {
            self.aggregates.remove(instrument);
            return None;
        }

        let aggregate = AggregatedIndicativeQuote {
            instrument: instrument.to_string(),
            best_bid,
            best_ask,
            indicative_bid_size: bid_size_at_best,
            indicative_ask_size: ask_size_at_best,
            num_providers,
            updated_at: latest_update,
        };

        self.aggregates
            .insert(instrument.to_string(), aggregate.clone());
        Some(aggregate)
    }
}

fn get_timestamp_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0)
}

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

    fn test_wallet(id: u8) -> WalletAddress {
        let mut bytes = [0u8; 20];
        bytes[19] = id;
        WalletAddress::from(alloy::primitives::Address::from(bytes))
    }

    fn make_entry(instrument: &str, bid: Decimal, ask: Decimal, ts: u64) -> IndicativeQuoteEntry {
        IndicativeQuoteEntry {
            instrument: instrument.to_string(),
            bid_price: bid,
            ask_price: ask,
            max_bid_size: dec!(10),
            max_ask_size: dec!(10),
            updated_at: ts,
        }
    }

    #[test]
    fn test_single_qp_update() {
        let cache = IndicativeQuoteCache::new(10_000);
        let qp = test_wallet(1);
        let now = get_timestamp_millis();

        cache.update(
            qp,
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );

        let agg = cache.get_aggregate("ETH-C-3000").unwrap();
        assert_eq!(agg.best_bid, dec!(120));
        assert_eq!(agg.best_ask, dec!(125));
        assert_eq!(agg.num_providers, 1);
    }

    #[test]
    fn test_multiple_qps_aggregation() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        // QP1: bid 120, ask 125
        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );
        // QP2: bid 121, ask 124 (better on both sides)
        cache.update(
            test_wallet(2),
            vec![make_entry("ETH-C-3000", dec!(121), dec!(124), now)],
        );

        let agg = cache.get_aggregate("ETH-C-3000").unwrap();
        assert_eq!(agg.best_bid, dec!(121));
        assert_eq!(agg.best_ask, dec!(124));
        assert_eq!(agg.num_providers, 2);
        assert_eq!(agg.indicative_bid_size, dec!(10)); // only QP2 at best
        assert_eq!(agg.indicative_ask_size, dec!(10)); // only QP2 at best
    }

    #[test]
    fn test_sizes_sum_at_same_price() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );
        cache.update(
            test_wallet(2),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );

        let agg = cache.get_aggregate("ETH-C-3000").unwrap();
        assert_eq!(agg.best_bid, dec!(120));
        assert_eq!(agg.indicative_bid_size, dec!(20)); // 10 + 10
        assert_eq!(agg.indicative_ask_size, dec!(20));
    }

    #[test]
    fn test_qp_disconnect_evicts() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );
        cache.update(
            test_wallet(2),
            vec![make_entry("ETH-C-3000", dec!(121), dec!(124), now)],
        );

        // QP2 disconnects
        let updates = cache.evict_qp(&test_wallet(2));

        let agg = cache.get_aggregate("ETH-C-3000").unwrap();
        assert_eq!(updates.len(), 1);
        assert!(updates[0].1.is_some());
        assert_eq!(agg.best_bid, dec!(120)); // falls back to QP1
        assert_eq!(agg.best_ask, dec!(125));
        assert_eq!(agg.num_providers, 1);
    }

    #[test]
    fn test_evict_last_qp_removes_aggregate() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );

        let updates = cache.evict_qp(&test_wallet(1));

        assert_eq!(updates.len(), 1);
        assert!(updates[0].1.is_none());
        assert!(cache.get_aggregate("ETH-C-3000").is_none());
    }

    #[test]
    fn test_empty_snapshot_returns_removal_update() {
        let cache = IndicativeQuoteCache::new(10_000);
        let qp = test_wallet(1);
        let now = get_timestamp_millis();

        cache.update(
            qp,
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), now)],
        );

        let updates = cache.update(qp, vec![]);

        assert_eq!(updates.len(), 1);
        assert_eq!(updates[0].0, "ETH-C-3000");
        assert!(updates[0].1.is_none());
        assert!(cache.get_aggregate("ETH-C-3000").is_none());
    }

    #[test]
    fn test_ttl_eviction() {
        let cache = IndicativeQuoteCache::new(5_000);
        let old_ts = get_timestamp_millis() - 10_000; // 10s ago, beyond 5s TTL

        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(120), dec!(125), old_ts)],
        );

        // Entry exists but is stale
        assert!(cache.get_aggregate("ETH-C-3000").is_some());

        cache.evict_stale();

        assert!(cache.get_aggregate("ETH-C-3000").is_none());
    }

    #[test]
    fn test_update_evicts_symbols_missing_from_new_snapshot() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        // First snapshot: QP1 quotes two instruments.
        cache.update(
            test_wallet(1),
            vec![
                make_entry("ETH-C-3000", dec!(120), dec!(125), now),
                make_entry("ETH-C-3100", dec!(90), dec!(95), now),
            ],
        );
        assert!(cache.get_aggregate("ETH-C-3000").is_some());
        assert!(cache.get_aggregate("ETH-C-3100").is_some());

        // Second snapshot omits ETH-C-3100 (e.g. QP can no longer price it).
        // The old entry must be evicted so the aggregate disappears.
        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(121), dec!(126), now + 1)],
        );
        assert!(cache.get_aggregate("ETH-C-3000").is_some());
        assert!(
            cache.get_aggregate("ETH-C-3100").is_none(),
            "ETH-C-3100 should be evicted when absent from new snapshot"
        );

        // Other QPs' quotes for the evicted instrument must NOT be touched.
        cache.update(
            test_wallet(2),
            vec![make_entry("ETH-C-3100", dec!(88), dec!(93), now + 2)],
        );
        // QP1 sends another snapshot missing ETH-C-3100 — QP2's entry remains.
        cache.update(
            test_wallet(1),
            vec![make_entry("ETH-C-3000", dec!(122), dec!(127), now + 3)],
        );
        assert!(
            cache.get_aggregate("ETH-C-3100").is_some(),
            "QP2's ETH-C-3100 entry must survive QP1's snapshot update"
        );
    }

    #[test]
    fn test_get_all_aggregates() {
        let cache = IndicativeQuoteCache::new(10_000);
        let now = get_timestamp_millis();

        cache.update(
            test_wallet(1),
            vec![
                make_entry("ETH-C-3000", dec!(120), dec!(125), now),
                make_entry("BTC-C-100000", dec!(5000), dec!(5100), now),
            ],
        );

        let all = cache.get_all_aggregates();
        assert_eq!(all.len(), 2);
    }
}