hypercall_vol_oracle/

databento_oracle.rs

//! Databento-backed vol oracle for CME products (initially GC gold options).
//!
//! This module holds everything Databento-specific behind a single file: the
//! CME symbology parser, expiry calendar, MBP-1 book cache, chain-to-IV
//! aggregator, a mockable feed trait, and the `RiskVolOracle` implementation
//! that publishes surfaces to the rest of the API. Nothing here is exposed
//! to consumers except the final `DatabentoVolOracle` struct.
//!
//! The build-out follows a test-driven sequence; see the cycle markers in
//! the `tests` submodules.

#![allow(dead_code)]

// ---------------------------------------------------------------------------
// Symbology (cycle 3)
// ---------------------------------------------------------------------------

/// CME Globex month codes. Standard across every listed product.
///
/// | Code | Month     | Code | Month     |
/// |------|-----------|------|-----------|
/// | F    | January   | N    | July      |
/// | G    | February  | Q    | August    |
/// | H    | March     | U    | September |
/// | J    | April     | V    | October   |
/// | K    | May       | X    | November  |
/// | M    | June      | Z    | December  |
fn cme_month_code_to_number(code: char) -> Option<u32> {
    match code {
        'F' => Some(1),
        'G' => Some(2),
        'H' => Some(3),
        'J' => Some(4),
        'K' => Some(5),
        'M' => Some(6),
        'N' => Some(7),
        'Q' => Some(8),
        'U' => Some(9),
        'V' => Some(10),
        'X' => Some(11),
        'Z' => Some(12),
        _ => None,
    }
}

/// A parsed CME futures contract symbol.
///
/// The wire format we accept is `<ROOT><MONTH><YEAR>` where ROOT is the
/// two-character product root (`GC` for gold futures), MONTH is one of the
/// CME month codes above, and YEAR is the last digit of the year. Example:
/// `GCM6` = June 2026 gold futures.
#[derive(Debug, Clone, PartialEq, Eq)]
struct CmeFutureSymbol {
    root: String,
    month_code: char,
    year_digit: u8,
}

impl CmeFutureSymbol {
    fn parse(raw: &str) -> Result<Self, String> {
        let trimmed = raw.trim();
        if trimmed.len() != 4 {
            return Err(format!(
                "expected 4-char CME future symbol, got {trimmed:?}"
            ));
        }
        let mut chars = trimmed.chars();
        let root: String = [chars.next().unwrap(), chars.next().unwrap()]
            .iter()
            .collect();
        let month_code = chars.next().unwrap();
        let year_char = chars.next().unwrap();

        if !root.chars().all(|c| c.is_ascii_uppercase()) {
            return Err(format!("future root must be uppercase ASCII: {root:?}"));
        }
        if cme_month_code_to_number(month_code).is_none() {
            return Err(format!("invalid CME month code {month_code:?}"));
        }
        let year_digit = year_char
            .to_digit(10)
            .ok_or_else(|| format!("year digit must be 0-9, got {year_char:?}"))?
            as u8;

        Ok(Self {
            root,
            month_code,
            year_digit,
        })
    }
}

/// Which side of the option — call or put.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CmeOptionSide {
    Call,
    Put,
}

/// A parsed CME option-on-future symbol.
///
/// The wire format we accept is `<ROOT><MONTH><YEAR> <C|P> <STRIKE>`, for
/// example `OGM6 C 2400`. Whitespace is flexible (any run of ASCII space).
#[derive(Debug, Clone, PartialEq)]
struct CmeOptionSymbol {
    root: String,
    month_code: char,
    year_digit: u8,
    side: CmeOptionSide,
    strike: f64,
}

impl CmeOptionSymbol {
    fn parse(raw: &str) -> Result<Self, String> {
        let parts: Vec<&str> = raw.split_ascii_whitespace().collect();
        if parts.len() != 3 {
            return Err(format!(
                "expected '<root><month><year> <C|P> <strike>', got {raw:?}"
            ));
        }
        let (contract, side_str, strike_str) = (parts[0], parts[1], parts[2]);

        if contract.len() != 4 {
            return Err(format!("contract segment must be 4 chars: {contract:?}"));
        }
        let mut chars = contract.chars();
        let root: String = [chars.next().unwrap(), chars.next().unwrap()]
            .iter()
            .collect();
        let month_code = chars.next().unwrap();
        let year_char = chars.next().unwrap();

        if !root.chars().all(|c| c.is_ascii_uppercase()) {
            return Err(format!("option root must be uppercase ASCII: {root:?}"));
        }
        if cme_month_code_to_number(month_code).is_none() {
            return Err(format!("invalid CME month code {month_code:?}"));
        }
        let year_digit = year_char
            .to_digit(10)
            .ok_or_else(|| format!("year digit must be 0-9, got {year_char:?}"))?
            as u8;

        let side = match side_str {
            "C" | "c" => CmeOptionSide::Call,
            "P" | "p" => CmeOptionSide::Put,
            other => return Err(format!("option side must be C or P, got {other:?}")),
        };

        let strike = strike_str
            .parse::<f64>()
            .map_err(|err| format!("invalid strike {strike_str:?}: {err}"))?;
        if !strike.is_finite() || strike <= 0.0 {
            return Err(format!("strike must be positive finite, got {strike}"));
        }

        Ok(Self {
            root,
            month_code,
            year_digit,
            side,
            strike,
        })
    }
}

// ---------------------------------------------------------------------------
// Expiry calendar (cycle 4)
// ---------------------------------------------------------------------------

use chrono::{DateTime, Datelike, NaiveDate, TimeZone, Utc, Weekday};

/// Compute the nominal expiry for a CME OG (gold-on-futures) monthly option
/// contract in the requested option-contract month.
///
/// CME rule: *"Trading terminates at 1:30 p.m. CT on the 4th last business
/// day of the month prior to the option contract month."* This function
/// implements the "4th last business day of the prior month" piece at
/// **daily** resolution and returns UTC midnight of that day.
///
/// # Limitations
///
/// The US-holiday calendar is **not** applied here — we treat Mon–Fri as
/// business days regardless of federal holidays. Near-end-of-month
/// holidays (e.g. Memorial Day on the last Monday of May) can shift the
/// real CME expiry by one calendar day vs. what this function returns.
/// For the nearest Jan–Dec 2026 months none of the federal holidays fall
/// inside the "last 4 business days" window, so the approximation is
/// correct for our current deployment surface; the cycle-4 unit tests
/// pin the specific months we've verified by hand. Production-grade
/// holiday handling is a v2 item.
fn og_monthly_expiry(option_year: i32, option_month: u32) -> Option<DateTime<Utc>> {
    if !(1..=12).contains(&option_month) {
        return None;
    }

    let (prior_year, prior_month) = if option_month == 1 {
        (option_year - 1, 12)
    } else {
        (option_year, option_month - 1)
    };

    let last_day = last_day_of_month(prior_year, prior_month)?;
    let mut date = NaiveDate::from_ymd_opt(prior_year, prior_month, last_day)?;

    let mut business_days_found = 0u32;
    loop {
        if is_weekday(date) {
            business_days_found += 1;
            if business_days_found == 4 {
                break;
            }
        }
        date = date.pred_opt()?;
    }

    Some(Utc.from_utc_datetime(&date.and_hms_opt(0, 0, 0)?))
}

fn last_day_of_month(year: i32, month: u32) -> Option<u32> {
    let (next_year, next_month) = if month == 12 {
        (year + 1, 1)
    } else {
        (year, month + 1)
    };
    NaiveDate::from_ymd_opt(next_year, next_month, 1)
        .and_then(|d| d.pred_opt())
        .map(|d| d.day())
}

fn is_weekday(date: NaiveDate) -> bool {
    !matches!(date.weekday(), Weekday::Sat | Weekday::Sun)
}

#[cfg(test)]
mod expiry_tests {
    use super::*;
    use chrono::TimeZone;

    /// OGM6 (June 2026 gold options): 4th last business day of May 2026.
    /// May 29 Fri (1), May 28 Thu (2), May 27 Wed (3), May 26 Tue (4).
    /// Memorial Day 2026 = May 25, which sits *outside* the final-4 window,
    /// so the no-holiday approximation matches CME's actual calendar here.
    #[test]
    fn ogm6_expires_on_2026_05_26() {
        let expiry = og_monthly_expiry(2026, 6).expect("OGM6 should have a valid expiry");
        let expected = Utc.with_ymd_and_hms(2026, 5, 26, 0, 0, 0).unwrap();
        assert_eq!(expiry, expected, "got {expiry}");
    }

    /// OGU6 (September 2026): Aug 31 Mon (1), Aug 28 Fri (2), Aug 27 Thu (3),
    /// Aug 26 Wed (4). No holidays in the window.
    #[test]
    fn ogu6_expires_on_2026_08_26() {
        let expiry = og_monthly_expiry(2026, 9).expect("OGU6 should have a valid expiry");
        let expected = Utc.with_ymd_and_hms(2026, 8, 26, 0, 0, 0).unwrap();
        assert_eq!(expiry, expected, "got {expiry}");
    }

    /// OGF7 (January 2027): wraps to prior year, Dec 2026.
    /// Dec 31 Thu (1), Dec 30 Wed (2), Dec 29 Tue (3), Dec 28 Mon (4).
    #[test]
    fn ogf7_wraps_to_prior_year_december() {
        let expiry = og_monthly_expiry(2027, 1).expect("OGF7 should have a valid expiry");
        let expected = Utc.with_ymd_and_hms(2026, 12, 28, 0, 0, 0).unwrap();
        assert_eq!(expiry, expected, "got {expiry}");
    }

    /// OGH7 (March 2027): Feb 26 Fri (1), Feb 25 Thu (2), Feb 24 Wed (3),
    /// Feb 23 Tue (4). Presidents Day 2027 (Feb 15) is outside the window.
    #[test]
    fn ogh7_expires_on_2027_02_23() {
        let expiry = og_monthly_expiry(2027, 3).expect("OGH7 should have a valid expiry");
        let expected = Utc.with_ymd_and_hms(2027, 2, 23, 0, 0, 0).unwrap();
        assert_eq!(expiry, expected, "got {expiry}");
    }

    #[test]
    fn invalid_month_returns_none() {
        assert!(og_monthly_expiry(2026, 0).is_none());
        assert!(og_monthly_expiry(2026, 13).is_none());
    }

    #[test]
    fn last_day_of_month_handles_leap_and_year_wrap() {
        assert_eq!(last_day_of_month(2024, 2), Some(29)); // leap
        assert_eq!(last_day_of_month(2025, 2), Some(28));
        assert_eq!(last_day_of_month(2026, 12), Some(31)); // wraps forward
        assert_eq!(last_day_of_month(2026, 4), Some(30));
    }

    #[test]
    fn is_weekday_classifies_all_seven_days() {
        let sunday = NaiveDate::from_ymd_opt(2026, 4, 5).unwrap();
        let monday = NaiveDate::from_ymd_opt(2026, 4, 6).unwrap();
        let friday = NaiveDate::from_ymd_opt(2026, 4, 10).unwrap();
        let saturday = NaiveDate::from_ymd_opt(2026, 4, 11).unwrap();
        assert!(!is_weekday(sunday));
        assert!(is_weekday(monday));
        assert!(is_weekday(friday));
        assert!(!is_weekday(saturday));
    }
}

// ---------------------------------------------------------------------------
// Chain → IV points (cycle 5)
// ---------------------------------------------------------------------------

use hypercall_margin::black_76::black_76_implied_vol;
use hypercall_margin::OptionType;

/// Quality filters applied to raw option quotes before we even try to solve
/// for IV. Tuned defaults aim to reject obvious garbage (one-sided markets,
/// locked books, stale prints) without being so aggressive that we eat the
/// thin wings of a real CME chain.
#[derive(Debug, Clone)]
struct ChainContext {
    /// Front-month futures price, used as the forward F in Black-76.
    forward: f64,
    /// Risk-free rate for discounting. `0.05` is the project-wide static
    /// assumption per earlier agreement.
    risk_free_rate: f64,
    /// "Now" in unix milliseconds. Supplied by the caller so tests are
    /// deterministic.
    now_ms: i64,
    /// Reject quotes whose `(ask − bid) / mid` exceeds this fraction.
    /// A value of `0.20` means "drop anything with > 20% spread".
    max_spread_fraction: f64,
    /// Reject quotes whose `ingested_at_ms` is older than this many
    /// milliseconds relative to `now_ms`.
    max_staleness_ms: i64,
}

impl Default for ChainContext {
    fn default() -> Self {
        Self {
            forward: 0.0,
            risk_free_rate: 0.05,
            now_ms: 0,
            max_spread_fraction: 0.20,
            max_staleness_ms: 60_000,
        }
    }
}

/// A raw option quote as we pull it off Databento. Normalised to the shape
/// the aggregator wants: one strike, one expiry, one side, BBO, and the
/// timestamp we received it at.
#[derive(Debug, Clone)]
struct OptionQuote {
    strike: f64,
    /// Expiry timestamp in unix **seconds** — matches `VolatilitySurface`'s
    /// existing convention.
    expiry_ts: i64,
    side: CmeOptionSide,
    bid: f64,
    ask: f64,
    /// Ingest timestamp in unix **milliseconds**.
    ingested_at_ms: i64,
}

/// A single (strike, expiry, iv) point fit from a quality quote. Ready to
/// insert into a `VolatilitySurface`.
#[derive(Debug, Clone, Copy, PartialEq)]
struct FittedIvPoint {
    strike: f64,
    expiry_ts: i64,
    iv: f64,
}

/// Seconds per year used to convert seconds-to-expiry into Black-76's
/// time-to-expiry parameter (τ). 365.0 matches the existing
/// `calculate_implied_volatility` callers in this repo.
const SECONDS_PER_YEAR: f64 = 365.0 * 86_400.0;

/// Fit an implied-vol point from each quote in the chain that survives
/// quality checks. Filters silently — the count of dropped quotes is
/// observable by comparing input/output lengths if a caller needs it.
fn compute_iv_points(quotes: &[OptionQuote], ctx: &ChainContext) -> Vec<FittedIvPoint> {
    let mut fitted = Vec::with_capacity(quotes.len());

    if !ctx.forward.is_finite() || ctx.forward <= 0.0 {
        return fitted;
    }

    for quote in quotes {
        // Structural BBO checks.
        if quote.bid <= 0.0 || quote.ask <= 0.0 || quote.bid >= quote.ask {
            continue;
        }
        if !quote.bid.is_finite() || !quote.ask.is_finite() {
            continue;
        }

        // Staleness.
        if ctx.now_ms.saturating_sub(quote.ingested_at_ms) > ctx.max_staleness_ms {
            continue;
        }

        // Spread width.
        let mid = 0.5 * (quote.bid + quote.ask);
        let spread = quote.ask - quote.bid;
        if mid <= 0.0 || spread / mid > ctx.max_spread_fraction {
            continue;
        }

        // Time to expiry in years. Drop anything already expired or
        // missing a positive tau.
        let now_secs = ctx.now_ms / 1000;
        let seconds_to_expiry = quote.expiry_ts.saturating_sub(now_secs);
        if seconds_to_expiry <= 0 {
            continue;
        }
        let tau = seconds_to_expiry as f64 / SECONDS_PER_YEAR;
        if !tau.is_finite() || tau <= 0.0 {
            continue;
        }

        let option_type = match quote.side {
            CmeOptionSide::Call => OptionType::Call,
            CmeOptionSide::Put => OptionType::Put,
        };

        // Below-intrinsic and non-convergence are both handled inside
        // the Black-76 wrapper via the underlying BS solver — they
        // come back as None, which we silently drop.
        let Some(iv) = black_76_implied_vol(
            &option_type,
            ctx.forward,
            quote.strike,
            tau,
            ctx.risk_free_rate,
            mid,
        ) else {
            continue;
        };

        if !iv.is_finite() || iv <= 0.0 {
            continue;
        }

        fitted.push(FittedIvPoint {
            strike: quote.strike,
            expiry_ts: quote.expiry_ts,
            iv,
        });
    }

    fitted
}

#[cfg(test)]
mod chain_tests {
    use super::*;
    use hypercall_margin::black_76::black_76_price;

    /// Helper: build a `ChainContext` with the cycle-5 defaults plus a
    /// supplied forward and `now` to keep tests readable.
    fn ctx(forward: f64, now_ms: i64) -> ChainContext {
        ChainContext {
            forward,
            now_ms,
            ..ChainContext::default()
        }
    }

    /// Helper: fabricate an option quote at a known sigma by pricing it
    /// through Black-76 and placing the bid/ask symmetrically around the
    /// resulting mid. The solver should recover sigma to high precision.
    fn quote_at_sigma(
        strike: f64,
        expiry_ts: i64,
        side: CmeOptionSide,
        ctx: &ChainContext,
        sigma: f64,
        half_spread: f64,
        ingested_at_ms: i64,
    ) -> OptionQuote {
        let option_type = match side {
            CmeOptionSide::Call => OptionType::Call,
            CmeOptionSide::Put => OptionType::Put,
        };
        let tau = ((expiry_ts - ctx.now_ms / 1000) as f64) / SECONDS_PER_YEAR;
        let mid = black_76_price(
            &option_type,
            ctx.forward,
            strike,
            tau,
            ctx.risk_free_rate,
            sigma,
        );
        OptionQuote {
            strike,
            expiry_ts,
            side,
            bid: mid - half_spread,
            ask: mid + half_spread,
            ingested_at_ms,
        }
    }

    /// 30 days from now in unix seconds, relative to a supplied `now_ms`.
    fn expiry_30d(now_ms: i64) -> i64 {
        now_ms / 1000 + 30 * 86_400
    }

    #[test]
    fn happy_path_recovers_sigma_for_three_strikes() {
        let now_ms = 1_775_000_000_000_i64;
        let ctx = ctx(4700.0, now_ms);
        let exp = expiry_30d(now_ms);
        let sigma = 0.35;

        let quotes = vec![
            quote_at_sigma(4500.0, exp, CmeOptionSide::Put, &ctx, sigma, 1.0, now_ms),
            quote_at_sigma(4700.0, exp, CmeOptionSide::Call, &ctx, sigma, 1.0, now_ms),
            quote_at_sigma(4900.0, exp, CmeOptionSide::Call, &ctx, sigma, 1.0, now_ms),
        ];

        let points = compute_iv_points(&quotes, &ctx);
        assert_eq!(points.len(), 3, "all 3 clean quotes should fit");
        for p in &points {
            assert!(
                (p.iv - sigma).abs() < 1e-3,
                "fitted iv {} should be close to {sigma}",
                p.iv
            );
        }
    }

    #[test]
    fn drops_zero_or_inverted_bids_and_asks() {
        let now_ms = 1_775_000_000_000i64;
        let ctx = ctx(4700.0, now_ms);
        let exp = expiry_30d(now_ms);

        let mut q = quote_at_sigma(4700.0, exp, CmeOptionSide::Call, &ctx, 0.3, 1.0, now_ms);
        let zero_bid = OptionQuote {
            bid: 0.0,
            ..q.clone()
        };
        let zero_ask = OptionQuote {
            ask: 0.0,
            ..q.clone()
        };
        let inverted = OptionQuote {
            bid: q.ask,
            ask: q.bid,
            ..q.clone()
        };
        // And a clean quote to anchor the test
        q.bid = q.bid.max(0.001);

        let points = compute_iv_points(&[zero_bid, zero_ask, inverted, q], &ctx);
        assert_eq!(points.len(), 1, "only the clean quote should survive");
    }

    #[test]
    fn drops_quotes_with_spread_wider_than_context() {
        let now_ms = 1_775_000_000_000i64;
        let ctx = ctx(4700.0, now_ms); // default max_spread_fraction = 0.20
        let exp = expiry_30d(now_ms);

        // Use a deep-OTM put so the mid is small and the 40-wide spread
        // is clearly > 20% of mid.
        let wide = {
            let mut q = quote_at_sigma(3800.0, exp, CmeOptionSide::Put, &ctx, 0.3, 20.0, now_ms);
            q.bid = 0.01;
            q.ask = 40.0;
            q
        };

        assert!(compute_iv_points(&[wide], &ctx).is_empty());
    }

    #[test]
    fn drops_stale_quotes() {
        let now_ms = 1_775_000_000_000i64;
        let ctx = ctx(4700.0, now_ms);
        let exp = expiry_30d(now_ms);

        // Ingested 2 minutes ago — beyond the 60s default staleness.
        let stale = quote_at_sigma(
            4700.0,
            exp,
            CmeOptionSide::Call,
            &ctx,
            0.3,
            1.0,
            now_ms - 120_000,
        );
        assert!(compute_iv_points(&[stale], &ctx).is_empty());
    }

    #[test]
    fn drops_below_intrinsic_quotes() {
        let now_ms = 1_775_000_000_000i64;
        let ctx = ctx(5000.0, now_ms);
        let exp = expiry_30d(now_ms);

        // A 4000-strike call has intrinsic ≈ 1000 in forward space. Price
        // it at mid = 100 — absurdly below intrinsic, solver must reject.
        let q = OptionQuote {
            strike: 4000.0,
            expiry_ts: exp,
            side: CmeOptionSide::Call,
            bid: 95.0,
            ask: 105.0,
            ingested_at_ms: now_ms,
        };
        assert!(compute_iv_points(&[q], &ctx).is_empty());
    }

    #[test]
    fn drops_already_expired_quotes() {
        let now_ms = 1_775_000_000_000i64;
        let ctx = ctx(4700.0, now_ms);
        let already_expired = now_ms / 1000 - 1;

        let q = OptionQuote {
            strike: 4700.0,
            expiry_ts: already_expired,
            side: CmeOptionSide::Call,
            bid: 50.0,
            ask: 52.0,
            ingested_at_ms: now_ms,
        };
        assert!(compute_iv_points(&[q], &ctx).is_empty());
    }

    #[test]
    fn empty_input_is_not_an_error() {
        let ctx = ctx(4700.0, 1_775_000_000_000);
        assert!(compute_iv_points(&[], &ctx).is_empty());
    }

    #[test]
    fn non_positive_forward_returns_empty() {
        let ctx = ctx(-1.0, 1_775_000_000_000);
        // Even well-formed quotes should drop on bad context.
        let exp = expiry_30d(ctx.now_ms);
        let q = OptionQuote {
            strike: 4700.0,
            expiry_ts: exp,
            side: CmeOptionSide::Call,
            bid: 50.0,
            ask: 52.0,
            ingested_at_ms: ctx.now_ms,
        };
        assert!(compute_iv_points(&[q], &ctx).is_empty());
    }
}

// ---------------------------------------------------------------------------
// MBP-1 book cache (cycle 6)
// ---------------------------------------------------------------------------

use std::collections::HashMap;

/// Databento's canonical 64-bit instrument identifier.
type InstrumentId = u64;

/// A single top-of-book snapshot. Databento MBP-1 records carry exactly
/// this shape (best bid/size + best ask/size + timestamp), so we just
/// store what we're given.
#[derive(Debug, Clone, Copy, PartialEq)]
pub(crate) struct Bbo {
    bid: f64,
    bid_size: u64,
    ask: f64,
    ask_size: u64,
    /// Databento event timestamp, in unix **milliseconds**. Used by the
    /// chain aggregator's staleness check upstream.
    ts_event_ms: i64,
}

/// Per-instrument top-of-book cache. Intentionally dumb: the Databento
/// MBP-1 feed already collapses the full book to a single BBO per update,
/// so there's no order book to maintain here — just the most-recent
/// snapshot per instrument.
///
/// The cache is a pure state machine with no I/O, no locks, and no async.
/// All synchronisation is the caller's problem.
#[derive(Debug, Default)]
struct BookCache {
    books: HashMap<InstrumentId, Bbo>,
}

impl BookCache {
    fn new() -> Self {
        Self::default()
    }

    /// Apply a new BBO snapshot for `instrument_id`. Replaces any prior
    /// entry unconditionally — Databento's MBP-1 updates are already
    /// post-aggregation, so "latest wins" is correct.
    fn apply(&mut self, instrument_id: InstrumentId, bbo: Bbo) {
        self.books.insert(instrument_id, bbo);
    }

    /// Read the current BBO for an instrument, or `None` if we have never
    /// received a quote for it.
    fn get(&self, instrument_id: InstrumentId) -> Option<&Bbo> {
        self.books.get(&instrument_id)
    }

    /// Forget an instrument (e.g. it expired or was removed from the
    /// subscription). Returns `true` if we had it and removed it.
    fn delete(&mut self, instrument_id: InstrumentId) -> bool {
        self.books.remove(&instrument_id).is_some()
    }

    fn len(&self) -> usize {
        self.books.len()
    }

    #[allow(dead_code)]
    fn is_empty(&self) -> bool {
        self.books.is_empty()
    }

    /// Iterate over `(instrument_id, bbo)` pairs for the chain aggregator.
    fn iter(&self) -> impl Iterator<Item = (InstrumentId, &Bbo)> {
        self.books.iter().map(|(id, bbo)| (*id, bbo))
    }
}

#[cfg(test)]
mod book_cache_tests {
    use super::*;

    fn sample_bbo(bid: f64, ask: f64, ts_ms: i64) -> Bbo {
        Bbo {
            bid,
            bid_size: 10,
            ask,
            ask_size: 10,
            ts_event_ms: ts_ms,
        }
    }

    #[test]
    fn apply_then_get_returns_the_bbo() {
        let mut cache = BookCache::new();
        let bbo = sample_bbo(100.0, 101.0, 1_700_000_000_000);
        cache.apply(42, bbo);
        assert_eq!(cache.get(42), Some(&bbo));
        assert_eq!(cache.len(), 1);
    }

    #[test]
    fn apply_replaces_existing_snapshot() {
        let mut cache = BookCache::new();
        cache.apply(42, sample_bbo(100.0, 101.0, 1));
        cache.apply(42, sample_bbo(100.5, 101.5, 2));
        let got = cache.get(42).expect("should still be present");
        assert_eq!(got.bid, 100.5);
        assert_eq!(got.ask, 101.5);
        assert_eq!(got.ts_event_ms, 2);
        assert_eq!(cache.len(), 1, "same instrument, still one entry");
    }

    #[test]
    fn get_on_unknown_instrument_is_none() {
        let cache = BookCache::new();
        assert!(cache.get(999).is_none());
    }

    #[test]
    fn delete_removes_and_reports() {
        let mut cache = BookCache::new();
        cache.apply(1, sample_bbo(1.0, 2.0, 0));
        cache.apply(2, sample_bbo(3.0, 4.0, 0));
        assert!(cache.delete(1));
        assert!(!cache.delete(1)); // second delete: nothing to remove
        assert_eq!(cache.len(), 1);
        assert!(cache.get(1).is_none());
        assert!(cache.get(2).is_some());
    }

    #[test]
    fn iter_yields_all_entries() {
        let mut cache = BookCache::new();
        cache.apply(10, sample_bbo(1.0, 2.0, 100));
        cache.apply(20, sample_bbo(3.0, 4.0, 200));
        cache.apply(30, sample_bbo(5.0, 6.0, 300));

        let mut collected: Vec<(InstrumentId, Bbo)> =
            cache.iter().map(|(id, bbo)| (id, *bbo)).collect();
        collected.sort_by_key(|(id, _)| *id);

        assert_eq!(collected.len(), 3);
        assert_eq!(collected[0].0, 10);
        assert_eq!(collected[1].0, 20);
        assert_eq!(collected[2].0, 30);
    }

    #[test]
    fn new_cache_is_empty() {
        let cache = BookCache::new();
        assert!(cache.is_empty());
        assert_eq!(cache.len(), 0);
    }
}

// ---------------------------------------------------------------------------
// Feed abstraction + fake feed for tests (cycle 7)
// ---------------------------------------------------------------------------

use async_trait::async_trait;

/// A product-type-agnostic view of a single Databento instrument
/// definition, normalised to the fields we actually consume.
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum InstrumentKind {
    Future {
        root: String,
        month_code: char,
        year_digit: u8,
    },
    Option {
        root: String,
        month_code: char,
        year_digit: u8,
        side: CmeOptionSide,
        strike: f64,
        expiry_ts: i64,
    },
}

/// Events we consume from a Databento feed, after the SDK-specific DBN
/// records have been adapted to our internal representation.
///
/// We deliberately do **not** expose Databento's native `dbn::Record`
/// types across this boundary — that would couple our provider to a
/// particular SDK version and make the fake-feed test harness awkward.
/// The real feed adapter (in cycle 8) is responsible for translating DBN
/// records into these four variants.
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum FeedRecord {
    /// An instrument definition update. Fired once per instrument when the
    /// stream starts, and again if a new instrument is listed mid-stream.
    Definition {
        instrument_id: InstrumentId,
        kind: InstrumentKind,
    },
    /// A top-of-book update for a previously-defined instrument.
    Mbp1 {
        instrument_id: InstrumentId,
        bbo: Bbo,
    },
    /// A transient feed error. The provider should treat this as a
    /// reconnect signal — not a hard failure — and continue.
    Error(String),
    /// The feed has closed cleanly (e.g. historical replay done). No
    /// further records will follow.
    EndOfStream,
}

/// Abstraction over a Databento live-or-historical feed, narrowed to the
/// two calls the provider needs: pull the next record, or stop.
///
/// Async because the real implementation wraps `databento::live::Client`
/// which is async; a synchronous `Iterator` would force us to block inside
/// the provider task.
#[async_trait]
pub(crate) trait DatabentoFeed: Send + Sync {
    /// Return the next record from the feed, or `None` once the feed has
    /// ended and will never produce another record. Transient errors are
    /// returned as `Some(FeedRecord::Error(..))` — only a terminal state
    /// yields `None`.
    async fn next_record(&mut self) -> Option<FeedRecord>;
}

/// Deterministic in-memory feed used exclusively in tests. Hand it a
/// `Vec<FeedRecord>` and it replays them in order, one per `next_record`
/// call, then returns `None` forever.
struct FakeFeed {
    records: std::collections::VecDeque<FeedRecord>,
}

impl FakeFeed {
    fn new(records: Vec<FeedRecord>) -> Self {
        Self {
            records: records.into(),
        }
    }
}

#[async_trait]
impl DatabentoFeed for FakeFeed {
    async fn next_record(&mut self) -> Option<FeedRecord> {
        self.records.pop_front()
    }
}

#[cfg(test)]
mod feed_tests {
    use super::*;

    fn sample_gc_definition() -> FeedRecord {
        FeedRecord::Definition {
            instrument_id: 1,
            kind: InstrumentKind::Future {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            },
        }
    }

    fn sample_og_definition() -> FeedRecord {
        FeedRecord::Definition {
            instrument_id: 2,
            kind: InstrumentKind::Option {
                root: "OG".to_string(),
                month_code: 'M',
                year_digit: 6,
                side: CmeOptionSide::Call,
                strike: 4700.0,
                expiry_ts: 1_777_000_000,
            },
        }
    }

    fn sample_bbo_update(instrument_id: InstrumentId) -> FeedRecord {
        FeedRecord::Mbp1 {
            instrument_id,
            bbo: Bbo {
                bid: 100.0,
                bid_size: 5,
                ask: 101.0,
                ask_size: 5,
                ts_event_ms: 1_700_000_000_000,
            },
        }
    }

    #[tokio::test]
    async fn fake_feed_replays_records_in_order() {
        let mut feed = FakeFeed::new(vec![
            sample_gc_definition(),
            sample_og_definition(),
            sample_bbo_update(1),
            sample_bbo_update(2),
        ]);

        let r1 = feed.next_record().await.expect("record 1");
        let r2 = feed.next_record().await.expect("record 2");
        let r3 = feed.next_record().await.expect("record 3");
        let r4 = feed.next_record().await.expect("record 4");
        let r5 = feed.next_record().await;

        assert!(matches!(
            r1,
            FeedRecord::Definition {
                instrument_id: 1,
                ..
            }
        ));
        assert!(matches!(
            r2,
            FeedRecord::Definition {
                instrument_id: 2,
                ..
            }
        ));
        assert!(matches!(
            r3,
            FeedRecord::Mbp1 {
                instrument_id: 1,
                ..
            }
        ));
        assert!(matches!(
            r4,
            FeedRecord::Mbp1 {
                instrument_id: 2,
                ..
            }
        ));
        assert!(r5.is_none(), "feed should be exhausted after 4 records");
    }

    #[tokio::test]
    async fn fake_feed_propagates_error_and_continues() {
        // Errors are transient — the provider sees them, reacts, then
        // keeps consuming records. A terminal feed yields None instead.
        let mut feed = FakeFeed::new(vec![
            FeedRecord::Error("transient: connection reset".to_string()),
            sample_gc_definition(),
            FeedRecord::EndOfStream,
        ]);

        let r1 = feed.next_record().await.expect("error record");
        assert!(matches!(r1, FeedRecord::Error(_)));

        let r2 = feed.next_record().await.expect("definition record");
        assert!(matches!(r2, FeedRecord::Definition { .. }));

        let r3 = feed.next_record().await.expect("end of stream marker");
        assert!(matches!(r3, FeedRecord::EndOfStream));

        let r4 = feed.next_record().await;
        assert!(
            r4.is_none(),
            "after EndOfStream and no more records, feed is exhausted"
        );
    }

    #[tokio::test]
    async fn empty_fake_feed_is_immediately_exhausted() {
        let mut feed = FakeFeed::new(vec![]);
        assert!(feed.next_record().await.is_none());
    }
}

// ---------------------------------------------------------------------------
// Orchestrator (cycle 8)
// ---------------------------------------------------------------------------

use super::vol_surface_cache::VolatilitySurface;

/// Stateful glue that turns a stream of `FeedRecord`s into a live vol
/// surface. Holds the instrument registry, the BBO cache, and the most
/// recent forward we've seen for the underlying futures contract.
///
/// The orchestrator does not own a feed — the oracle (cycle 11) drives
/// the async loop and calls `ingest` for each record. This makes
/// testing trivial: hand-roll a `Vec<FeedRecord>` and ingest them
/// synchronously.
///
/// # Front-month selection
///
/// The forward price used for Black-76 is the most recent GC futures
/// BBO mid we've observed. If multiple GC contracts stream
/// simultaneously (e.g. June + August) the forward will flip-flop
/// between them, which is wrong. In practice the Databento subscription
/// is pinned to the front-month contract only, so this isn't a problem
/// in dev-jake's configuration; a smarter "nearest non-expired contract"
/// rule is a v2 item if we ever subscribe to the full curve.
#[derive(Debug)]
struct DatabentoOrchestrator {
    book_cache: BookCache,
    instruments: HashMap<InstrumentId, InstrumentKind>,
    current_forward: Option<f64>,
    risk_free_rate: f64,
    max_spread_fraction: f64,
    max_staleness_ms: i64,
    /// km:GOLD / CME GC scale factor. 1.0 for dev-jake since km:GOLD
    /// tracks XAU/USD 1:1 per the 2026-04 research. Stored here so
    /// `build_surface` can apply it uniformly on every rebuild.
    strike_scale: f64,
}

impl DatabentoOrchestrator {
    fn new(risk_free_rate: f64, strike_scale: f64) -> Self {
        Self {
            book_cache: BookCache::new(),
            instruments: HashMap::new(),
            current_forward: None,
            risk_free_rate,
            max_spread_fraction: 0.20,
            max_staleness_ms: 60_000,
            strike_scale,
        }
    }

    /// Process one record from the feed. Quiet on transient errors and
    /// end-of-stream — those are the caller's concern.
    fn ingest(&mut self, record: FeedRecord) {
        match record {
            FeedRecord::Definition {
                instrument_id,
                kind,
            } => {
                self.instruments.insert(instrument_id, kind);
            }
            FeedRecord::Mbp1 { instrument_id, bbo } => {
                self.book_cache.apply(instrument_id, bbo);
                if let Some(InstrumentKind::Future { root, .. }) =
                    self.instruments.get(&instrument_id)
                {
                    if root == "GC" {
                        let mid = 0.5 * (bbo.bid + bbo.ask);
                        if mid.is_finite() && mid > 0.0 {
                            self.current_forward = Some(mid);
                        }
                    }
                }
            }
            FeedRecord::Error(_) | FeedRecord::EndOfStream => {}
        }
    }

    /// Build a fresh `VolatilitySurface` from the current cached state.
    /// Returns an empty surface if we don't yet have a forward (no GC
    /// BBO has arrived) — callers can check `.is_empty()` and decide
    /// whether to expose the oracle as ready.
    fn build_surface(&self, now_ms: i64) -> VolatilitySurface {
        let mut surface = VolatilitySurface::new();
        let Some(forward) = self.current_forward else {
            return surface;
        };

        let ctx = ChainContext {
            forward,
            risk_free_rate: self.risk_free_rate,
            now_ms,
            max_spread_fraction: self.max_spread_fraction,
            max_staleness_ms: self.max_staleness_ms,
        };

        // Solve IV in CME space (forward and strike both in CME units), then
        // apply `strike_scale` only at the surface-insert step. Black-76 IV
        // is scale-invariant under uniform (F, K) rescaling — log-moneyness
        // is unchanged — so this produces the correct IV and places the
        // point at a km-space strike that downstream lookups can find.
        let mut quotes = Vec::new();
        for (instrument_id, bbo) in self.book_cache.iter() {
            let Some(kind) = self.instruments.get(&instrument_id) else {
                continue;
            };
            if let InstrumentKind::Option {
                side,
                strike,
                expiry_ts,
                ..
            } = kind
            {
                quotes.push(OptionQuote {
                    strike: *strike,
                    expiry_ts: *expiry_ts,
                    side: *side,
                    bid: bbo.bid,
                    ask: bbo.ask,
                    // Use the surface-build time, not CME's ts_recv. CME
                    // emits MBP1 updates per instrument only when the
                    // BBO changes, so for illiquid OG options the "last
                    // event" can be many minutes old even though the
                    // quote in book_cache is the current best. The 60s
                    // per-quote staleness in compute_iv_points was then
                    // dropping every option — surface stayed at 0 points
                    // with 800k+ MBP1s in flight. Feed-level freshness is
                    // already gated by `state.last_update_ts_ms` against
                    // `config.staleness_threshold` in `status()`, so we
                    // don't lose the dead-feed protection by doing this.
                    ingested_at_ms: now_ms,
                });
            }
        }

        for point in compute_iv_points(&quotes, &ctx) {
            surface.insert(point.strike * self.strike_scale, point.expiry_ts, point.iv);
        }

        // Project onto the no-arb cone in price space. Black-76 call prices
        // equal BS call prices at r=0 with spot replaced by the (already
        // discounted) forward, so we can reuse the shared BS-based sanitizer
        // by passing the scaled forward and rate=0. Scale the forward the
        // same way we scale strikes so both live in the platform's km-space.
        let scaled_forward = forward * self.strike_scale;
        if scaled_forward > 0.0 {
            let clamps = surface.sanitize_arb_free(scaled_forward, 0.0);
            if clamps > 0 {
                // This orchestrator is GOLD-only (keyed off CME GC as the
                // forward root above), so we can hardcode the underlying
                // label on the metric.
                metrics::counter!(
                    "ht_vol_surface_arb_clamps_total",
                    "provider" => "databento",
                    "underlying" => "GOLD"
                )
                .increment(clamps as u64);
            }
        }

        surface
    }

    fn current_forward(&self) -> Option<f64> {
        self.current_forward
    }
}

#[cfg(test)]
mod orchestrator_tests {
    use super::*;
    use hypercall_margin::black_76::black_76_price;

    fn now_ms() -> i64 {
        1_775_000_000_000
    }

    fn expiry_30d() -> i64 {
        now_ms() / 1000 + 30 * 86_400
    }

    /// Helper: build an MBP-1 record for an option with bid/ask generated
    /// from a known sigma, matching the chain_tests helper.
    fn option_mbp1(
        instrument_id: InstrumentId,
        forward: f64,
        strike: f64,
        side: CmeOptionSide,
        sigma: f64,
    ) -> FeedRecord {
        let tau = 30.0 / 365.0;
        let opt_type = match side {
            CmeOptionSide::Call => OptionType::Call,
            CmeOptionSide::Put => OptionType::Put,
        };
        let mid = black_76_price(&opt_type, forward, strike, tau, 0.05, sigma);
        FeedRecord::Mbp1 {
            instrument_id,
            bbo: Bbo {
                bid: mid - 1.0,
                bid_size: 10,
                ask: mid + 1.0,
                ask_size: 10,
                ts_event_ms: now_ms(),
            },
        }
    }

    #[test]
    fn new_orchestrator_has_no_forward_and_empty_surface() {
        let orch = DatabentoOrchestrator::new(0.05, 1.0);
        assert!(orch.current_forward().is_none());
        assert!(orch.build_surface(now_ms()).is_empty());
    }

    #[test]
    fn gc_future_update_populates_forward() {
        let mut orch = DatabentoOrchestrator::new(0.05, 1.0);
        orch.ingest(FeedRecord::Definition {
            instrument_id: 1,
            kind: InstrumentKind::Future {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            },
        });
        orch.ingest(FeedRecord::Mbp1 {
            instrument_id: 1,
            bbo: Bbo {
                bid: 4700.0,
                bid_size: 5,
                ask: 4702.0,
                ask_size: 5,
                ts_event_ms: now_ms(),
            },
        });
        assert_eq!(orch.current_forward(), Some(4701.0));
    }

    #[test]
    fn option_quote_without_forward_yields_empty_surface() {
        let mut orch = DatabentoOrchestrator::new(0.05, 1.0);
        // Define and quote an OG option, but NO GC future — orchestrator
        // has no forward, so build_surface returns empty even though a
        // valid BBO is sitting in the cache.
        orch.ingest(FeedRecord::Definition {
            instrument_id: 10,
            kind: InstrumentKind::Option {
                root: "OG".to_string(),
                month_code: 'M',
                year_digit: 6,
                side: CmeOptionSide::Call,
                strike: 4700.0,
                expiry_ts: expiry_30d(),
            },
        });
        orch.ingest(option_mbp1(
            10,
            4700.0, // "pretend forward" for pricing, not actually known to orch
            4700.0,
            CmeOptionSide::Call,
            0.35,
        ));

        assert!(orch.build_surface(now_ms()).is_empty());
    }

    #[test]
    fn full_pipeline_yields_surface_with_expected_point() {
        let mut orch = DatabentoOrchestrator::new(0.05, 1.0);

        // GC future definition + BBO → populates forward
        orch.ingest(FeedRecord::Definition {
            instrument_id: 1,
            kind: InstrumentKind::Future {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            },
        });
        orch.ingest(FeedRecord::Mbp1 {
            instrument_id: 1,
            bbo: Bbo {
                bid: 4699.5,
                bid_size: 5,
                ask: 4700.5,
                ask_size: 5,
                ts_event_ms: now_ms(),
            },
        });

        // OG option definition + BBO (priced at known sigma = 0.32)
        let exp = expiry_30d();
        orch.ingest(FeedRecord::Definition {
            instrument_id: 10,
            kind: InstrumentKind::Option {
                root: "OG".to_string(),
                month_code: 'M',
                year_digit: 6,
                side: CmeOptionSide::Call,
                strike: 4700.0,
                expiry_ts: exp,
            },
        });
        orch.ingest(option_mbp1(10, 4700.0, 4700.0, CmeOptionSide::Call, 0.32));

        let surface = orch.build_surface(now_ms());
        assert!(!surface.is_empty(), "surface should have the option point");

        let iv = surface
            .get(4700.0, exp)
            .expect("4700@exp should be in the surface");
        assert!(
            (iv - 0.32).abs() < 5e-3,
            "recovered iv {iv} should be near 0.32",
        );
    }

    #[test]
    fn multiple_options_populate_multiple_points() {
        let mut orch = DatabentoOrchestrator::new(0.05, 1.0);

        orch.ingest(FeedRecord::Definition {
            instrument_id: 1,
            kind: InstrumentKind::Future {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            },
        });
        orch.ingest(FeedRecord::Mbp1 {
            instrument_id: 1,
            bbo: Bbo {
                bid: 4699.5,
                bid_size: 5,
                ask: 4700.5,
                ask_size: 5,
                ts_event_ms: now_ms(),
            },
        });

        let exp = expiry_30d();
        for (id, strike, side) in [
            (10, 4500.0, CmeOptionSide::Put),
            (11, 4700.0, CmeOptionSide::Call),
            (12, 4900.0, CmeOptionSide::Call),
        ] {
            orch.ingest(FeedRecord::Definition {
                instrument_id: id,
                kind: InstrumentKind::Option {
                    root: "OG".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                    side,
                    strike,
                    expiry_ts: exp,
                },
            });
            orch.ingest(option_mbp1(id, 4700.0, strike, side, 0.3));
        }

        let surface = orch.build_surface(now_ms());
        assert_eq!(surface.len(), 3);
        assert!(surface.get(4500.0, exp).is_some());
        assert!(surface.get(4700.0, exp).is_some());
        assert!(surface.get(4900.0, exp).is_some());
    }

    #[test]
    fn error_and_end_of_stream_are_noop_on_orchestrator() {
        let mut orch = DatabentoOrchestrator::new(0.05, 1.0);
        orch.ingest(FeedRecord::Error("transient".to_string()));
        orch.ingest(FeedRecord::EndOfStream);
        // Still clean state, still empty surface.
        assert!(orch.current_forward().is_none());
        assert!(orch.build_surface(now_ms()).is_empty());
    }

    #[test]
    fn strike_scale_places_points_in_km_space() {
        // CME GC quotes in ~$2350 range; km:GOLD target space is ~$4700.
        // strike_scale = 2.0 mirrors a hypothetical 2x rescaling. IV is
        // scale-invariant, so the orchestrator should solve in CME
        // space and only rescale strikes at the surface-insert step.
        let mut orch = DatabentoOrchestrator::new(0.05, 2.0);

        orch.ingest(FeedRecord::Definition {
            instrument_id: 1,
            kind: InstrumentKind::Future {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            },
        });
        orch.ingest(FeedRecord::Mbp1 {
            instrument_id: 1,
            bbo: Bbo {
                bid: 2349.5,
                bid_size: 5,
                ask: 2350.5,
                ask_size: 5,
                ts_event_ms: now_ms(),
            },
        });

        let exp = expiry_30d();
        orch.ingest(FeedRecord::Definition {
            instrument_id: 10,
            kind: InstrumentKind::Option {
                root: "OG".to_string(),
                month_code: 'M',
                year_digit: 6,
                side: CmeOptionSide::Call,
                strike: 2350.0,
                expiry_ts: exp,
            },
        });
        // Mid generated against CME forward = 2350, CME strike = 2350.
        orch.ingest(option_mbp1(10, 2350.0, 2350.0, CmeOptionSide::Call, 0.3));

        let surface = orch.build_surface(now_ms());
        let iv = surface
            .get(4700.0, exp)
            .expect("scale=2 should place the point at km-space strike 4700");
        assert!(
            (iv - 0.3).abs() < 5e-3,
            "iv should be scale-invariant: got {iv}, expected 0.3",
        );
        // And the CME-space strike should NOT be in the surface.
        assert!(
            surface.get(2350.0, exp).is_none(),
            "CME-space strike should not exist in a km-space surface",
        );
    }
}

// ---------------------------------------------------------------------------
// DatabentoVolOracle — the public face (cycle 11)
// ---------------------------------------------------------------------------

use super::risk_oracle::{
    RiskVolOracle, VolLookupError, VolOracleStatus, VolProviderKind, VolSurfaceSnapshot,
};
use std::sync::{Arc, RwLock};
use std::time::Duration;

/// Config for the Databento vol oracle.
#[derive(Debug, Clone)]
pub struct DatabentoVolOracleConfig {
    /// Hypercall underlying this oracle serves (e.g. `"GOLD"`). The
    /// oracle rejects `get_iv` calls for any other underlying.
    pub underlying: String,
    /// Risk-free rate used by the Black-76 wrapper. Project default is
    /// 0.05 per the r-rate-research agreement.
    pub risk_free_rate: f64,
    /// Multiplier applied to CME option strikes to land them in the
    /// Hypercall underlying's price space. For km:GOLD this is `1.0`.
    pub strike_scale: f64,
    /// How long before a non-refreshed surface is considered stale.
    pub staleness_threshold: Duration,
}

#[derive(Debug)]
struct DatabentoOracleState {
    orchestrator: DatabentoOrchestrator,
    surface: VolatilitySurface,
    last_update_ts_ms: Option<i64>,
    last_error: Option<String>,
    messages_received: u64,
}

/// A `RiskVolOracle` backed by a Databento feed. Owns a
/// `DatabentoOrchestrator` under an `RwLock` and a cached
/// `VolatilitySurface` that is rebuilt on every ingested record.
///
/// Callers drive the oracle by calling [`ingest`](Self::ingest) for each
/// `FeedRecord`, or by awaiting [`run`](Self::run) with an async
/// `DatabentoFeed`. In production the factory (cycle 12) spawns a task
/// that holds a `DatabentoFeed` over `databento::live::Client` and loops
/// on `run`. In tests, `ingest` is called synchronously with records
/// from `FakeFeed`.
pub struct DatabentoVolOracle {
    config: DatabentoVolOracleConfig,
    state: Arc<RwLock<DatabentoOracleState>>,
}

impl DatabentoVolOracle {
    pub fn new(config: DatabentoVolOracleConfig) -> Self {
        let orchestrator = DatabentoOrchestrator::new(config.risk_free_rate, config.strike_scale);
        let state = DatabentoOracleState {
            orchestrator,
            surface: VolatilitySurface::new(),
            last_update_ts_ms: None,
            last_error: None,
            messages_received: 0,
        };
        Self {
            config,
            state: Arc::new(RwLock::new(state)),
        }
    }

    /// Directly set the `last_error` field without ingesting a record.
    /// Used by the live feed task helper to surface connect-time errors
    /// (which don't flow through the normal `FeedRecord::Error` path
    /// because they happen before the run loop starts).
    pub fn record_external_error(&self, message: impl Into<String>) {
        let mut state = self.state.write().expect("databento oracle state poisoned");
        state.last_error = Some(message.into());
    }

    /// Ingest a single record at "now" = `now_ms`. Tests pass a pinned
    /// timestamp; the async `run` loop uses `Utc::now().timestamp_millis()`.
    fn ingest(&self, record: FeedRecord, now_ms: i64) {
        let mut state = self.state.write().expect("databento oracle state poisoned");
        match &record {
            FeedRecord::Error(msg) => {
                state.last_error = Some(msg.clone());
            }
            FeedRecord::Mbp1 { .. } => {
                state.messages_received += 1;
                state.last_error = None;
            }
            _ => {}
        }
        state.orchestrator.ingest(record);
        state.surface = state.orchestrator.build_surface(now_ms);
        if !state.surface.is_empty() {
            state.last_update_ts_ms = Some(now_ms);
        }
    }

    /// Drive an async feed until it is exhausted. In production this is
    /// spawned on a long-lived tokio task; it returns only when the feed
    /// yields `None` (terminal state). Transient `FeedRecord::Error`
    /// records are recorded on the oracle but do not break the loop.
    pub(crate) async fn run<F: DatabentoFeed>(&self, mut feed: F) {
        let mut record_count: u64 = 0;
        let mut last_log = std::time::Instant::now();
        tracing::info!(
            underlying = %self.config.underlying,
            "Databento oracle run loop starting — waiting for first record"
        );
        while let Some(record) = feed.next_record().await {
            record_count += 1;
            // Log progress every 30s or on the first few records
            if record_count <= 5 || last_log.elapsed().as_secs() >= 30 {
                let state = self.state.read().expect("state");
                tracing::info!(
                    underlying = %self.config.underlying,
                    record_count,
                    messages_received = state.messages_received,
                    has_forward = state.orchestrator.current_forward().is_some(),
                    forward = ?state.orchestrator.current_forward(),
                    surface_points = state.surface.len(),
                    record_type = match &record {
                        FeedRecord::Definition { .. } => "definition",
                        FeedRecord::Mbp1 { .. } => "mbp1",
                        FeedRecord::Error(_) => "error",
                        FeedRecord::EndOfStream => "end_of_stream",
                    },
                    "Databento oracle ingesting record"
                );
                last_log = std::time::Instant::now();
            }
            self.ingest(record, Utc::now().timestamp_millis());
        }
        tracing::warn!(
            underlying = %self.config.underlying,
            record_count,
            "Databento oracle run loop ended (feed exhausted)"
        );
    }

    /// Current oracle health for the single configured underlying.
    fn status(&self) -> VolOracleStatus {
        let state = self.state.read().expect("databento oracle state poisoned");
        let last_update_ts_ms = state.last_update_ts_ms;
        let staleness_seconds = last_update_ts_ms
            .map(|ts| ((Utc::now().timestamp_millis() - ts) as f64 / 1000.0).max(0.0));
        let ready = staleness_seconds
            .map(|age| age <= self.config.staleness_threshold.as_secs_f64())
            .unwrap_or(false)
            && !state.surface.is_empty();
        let connected = state.orchestrator.current_forward().is_some();

        VolOracleStatus {
            underlying: self.config.underlying.clone(),
            provider: VolProviderKind::Databento,
            route_facing: true,
            connected,
            ready,
            last_update_ts_ms,
            staleness_seconds,
            staleness_threshold_seconds: Some(self.config.staleness_threshold.as_secs_f64()),
            surface_points: state.surface.len(),
            messages_received: state.messages_received,
            last_error: state.last_error.clone(),
        }
    }
}

impl RiskVolOracle for DatabentoVolOracle {
    fn get_iv(&self, underlying: &str, strike: f64, expiry_ts: i64) -> Result<f64, VolLookupError> {
        if underlying != self.config.underlying {
            return Err(VolLookupError::UnsupportedUnderlying {
                underlying: underlying.to_string(),
            });
        }

        let status = self.status();
        if !status.connected {
            return Err(VolLookupError::UnhealthyProvider {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Databento,
                reason: status
                    .last_error
                    .unwrap_or_else(|| "no GC forward received yet".to_string()),
            });
        }
        if !status.ready {
            return Err(VolLookupError::StaleSurface {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Databento,
                staleness_seconds: status.staleness_seconds.unwrap_or(f64::INFINITY),
                threshold_seconds: self.config.staleness_threshold.as_secs_f64(),
            });
        }

        let state = self.state.read().expect("databento oracle state poisoned");
        state
            .surface
            .get_interpolated(strike, expiry_ts)
            .ok_or_else(|| VolLookupError::MissingSurface {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Databento,
                strike,
                expiry_ts,
            })
    }

    fn statuses(&self) -> Vec<VolOracleStatus> {
        vec![self.status()]
    }

    /// Expose the live surface so `/monitoring/vol-surface` and
    /// `ServerVolProvider` (the MM's client) can read it. Without this
    /// override the trait default returned `None` and the MM never saw
    /// databento-sourced GOLD quotes even while the oracle held them —
    /// Prometheus showed `ht_vol_surface_points{databento,GOLD} = 1896`
    /// while the endpoint returned `surfaces: []`.
    fn get_surface_snapshot(&self, underlying: &str) -> Option<VolSurfaceSnapshot> {
        if underlying != self.config.underlying {
            return None;
        }
        let state = self.state.read().expect("databento oracle state poisoned");
        if state.surface.is_empty() {
            return None;
        }
        Some(VolSurfaceSnapshot {
            underlying: underlying.to_string(),
            last_update_ts_ms: state.last_update_ts_ms,
            expiries: state.surface.expiries().iter().copied().collect(),
            strike_points: state.surface.export_all_points(),
            delta_curves: state.surface.export_delta_curves(),
            atm_vols: state.surface.export_atm_vols(),
            spot_price: state.orchestrator.current_forward(),
        })
    }

    fn supports_surface_snapshots(&self) -> bool {
        true
    }
}

#[cfg(test)]
mod oracle_tests {
    use super::*;
    use hypercall_margin::black_76::black_76_price;

    fn now_ms() -> i64 {
        1_775_000_000_000
    }

    fn expiry_30d() -> i64 {
        now_ms() / 1000 + 30 * 86_400
    }

    fn config_for_gold() -> DatabentoVolOracleConfig {
        DatabentoVolOracleConfig {
            underlying: "GOLD".to_string(),
            risk_free_rate: 0.05,
            strike_scale: 1.0,
            // 10 minutes — wide enough that ingested records aren't
            // immediately stale under the Utc::now() staleness check.
            staleness_threshold: Duration::from_secs(60 * 60 * 24 * 365 * 100),
        }
    }

    fn option_mbp1(
        instrument_id: InstrumentId,
        forward: f64,
        strike: f64,
        sigma: f64,
    ) -> FeedRecord {
        let tau = 30.0 / 365.0;
        let mid = black_76_price(&OptionType::Call, forward, strike, tau, 0.05, sigma);
        FeedRecord::Mbp1 {
            instrument_id,
            bbo: Bbo {
                bid: mid - 1.0,
                bid_size: 10,
                ask: mid + 1.0,
                ask_size: 10,
                ts_event_ms: now_ms(),
            },
        }
    }

    #[test]
    fn brand_new_oracle_reports_disconnected_and_get_iv_errors() {
        let oracle = DatabentoVolOracle::new(config_for_gold());
        let status = oracle.statuses();
        assert_eq!(status.len(), 1);
        assert!(!status[0].connected);
        assert!(!status[0].ready);
        assert_eq!(status[0].surface_points, 0);

        let err = oracle
            .get_iv("GOLD", 4700.0, expiry_30d())
            .expect_err("should fail without any data");
        assert!(matches!(err, VolLookupError::UnhealthyProvider { .. }));
    }

    #[test]
    fn ingesting_gc_only_is_connected_but_not_ready() {
        let oracle = DatabentoVolOracle::new(config_for_gold());
        oracle.ingest(
            FeedRecord::Definition {
                instrument_id: 1,
                kind: InstrumentKind::Future {
                    root: "GC".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                },
            },
            now_ms(),
        );
        oracle.ingest(
            FeedRecord::Mbp1 {
                instrument_id: 1,
                bbo: Bbo {
                    bid: 4699.5,
                    bid_size: 5,
                    ask: 4700.5,
                    ask_size: 5,
                    ts_event_ms: now_ms(),
                },
            },
            now_ms(),
        );

        let status = oracle.statuses();
        assert!(status[0].connected, "forward arrived -> connected");
        assert!(!status[0].ready, "no option points yet -> not ready");
    }

    #[test]
    fn full_ingest_yields_queryable_surface_and_ready_status() {
        let oracle = DatabentoVolOracle::new(config_for_gold());
        // Future
        oracle.ingest(
            FeedRecord::Definition {
                instrument_id: 1,
                kind: InstrumentKind::Future {
                    root: "GC".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                },
            },
            now_ms(),
        );
        oracle.ingest(
            FeedRecord::Mbp1 {
                instrument_id: 1,
                bbo: Bbo {
                    bid: 4699.5,
                    bid_size: 5,
                    ask: 4700.5,
                    ask_size: 5,
                    ts_event_ms: now_ms(),
                },
            },
            now_ms(),
        );
        // Option
        let exp = expiry_30d();
        oracle.ingest(
            FeedRecord::Definition {
                instrument_id: 10,
                kind: InstrumentKind::Option {
                    root: "OG".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                    side: CmeOptionSide::Call,
                    strike: 4700.0,
                    expiry_ts: exp,
                },
            },
            now_ms(),
        );
        oracle.ingest(option_mbp1(10, 4700.0, 4700.0, 0.31), now_ms());

        let status = oracle.statuses();
        assert!(status[0].connected);
        assert!(status[0].ready);
        assert_eq!(status[0].surface_points, 1);
        assert!(status[0].messages_received >= 1);

        let iv = oracle
            .get_iv("GOLD", 4700.0, exp)
            .expect("populated surface should lookup cleanly");
        assert!((iv - 0.31).abs() < 5e-3, "got {iv}");
    }

    #[test]
    fn unsupported_underlying_rejects() {
        let oracle = DatabentoVolOracle::new(config_for_gold());
        let err = oracle
            .get_iv("BTC", 100_000.0, expiry_30d())
            .expect_err("BTC is not our underlying");
        assert!(matches!(err, VolLookupError::UnsupportedUnderlying { .. }));
    }

    #[test]
    fn staleness_threshold_gates_readiness() {
        // Tight staleness: 10ms. Even a fresh ingest should flip to stale
        // almost immediately once the Utc::now() status check runs.
        let mut config = config_for_gold();
        config.staleness_threshold = Duration::from_millis(10);
        let oracle = DatabentoVolOracle::new(config);

        // Seed with a far-past timestamp so staleness_seconds is huge.
        let ancient = 100_000_000i64;
        oracle.ingest(
            FeedRecord::Definition {
                instrument_id: 1,
                kind: InstrumentKind::Future {
                    root: "GC".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                },
            },
            ancient,
        );
        oracle.ingest(
            FeedRecord::Mbp1 {
                instrument_id: 1,
                bbo: Bbo {
                    bid: 4699.5,
                    bid_size: 5,
                    ask: 4700.5,
                    ask_size: 5,
                    ts_event_ms: ancient,
                },
            },
            ancient,
        );
        let exp = expiry_30d();
        oracle.ingest(
            FeedRecord::Definition {
                instrument_id: 10,
                kind: InstrumentKind::Option {
                    root: "OG".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                    side: CmeOptionSide::Call,
                    strike: 4700.0,
                    expiry_ts: exp,
                },
            },
            ancient,
        );
        oracle.ingest(option_mbp1(10, 4700.0, 4700.0, 0.3), ancient);

        let err = oracle
            .get_iv("GOLD", 4700.0, exp)
            .expect_err("ancient surface should be stale");
        assert!(matches!(err, VolLookupError::StaleSurface { .. }));
    }

    #[test]
    fn error_records_track_last_error_field() {
        let oracle = DatabentoVolOracle::new(config_for_gold());
        oracle.ingest(
            FeedRecord::Error("connection reset by peer".to_string()),
            now_ms(),
        );
        let status = oracle.statuses();
        assert_eq!(
            status[0].last_error.as_deref(),
            Some("connection reset by peer")
        );
    }

    #[tokio::test]
    async fn run_drains_a_fake_feed() {
        // Unlike the synchronous ingest tests (which use a frozen now_ms
        // so the chain's staleness check always passes), `run` uses the
        // real wall clock. Every record's ts_event_ms and the option
        // expiry must be anchored to real-now, otherwise the chain
        // aggregator drops everything as stale.
        use chrono::Utc;
        let oracle = DatabentoVolOracle::new(config_for_gold());
        let real_now_ms = Utc::now().timestamp_millis();
        let real_exp = Utc::now().timestamp() + 30 * 86_400;

        let forward = 4700.0;
        let strike = 4700.0;
        let sigma = 0.33;
        let tau = 30.0 / 365.0;
        let mid = black_76_price(&OptionType::Call, forward, strike, tau, 0.05, sigma);

        let feed = FakeFeed::new(vec![
            FeedRecord::Definition {
                instrument_id: 1,
                kind: InstrumentKind::Future {
                    root: "GC".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                },
            },
            FeedRecord::Mbp1 {
                instrument_id: 1,
                bbo: Bbo {
                    bid: forward - 0.5,
                    bid_size: 5,
                    ask: forward + 0.5,
                    ask_size: 5,
                    ts_event_ms: real_now_ms,
                },
            },
            FeedRecord::Definition {
                instrument_id: 10,
                kind: InstrumentKind::Option {
                    root: "OG".to_string(),
                    month_code: 'M',
                    year_digit: 6,
                    side: CmeOptionSide::Call,
                    strike,
                    expiry_ts: real_exp,
                },
            },
            FeedRecord::Mbp1 {
                instrument_id: 10,
                bbo: Bbo {
                    bid: mid - 1.0,
                    bid_size: 5,
                    ask: mid + 1.0,
                    ask_size: 5,
                    ts_event_ms: real_now_ms,
                },
            },
            FeedRecord::EndOfStream,
        ]);

        oracle.run(feed).await;

        let status = oracle.statuses();
        assert!(
            status[0].ready,
            "feed should have populated a ready surface"
        );
        assert_eq!(status[0].surface_points, 1);
    }
}

// ---------------------------------------------------------------------------
// Real Databento live feed adapter (cycle 15)
// ---------------------------------------------------------------------------

use databento::dbn::{InstrumentClass, InstrumentDefMsg, Mbp1Msg, SType, Schema};
use databento::live::Subscription as DatabentoSubscription;
use databento::{LiveClient, Symbols};
use time::{Duration as TimeDuration, OffsetDateTime};

/// How far back to replay MBP1 on connect. Only used to re-seed `book_cache`
/// on pod restart — 5 minutes is plenty, the live tail does the rest.
const DATABENTO_MBP1_REPLAY_MINUTES: i64 = 5;

/// How long to wait (idle seconds) before we consider the Definition
/// snapshot drained and shut that client down. Local reproduction shows
/// the full `GLBX.MDP3` GC+OG directory (43 730 records) drains in
/// ~5s — a 3s idle window is a comfortable cutoff.
const DATABENTO_DEFINITION_DRAIN_IDLE_SECS: u64 = 3;

/// Adapter that wraps Databento Live behind our `DatabentoFeed` trait.
///
/// Databento's Live gateway deadlocks the MBP1 stream when a single
/// session carries a Definition subscription with `start = UNIX_EPOCH`
/// (full snapshot) alongside an MBP1 subscription with any other start —
/// verified both on staging (no MBP1 records after the 43k def snapshot)
/// and in a local harness. Splitting into two independent Live clients
/// on separate sessions avoids the deadlock: the Definition client
/// drains the snapshot and is dropped, and the MBP1 client streams
/// live quotes with a short 5-minute replay to re-seed the book.
///
/// The `DatabentoFeed::next_record` surface is preserved: both reader
/// tasks feed a shared unbounded mpsc and `next_record` just receives
/// from it.
pub struct DatabentoLiveFeed {
    rx: tokio::sync::mpsc::UnboundedReceiver<FeedRecord>,
    // Tokio JoinHandles don't abort on drop, but the reader tasks will
    // exit cleanly when their `UnboundedSender` drops (i.e. when this
    // struct and its `rx` are dropped and they try to send). For a
    // process-lifetime oracle that's good enough.
    _def_task: tokio::task::JoinHandle<()>,
    _mbp1_task: tokio::task::JoinHandle<()>,
}

impl DatabentoLiveFeed {
    /// Connect to Databento Live using two independent sessions:
    /// one for the Definition snapshot (`start = EPOCH`), one for
    /// live MBP1 quotes (`start = now − 5 min` for a short re-seed).
    pub async fn connect(api_key: &str, dataset: &str) -> anyhow::Result<Self> {
        let symbols = Symbols::Symbols(vec!["GC.FUT".to_string(), "OG.OPT".to_string()]);

        // --- Session 1: Definition snapshot ---
        let mut def_client = LiveClient::builder()
            .key(api_key)?
            .dataset(dataset.to_string())
            .build()
            .await?;
        def_client
            .subscribe(
                DatabentoSubscription::builder()
                    .symbols(symbols.clone())
                    .schema(Schema::Definition)
                    .stype_in(SType::Parent)
                    .start(OffsetDateTime::UNIX_EPOCH)
                    .build(),
            )
            .await?;
        def_client.start().await?;

        // --- Session 2: live MBP1 with short replay ---
        let mbp1_start =
            OffsetDateTime::now_utc() - TimeDuration::minutes(DATABENTO_MBP1_REPLAY_MINUTES);
        let mut mbp1_client = LiveClient::builder()
            .key(api_key)?
            .dataset(dataset.to_string())
            .build()
            .await?;
        mbp1_client
            .subscribe(
                DatabentoSubscription::builder()
                    .symbols(symbols)
                    .schema(Schema::Mbp1)
                    .stype_in(SType::Parent)
                    .start(mbp1_start)
                    .build(),
            )
            .await?;
        mbp1_client.start().await?;

        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

        // Def reader: drain the snapshot until the stream goes idle,
        // then exit (and drop the client). Live definitions that arrive
        // after the initial snapshot are rare and would require
        // long-lived session management to forward cleanly.
        let def_tx = tx.clone();
        let def_task = tokio::spawn(Self::run_definition_reader(def_client, def_tx));

        // MBP1 reader: pump records into the channel forever.
        let mbp1_tx = tx;
        let mbp1_task = tokio::spawn(Self::run_mbp1_reader(mbp1_client, mbp1_tx));

        Ok(Self {
            rx,
            _def_task: def_task,
            _mbp1_task: mbp1_task,
        })
    }

    /// Drive the Definition-only client until it goes idle, converting
    /// each matching `InstrumentDefMsg` into `FeedRecord::Definition`.
    async fn run_definition_reader(
        mut client: LiveClient,
        tx: tokio::sync::mpsc::UnboundedSender<FeedRecord>,
    ) {
        let idle = std::time::Duration::from_secs(DATABENTO_DEFINITION_DRAIN_IDLE_SECS);
        let poll = std::time::Duration::from_millis(500);
        let mut seen = 0u64;
        let mut sent = 0u64;
        let mut last_activity = std::time::Instant::now();
        loop {
            match tokio::time::timeout(poll, client.next_record()).await {
                Err(_) => {
                    // Idle tick. Only treat quiet as "snapshot done" after
                    // we've received something — otherwise we'd bail before
                    // the first record on a slow-start session.
                    if seen > 0 && last_activity.elapsed() > idle {
                        tracing::info!(
                            defs_seen = seen,
                            defs_sent = sent,
                            "Databento definition snapshot drained"
                        );
                        break;
                    }
                }
                Ok(Ok(Some(rec))) => {
                    seen += 1;
                    last_activity = std::time::Instant::now();
                    if let Some(def) = rec.get::<InstrumentDefMsg>() {
                        if let Some(kind) = Self::definition_to_kind(def) {
                            sent += 1;
                            if tx
                                .send(FeedRecord::Definition {
                                    instrument_id: def.hd.instrument_id as InstrumentId,
                                    kind,
                                })
                                .is_err()
                            {
                                tracing::info!(
                                    "Databento def receiver dropped; stopping def reader"
                                );
                                return;
                            }
                        }
                    }
                }
                Ok(Ok(None)) => {
                    tracing::info!(defs_seen = seen, "Databento def stream ended");
                    break;
                }
                Ok(Err(e)) => {
                    tracing::error!(error = %e, "Databento def client error");
                    let _ = tx.send(FeedRecord::Error(format!("databento def: {e}")));
                    break;
                }
            }
        }
    }

    /// Drive the MBP1 client forever, converting matching `Mbp1Msg` into
    /// `FeedRecord::Mbp1` and forwarding them through the channel.
    async fn run_mbp1_reader(
        mut client: LiveClient,
        tx: tokio::sync::mpsc::UnboundedSender<FeedRecord>,
    ) {
        loop {
            match client.next_record().await {
                Ok(Some(rec)) => {
                    if let Some(msg) = rec.get::<Mbp1Msg>() {
                        let Some(bbo) = Self::mbp1_to_bbo(msg) else {
                            continue;
                        };
                        if tx
                            .send(FeedRecord::Mbp1 {
                                instrument_id: msg.hd.instrument_id as InstrumentId,
                                bbo,
                            })
                            .is_err()
                        {
                            tracing::info!("Databento mbp1 receiver dropped; stopping mbp1 reader");
                            return;
                        }
                    }
                }
                Ok(None) => {
                    tracing::warn!("Databento mbp1 stream ended");
                    let _ = tx.send(FeedRecord::EndOfStream);
                    return;
                }
                Err(e) => {
                    tracing::error!(error = %e, "Databento mbp1 client error");
                    let _ = tx.send(FeedRecord::Error(format!("databento mbp1: {e}")));
                    return;
                }
            }
        }
    }

    /// Translate a Databento `InstrumentDefMsg` into our internal
    /// `InstrumentKind`, or return `None` for instruments we don't
    /// care about (non-GC futures, non-gold options, unknown classes).
    fn definition_to_kind(def: &InstrumentDefMsg) -> Option<InstrumentKind> {
        let class = def.instrument_class().ok()?;
        let asset = def.asset().ok()?.to_string();

        match class {
            InstrumentClass::Future => {
                if asset != "GC" {
                    return None;
                }
                // We don't use month_code/year_digit downstream — the
                // orchestrator only gates on `root == "GC"`. Use
                // placeholder values.
                Some(InstrumentKind::Future {
                    root: asset,
                    month_code: '?',
                    year_digit: 0,
                })
            }
            InstrumentClass::Call | InstrumentClass::Put => {
                // OG options on GC futures have `asset == "OG"` in the
                // current GLBX.MDP3 definitions (verified against a live
                // snapshot: 19 445 Calls + 19 445 Puts all carry asset=OG).
                // The previous check insisted on "GC" and silently filtered
                // every option out, leaving the oracle with no options to
                // fit. Anything else is a different product.
                if asset != "OG" {
                    return None;
                }
                let side = if matches!(class, InstrumentClass::Call) {
                    CmeOptionSide::Call
                } else {
                    CmeOptionSide::Put
                };
                let strike = def.strike_price as f64 / 1e9;
                if !strike.is_finite() || strike <= 0.0 {
                    return None;
                }
                // expiration is u64 nanoseconds since epoch.
                let expiry_ts = (def.expiration / 1_000_000_000) as i64;
                if expiry_ts <= 0 {
                    return None;
                }
                Some(InstrumentKind::Option {
                    root: "OG".to_string(),
                    month_code: '?',
                    year_digit: 0,
                    side,
                    strike,
                    expiry_ts,
                })
            }
            _ => None,
        }
    }

    fn mbp1_to_bbo(msg: &Mbp1Msg) -> Option<Bbo> {
        let level = msg.levels.first()?;
        let bid = level.bid_px as f64 / 1e9;
        let ask = level.ask_px as f64 / 1e9;
        if !bid.is_finite() || !ask.is_finite() {
            return None;
        }
        Some(Bbo {
            bid,
            bid_size: level.bid_sz as u64,
            ask,
            ask_size: level.ask_sz as u64,
            ts_event_ms: (msg.ts_recv / 1_000_000) as i64,
        })
    }
}

#[async_trait]
impl DatabentoFeed for DatabentoLiveFeed {
    async fn next_record(&mut self) -> Option<FeedRecord> {
        // The reader tasks do all the filtering and variant-building; here
        // we just receive. `recv` returns `None` when both tasks have exited
        // (and therefore both halves of the channel are dropped), which the
        // oracle treats as a terminal condition.
        self.rx.recv().await
    }
}

/// Initial delay before the first reconnect attempt.
const RECONNECT_BASE_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
/// Maximum delay between reconnect attempts (capped exponential backoff).
const RECONNECT_MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(300);

/// Spawns a detached tokio task that connects to Databento, drives the
/// oracle's `run` loop, and automatically reconnects with exponential
/// backoff when the feed drops or fails to connect.
pub fn spawn_databento_live_feed_task(
    oracle: Arc<DatabentoVolOracle>,
    api_key: String,
    dataset: String,
) {
    tokio::spawn(async move {
        let mut attempt: u32 = 0;
        loop {
            match DatabentoLiveFeed::connect(&api_key, &dataset).await {
                Ok(feed) => {
                    let connected_at = tokio::time::Instant::now();
                    tracing::info!(
                        dataset = %dataset,
                        "Databento live feed connected — running oracle loop"
                    );
                    oracle.run(feed).await;
                    if connected_at.elapsed() > std::time::Duration::from_secs(60) {
                        attempt = 0;
                    }
                    let msg = "Databento live feed run loop exited — will reconnect";
                    tracing::warn!("{msg}");
                    oracle.record_external_error(msg.to_string());
                }
                Err(err) => {
                    let msg = format!("Failed to connect Databento live feed ({dataset}): {err}");
                    tracing::error!(error = %msg, attempt, "Databento connect failed — will retry");
                    oracle.record_external_error(msg);
                }
            }

            let delay = RECONNECT_BASE_DELAY.saturating_mul(2u32.saturating_pow(attempt.min(6)));
            let delay = delay.min(RECONNECT_MAX_DELAY);
            tracing::info!(
                dataset = %dataset,
                delay_secs = delay.as_secs(),
                attempt,
                "Databento feed reconnecting after backoff"
            );
            tokio::time::sleep(delay).await;
            attempt = attempt.saturating_add(1);
        }
    });
}

#[cfg(test)]
mod symbology_tests {
    use super::*;

    #[test]
    fn month_codes_map_to_numbers() {
        let pairs = [
            ('F', 1),
            ('G', 2),
            ('H', 3),
            ('J', 4),
            ('K', 5),
            ('M', 6),
            ('N', 7),
            ('Q', 8),
            ('U', 9),
            ('V', 10),
            ('X', 11),
            ('Z', 12),
        ];
        for (code, expected) in pairs {
            assert_eq!(cme_month_code_to_number(code), Some(expected), "{code}");
        }
        assert_eq!(cme_month_code_to_number('A'), None);
        assert_eq!(cme_month_code_to_number('I'), None); // not a valid CME code
    }

    #[test]
    fn parses_gcm6_as_gold_june_2026_future() {
        let parsed = CmeFutureSymbol::parse("GCM6").expect("GCM6 should parse");
        assert_eq!(
            parsed,
            CmeFutureSymbol {
                root: "GC".to_string(),
                month_code: 'M',
                year_digit: 6,
            }
        );
    }

    #[test]
    fn parses_ogm6_c_2400_as_june_2026_gold_call_at_2400() {
        let parsed = CmeOptionSymbol::parse("OGM6 C 2400").expect("OGM6 C 2400 should parse");
        assert_eq!(
            parsed,
            CmeOptionSymbol {
                root: "OG".to_string(),
                month_code: 'M',
                year_digit: 6,
                side: CmeOptionSide::Call,
                strike: 2400.0,
            }
        );
    }

    #[test]
    fn parses_put_side() {
        let parsed = CmeOptionSymbol::parse("OGM6 P 4800").expect("put should parse");
        assert_eq!(parsed.side, CmeOptionSide::Put);
        assert_eq!(parsed.strike, 4800.0);
    }

    #[test]
    fn parses_fractional_strike() {
        // Some CME products list fractional strikes (e.g. $0.25 ticks).
        let parsed = CmeOptionSymbol::parse("OGM6 C 4700.5").expect("fractional should parse");
        assert_eq!(parsed.strike, 4700.5);
    }

    #[test]
    fn parses_future_with_extra_whitespace() {
        assert_eq!(
            CmeFutureSymbol::parse("  GCM6  ").unwrap().root,
            "GC".to_string()
        );
    }

    #[test]
    fn rejects_wrong_length_future() {
        assert!(CmeFutureSymbol::parse("").is_err());
        assert!(CmeFutureSymbol::parse("GC").is_err());
        assert!(CmeFutureSymbol::parse("GCM62").is_err());
    }

    #[test]
    fn rejects_lowercase_root() {
        assert!(CmeFutureSymbol::parse("gcM6").is_err());
        assert!(CmeOptionSymbol::parse("ogM6 C 2400").is_err());
    }

    #[test]
    fn rejects_invalid_month_code() {
        // 'I' is not in the CME codes.
        assert!(CmeFutureSymbol::parse("GCI6").is_err());
        assert!(CmeOptionSymbol::parse("OGI6 C 2400").is_err());
    }

    #[test]
    fn rejects_non_digit_year() {
        assert!(CmeFutureSymbol::parse("GCMX").is_err());
    }

    #[test]
    fn rejects_non_cp_side() {
        assert!(CmeOptionSymbol::parse("OGM6 X 2400").is_err());
    }

    #[test]
    fn rejects_non_positive_strike() {
        assert!(CmeOptionSymbol::parse("OGM6 C 0").is_err());
        assert!(CmeOptionSymbol::parse("OGM6 C -100").is_err());
        assert!(CmeOptionSymbol::parse("OGM6 C not_a_number").is_err());
    }

    #[test]
    fn rejects_option_missing_parts() {
        assert!(CmeOptionSymbol::parse("OGM6 C").is_err());
        assert!(CmeOptionSymbol::parse("OGM6").is_err());
        assert!(CmeOptionSymbol::parse("").is_err());
    }
}