hypercall_vol_oracle/

blockscholes_oracle.rs

//! Block Scholes Volatility Oracle implementation.
//!
//! This module provides a WebSocket-based oracle for fetching real-time
//! implied volatility data from the Block Scholes API.

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use async_trait::async_trait;
use chrono::Utc;
use futures::{SinkExt, StreamExt};
use metrics::{counter, gauge};
use sonic_rs::{JsonContainerTrait, JsonValueTrait};
use tokio::task::JoinHandle;
use tokio_tungstenite::{connect_async, tungstenite::Message};
use tracing::{debug, error, info, warn};

use super::blockscholes_types::{
    AtmVolUpdateMessage, BlockScholesAuth, BlockScholesMessage, BlockScholesSubscribe,
    VolSurfaceUpdateMessage,
};
use super::risk_oracle::{
    RiskVolOracle, VolLookupError, VolOracleStatus, VolProviderKind, VolSurfaceSnapshot,
};
use super::vol_surface_cache::VolatilitySurface;
use crate::PollingVolOracle;

/// Default WebSocket URL for Block Scholes API
pub const DEFAULT_WS_URL: &str = "wss://prod-websocket-api.blockscholes.com/";

/// Default reconnection delay in milliseconds
pub const DEFAULT_RECONNECT_DELAY_MS: u64 = 5000;

/// Default heartbeat interval in milliseconds
pub const DEFAULT_HEARTBEAT_INTERVAL_MS: u64 = 30000;

/// Default cache TTL in milliseconds
pub const DEFAULT_CACHE_TTL_MS: u64 = 60000;

/// Configuration for the Block Scholes Volatility Oracle.
#[derive(Debug, Clone)]
pub struct BlockScholesVolOracleConfig {
    /// WebSocket URL for Block Scholes API
    pub ws_url: String,
    /// API key for authentication (when credentials are available)
    pub api_key: Option<String>,
    /// API secret for authentication (if required)
    pub api_secret: Option<String>,
    /// Symbols to track (e.g., ["BTC", "ETH"])
    pub symbols: Vec<String>,
    /// Strikes to subscribe to by symbol (e.g., {"BTC": [80000, 90000, 100000]})
    /// If empty, only ATM IV (delta=0.5) will be subscribed
    pub strikes: std::collections::HashMap<String, Vec<f64>>,
    /// Expiries to subscribe to in ISO 8601 format (e.g., "2026-03-27T08:00:00Z")
    /// These should be "listed" expiries from the exchange (Deribit quarterly/monthly)
    pub expiries: Vec<String>,
    /// Reconnection delay in milliseconds
    pub reconnect_delay_ms: u64,
    /// Maximum reconnection attempts (0 = infinite)
    pub max_reconnect_attempts: u32,
    /// Heartbeat interval in milliseconds
    pub heartbeat_interval_ms: u64,
    /// Cache TTL in milliseconds (how long to consider data fresh)
    pub cache_ttl_ms: u64,
}

impl Default for BlockScholesVolOracleConfig {
    fn default() -> Self {
        Self {
            ws_url: DEFAULT_WS_URL.to_string(),
            api_key: None,
            api_secret: None,
            symbols: vec!["BTC".to_string(), "ETH".to_string()],
            strikes: std::collections::HashMap::new(),
            expiries: vec!["2026-03-27T08:00:00Z".to_string()], // Default Q1 2026 expiry
            reconnect_delay_ms: DEFAULT_RECONNECT_DELAY_MS,
            max_reconnect_attempts: 0, // Infinite
            heartbeat_interval_ms: DEFAULT_HEARTBEAT_INTERVAL_MS,
            cache_ttl_ms: DEFAULT_CACHE_TTL_MS,
        }
    }
}

/// Internal state for the volatility oracle.
#[derive(Debug, Default)]
struct OracleState {
    /// Volatility surface cache by symbol
    surfaces: HashMap<String, VolatilitySurface>,
    /// Last successful update timestamp by symbol (milliseconds)
    last_update_timestamps: HashMap<String, i64>,
    /// Index (spot) prices by symbol, used for delta→strike conversion
    index_prices: HashMap<String, f64>,
    /// Connection health status
    is_connected: bool,
    /// Reconnection attempt counter
    reconnect_attempts: u32,
    /// Last provider error for reporting
    last_error: Option<String>,
}

/// Block Scholes Volatility Oracle implementation.
///
/// Connects to Block Scholes WebSocket API to receive real-time
/// volatility surface updates.
pub struct BlockScholesVolOracle {
    config: BlockScholesVolOracleConfig,
    state: Arc<RwLock<OracleState>>,
    messages_received: AtomicU64,
}

impl BlockScholesVolOracle {
    /// Create a new oracle with the given configuration.
    pub fn new(config: BlockScholesVolOracleConfig) -> Self {
        Self {
            config,
            state: Arc::new(RwLock::new(OracleState::default())),
            messages_received: AtomicU64::new(0),
        }
    }

    /// Create with default configuration.
    pub fn with_defaults() -> Self {
        Self::new(BlockScholesVolOracleConfig::default())
    }

    /// Get the oracle configuration.
    pub fn config(&self) -> &BlockScholesVolOracleConfig {
        &self.config
    }

    fn status_for(&self, symbol: &str) -> VolOracleStatus {
        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");
        let last_update_ts_ms = state.last_update_timestamps.get(symbol).copied();
        let staleness_seconds = last_update_ts_ms
            .map(|ts| ((Utc::now().timestamp_millis() - ts) as f64 / 1000.0).max(0.0));
        let surface_points = state
            .surfaces
            .get(symbol)
            .map(VolatilitySurface::len)
            .unwrap_or(0);
        let ready = state.is_connected
            && surface_points > 0
            && staleness_seconds
                .map(|age| age <= self.config.cache_ttl_ms as f64 / 1000.0)
                .unwrap_or(false);

        VolOracleStatus {
            underlying: symbol.to_string(),
            provider: VolProviderKind::BlockScholes,
            route_facing: true,
            connected: state.is_connected,
            ready,
            last_update_ts_ms,
            staleness_seconds,
            staleness_threshold_seconds: Some(self.config.cache_ttl_ms as f64 / 1000.0),
            surface_points,
            messages_received: self.messages_received.load(Ordering::Relaxed),
            last_error: state.last_error.clone(),
        }
    }

    /// Connect to WebSocket and process messages.
    async fn connect_and_run(self: &Arc<Self>) -> Result<()> {
        let url = &self.config.ws_url;
        info!("Connecting to Block Scholes WebSocket: {}", url);

        let (ws_stream, _) = connect_async(url)
            .await
            .context("Failed to connect to Block Scholes WebSocket")?;

        let (mut write, mut read) = ws_stream.split();

        // Update connection state
        {
            let mut state = self
                .state
                .write()
                .expect("block scholes vol oracle state poisoned");
            state.is_connected = true;
            state.reconnect_attempts = 0;
            state.last_error = None;
        }

        gauge!(
            "ht_vol_oracle_connected",
            "provider" => VolProviderKind::BlockScholes.as_str()
        )
        .set(1.0);

        info!("Connected to Block Scholes WebSocket");

        // Authenticate if credentials are available
        if let Some(ref api_key) = self.config.api_key {
            let auth_msg = BlockScholesAuth::new(api_key, self.config.api_secret.clone());
            let json = sonic_rs::to_string(&auth_msg)?;
            write.send(Message::Text(json)).await?;
            info!("Sent authentication request");
        }

        // Subscribe to volatility feeds for each symbol and expiry
        // Block Scholes requires "listed" expiries (actual exchange expiries)
        for symbol in &self.config.symbols {
            for expiry in &self.config.expiries {
                // Subscribe to delta grid IV instead of strike-level IV.
                // Strike-level subscriptions require exchange-listed strikes
                // (Block Scholes rejects unlisted strikes with -2610 errors).
                // Delta grid doesn't need listed strikes and gives us enough
                // points to interpolate a full vol surface via BS delta inversion.
                let delta_subscribe = BlockScholesSubscribe::delta_iv(
                    symbol.clone(),
                    expiry.clone(),
                    Some(format!("delta_{}_{}", symbol, expiry)),
                );
                let json = sonic_rs::to_string(&delta_subscribe)?;
                write.send(Message::Text(json)).await?;
                info!(
                    "Subscribed to delta.iv grid for {} at expiry {}",
                    symbol, expiry
                );
            }

            // Also subscribe to index price for the symbol
            let index_subscribe = BlockScholesSubscribe::index_price(
                symbol.clone(),
                Some(format!("index_{}", symbol)),
            );
            let json = sonic_rs::to_string(&index_subscribe)?;
            write.send(Message::Text(json)).await?;
            info!("Subscribed to index price for {}", symbol);
        }

        // Set up heartbeat interval
        let mut heartbeat =
            tokio::time::interval(Duration::from_millis(self.config.heartbeat_interval_ms));

        // Staleness watchdog: if no data messages arrive for 30s, the connection
        // is stalled (TCP alive via pings but Block Scholes not sending data).
        // Block Scholes sends data every 1s, so 30s = 30x tolerance.
        // Vol surface stale PD alert fires at 120s, so 30s gives ~90s buffer
        // for the reconnect + re-subscribe to complete before paging.
        let staleness_timeout = Duration::from_secs(30);
        let mut last_data_message = Instant::now();

        // Process messages
        loop {
            tokio::select! {
                msg = read.next() => {
                    match msg {
                        Some(Ok(Message::Text(text))) => {
                            last_data_message = Instant::now();
                            self.handle_message(&text).await;
                        }
                        Some(Ok(Message::Binary(data))) => {
                            last_data_message = Instant::now();
                            // Try to parse binary as JSON text
                            if let Ok(text) = String::from_utf8(data) {
                                self.handle_message(&text).await;
                            }
                        }
                        Some(Ok(Message::Close(frame))) => {
                            info!("WebSocket closed by server: {:?}", frame);
                            break;
                        }
                        Some(Ok(Message::Ping(data))) => {
                            if let Err(e) = write.send(Message::Pong(data)).await {
                                error!("Failed to send pong: {}", e);
                            }
                        }
                        Some(Ok(Message::Pong(_))) => {
                            debug!("Received pong");
                        }
                        Some(Ok(Message::Frame(_))) => {
                            // Raw frame, ignore
                        }
                        Some(Err(e)) => {
                            error!("WebSocket error: {}", e);
                            break;
                        }
                        None => {
                            info!("WebSocket stream ended");
                            break;
                        }
                    }
                }
                _ = heartbeat.tick() => {
                    // Send ping to keep connection alive
                    if let Err(e) = write.send(Message::Ping(vec![])).await {
                        error!("Failed to send ping: {}", e);
                        break;
                    }
                    debug!("Sent heartbeat ping");

                    // Check for stalled connection
                    if last_data_message.elapsed() > staleness_timeout {
                        warn!(
                            "Block Scholes WebSocket stalled: no data messages for {:.0}s, forcing reconnect",
                            last_data_message.elapsed().as_secs_f64()
                        );
                        counter!("ht_vol_oracle_stale_reconnects_total", "provider" => VolProviderKind::BlockScholes.as_str()).increment(1);
                        break;
                    }
                }
            }
        }

        // Mark as disconnected
        {
            let mut state = self
                .state
                .write()
                .expect("block scholes vol oracle state poisoned");
            state.is_connected = false;
        }

        gauge!(
            "ht_vol_oracle_connected",
            "provider" => VolProviderKind::BlockScholes.as_str()
        )
        .set(0.0);

        Ok(())
    }

    /// Handle incoming WebSocket message.
    async fn handle_message(&self, text: &str) {
        // First try to parse as JSON-RPC 2.0 response
        if let Ok(rpc) = sonic_rs::from_str::<super::blockscholes_types::JsonRpcResponse>(text) {
            if let Some(ref err) = rpc.error {
                error!(
                    "Block Scholes RPC error (code {}): {}",
                    err.code, err.message
                );
                return;
            }

            if rpc.is_success() {
                // Check if this is an auth response
                if rpc.result == Some(sonic_rs::json!("ok")) {
                    info!("Authentication successful");
                    return;
                }
                // Other success responses
                debug!("RPC success: {:?}", rpc.result);
                return;
            }

            // Check for subscription data push (method="subscription" with params)
            if rpc.method.as_deref() == Some("subscription") {
                if let Some(params) = rpc.params {
                    self.process_subscription_data(params).await;
                }
                return;
            }
        }

        // Fall back to trying the custom message format
        match sonic_rs::from_str::<BlockScholesMessage>(text) {
            Ok(msg) => match msg {
                BlockScholesMessage::VolSurfaceUpdate(update) => {
                    self.process_vol_surface_update(update).await;
                }
                BlockScholesMessage::AtmVolUpdate(update) => {
                    self.process_atm_vol_update(update).await;
                }
                BlockScholesMessage::Subscribed { channel, symbol } => {
                    info!("Confirmed subscription: {} for {}", channel, symbol);
                }
                BlockScholesMessage::Unsubscribed { channel, symbol } => {
                    info!("Confirmed unsubscription: {} for {}", channel, symbol);
                }
                BlockScholesMessage::Authenticated { message } => {
                    info!("Authentication successful: {:?}", message);
                }
                BlockScholesMessage::Error { message, code } => {
                    error!("Block Scholes error (code {:?}): {}", code, message);
                }
                BlockScholesMessage::Heartbeat { timestamp } => {
                    debug!("Received server heartbeat: {}", timestamp);
                }
                BlockScholesMessage::Info { message, version } => {
                    info!("Server info: {:?}, version: {:?}", message, version);
                }
            },
            Err(e) => {
                debug!("Failed to parse message: {} - {}", e, text);
            }
        }
    }

    /// Process Block Scholes subscription data.
    /// Format: [{"data": {"values": [{"sid": "...", "v": 0.42, ...}], "timestamp": 123}, "client_id": "..."}]
    async fn process_subscription_data(&self, params: sonic_rs::Value) {
        let params_array = match params.as_array() {
            Some(arr) => arr,
            None => {
                debug!("Subscription params is not an array: {:?}", params);
                return;
            }
        };

        let mut state = self
            .state
            .write()
            .expect("block scholes vol oracle state poisoned");

        for param in params_array {
            let client_id = param
                .get("client_id")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let data = match param.get("data") {
                Some(d) => d,
                None => continue,
            };

            let timestamp = data.get("timestamp").and_then(|v| v.as_i64());
            let values = match data.get("values").and_then(|v| v.as_array()) {
                Some(v) => v,
                None => continue,
            };

            for value in values {
                let sid = value.get("sid").and_then(|v| v.as_str()).unwrap_or("");

                // Extract base asset from client_id (format: "delta_BTC_expiry", "strike_BTC_expiry", or "index_BTC")
                let base_asset = if client_id.starts_with("delta_")
                    || client_id.starts_with("atm_")
                    || client_id.starts_with("strike_")
                {
                    client_id.split('_').nth(1).unwrap_or("BTC")
                } else if let Some(index_symbol) = client_id.strip_prefix("index_") {
                    index_symbol
                } else {
                    // Try to extract from sid
                    sid.split('_').nth(1).unwrap_or("BTC")
                };

                // Extract expiry timestamp from client_id if available
                // Format: "strike_BTC_2026-03-27T08:00:00Z" or "delta_BTC_2026-03-27T08:00:00Z"
                let expiry_ts = client_id
                    .split('_')
                    .nth(2)
                    .and_then(|exp| chrono::DateTime::parse_from_rfc3339(exp).ok())
                    .map(|dt| dt.timestamp())
                    .unwrap_or_else(|| timestamp.unwrap_or(0) / 1000);

                // Handle strike.iv feed (IV by strike price)
                if let Some(strikes) = value.get("strike").and_then(|v| v.as_array()) {
                    if let Some(ivs) = value.get("v").and_then(|v| v.as_array()) {
                        let surface = state.surfaces.entry(base_asset.to_string()).or_default();

                        for (strike, iv) in strikes.iter().zip(ivs.iter()) {
                            if let (Some(s), Some(v)) = (strike.as_f64(), iv.as_f64()) {
                                surface.insert(s, expiry_ts, v);
                                debug!(
                                    "Updated strike IV for {} strike={} expiry={}: {:.2}%",
                                    base_asset,
                                    s,
                                    expiry_ts,
                                    v * 100.0
                                );
                            }
                        }
                        info!(
                            "Updated {} strike IVs for {} at expiry {}",
                            strikes.len(),
                            base_asset,
                            expiry_ts
                        );
                    }
                }
                // Handle delta.iv feed — store delta-IV curve for smile interpolation
                else if let Some(deltas) = value.get("delta").and_then(|v| v.as_array()) {
                    if let Some(ivs) = value.get("v").and_then(|v| v.as_array()) {
                        let surface = state.surfaces.entry(base_asset.to_string()).or_default();

                        let mut count = 0u32;
                        for (delta, iv) in deltas.iter().zip(ivs.iter()) {
                            if let (Some(d), Some(v)) = (delta.as_f64(), iv.as_f64()) {
                                // Store in delta-IV curve for smile interpolation
                                surface.set_delta_iv(expiry_ts, d, v);
                                count += 1;

                                // Also store ATM vol for delta ≈ 0.5
                                if (d - 0.5).abs() < 0.01 {
                                    surface.set_atm_vol(expiry_ts, v);
                                }
                            }
                        }

                        if count > 0 {
                            debug!(
                                "Updated {} delta IVs for {} at expiry {}",
                                count, base_asset, expiry_ts
                            );
                        }
                    }
                }
                // Handle index.px feed — store spot price for delta→strike conversion
                else if let Some(price) = value.get("v").and_then(|v| v.as_f64()) {
                    state.index_prices.insert(base_asset.to_string(), price);
                    debug!("Index price for {}: ${:.2}", base_asset, price);
                    // Don't update last_update_timestamp for index prices —
                    // only IV data should affect staleness checks.
                    continue;
                }
                if let Some(timestamp) = timestamp {
                    state
                        .last_update_timestamps
                        .insert(base_asset.to_string(), timestamp);
                }
            }
        }

        self.messages_received.fetch_add(1, Ordering::Relaxed);
        counter!(
            "ht_vol_oracle_messages_received_total",
            "provider" => VolProviderKind::BlockScholes.as_str()
        )
        .increment(1);
    }

    /// Process volatility surface update.
    async fn process_vol_surface_update(&self, update: VolSurfaceUpdateMessage) {
        let mut state = self
            .state
            .write()
            .expect("block scholes vol oracle state poisoned");

        let surface = state.surfaces.entry(update.symbol.clone()).or_default();

        // Update the surface with new data points
        for point in &update.points {
            surface.insert(point.strike, point.expiry, point.iv);
        }

        // Update ATM vols if provided
        for atm in &update.atm_vols {
            surface.set_atm_vol(atm.expiry, atm.iv);
        }

        state
            .last_update_timestamps
            .insert(update.symbol.clone(), Utc::now().timestamp_millis());
        self.messages_received.fetch_add(1, Ordering::Relaxed);
        counter!(
            "ht_vol_oracle_messages_received_total",
            "provider" => VolProviderKind::BlockScholes.as_str(),
            "underlying" => update.symbol.clone()
        )
        .increment(1);

        debug!(
            "Updated vol surface for {}: {} points, {} ATM vols",
            update.symbol,
            update.points.len(),
            update.atm_vols.len()
        );
    }

    /// Process ATM volatility update.
    async fn process_atm_vol_update(&self, update: AtmVolUpdateMessage) {
        let mut state = self
            .state
            .write()
            .expect("block scholes vol oracle state poisoned");

        let surface = state.surfaces.entry(update.symbol.clone()).or_default();

        for point in &update.points {
            surface.set_atm_vol(point.expiry, point.iv);
        }

        state
            .last_update_timestamps
            .insert(update.symbol.clone(), Utc::now().timestamp_millis());
        self.messages_received.fetch_add(1, Ordering::Relaxed);
        counter!(
            "ht_vol_oracle_messages_received_total",
            "provider" => VolProviderKind::BlockScholes.as_str(),
            "underlying" => update.symbol.clone()
        )
        .increment(1);

        debug!(
            "Updated ATM vols for {}: {} expiries",
            update.symbol,
            update.points.len()
        );
    }

    /// Set volatility data directly (for testing).
    ///
    /// This allows tests to inject volatility data without making network requests.
    pub async fn set_vol_for_testing(&self, symbol: &str, strike: f64, expiry: i64, iv: f64) {
        let mut state = self
            .state
            .write()
            .expect("block scholes vol oracle state poisoned");
        let surface = state.surfaces.entry(symbol.to_string()).or_default();
        surface.insert(strike, expiry, iv);
        state
            .last_update_timestamps
            .insert(symbol.to_string(), Utc::now().timestamp_millis());
        state.is_connected = true;
        info!(
            "Test mode: Set vol for {} strike={} expiry={} to {:.2}%",
            symbol,
            strike,
            expiry,
            iv * 100.0
        );
    }

    /// Set ATM volatility directly (for testing).
    pub async fn set_atm_vol_for_testing(&self, symbol: &str, expiry: i64, iv: f64) {
        let mut state = self
            .state
            .write()
            .expect("block scholes vol oracle state poisoned");
        let surface = state.surfaces.entry(symbol.to_string()).or_default();
        surface.set_atm_vol(expiry, iv);
        state
            .last_update_timestamps
            .insert(symbol.to_string(), Utc::now().timestamp_millis());
        state.is_connected = true;
        info!(
            "Test mode: Set ATM vol for {} expiry={} to {:.2}%",
            symbol,
            expiry,
            iv * 100.0
        );
    }
}

#[async_trait]
impl PollingVolOracle for BlockScholesVolOracle {
    fn start_polling(self: Arc<Self>) -> JoinHandle<()> {
        let oracle = Arc::clone(&self);

        tokio::spawn(async move {
            loop {
                match oracle.connect_and_run().await {
                    Ok(()) => {
                        info!("WebSocket connection closed gracefully");
                        // Check if this was intentional shutdown
                        // For now, always attempt to reconnect
                    }
                    Err(e) => {
                        error!("WebSocket connection error: {}", e);
                    }
                }

                // Update state and check reconnection limit
                let should_stop = {
                    let mut state = oracle
                        .state
                        .write()
                        .expect("block scholes vol oracle state poisoned");
                    state.is_connected = false;
                    state.reconnect_attempts += 1;
                    state.last_error = Some("websocket disconnected".to_string());

                    oracle.config.max_reconnect_attempts > 0
                        && state.reconnect_attempts >= oracle.config.max_reconnect_attempts
                };

                if should_stop {
                    error!("Max reconnection attempts reached, stopping oracle");
                    break;
                }

                let attempts = oracle
                    .state
                    .read()
                    .expect("block scholes vol oracle state poisoned")
                    .reconnect_attempts;
                info!(
                    "Reconnecting in {}ms (attempt {})",
                    oracle.config.reconnect_delay_ms, attempts
                );

                tokio::time::sleep(Duration::from_millis(oracle.config.reconnect_delay_ms)).await;
            }
        })
    }

    async fn get_vol(&self, symbol: &str, strike: f64, expiry: i64) -> Option<f64> {
        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");
        let spot = state.index_prices.get(symbol).copied();
        state
            .surfaces
            .get(symbol)
            .and_then(|surface| surface.get_interpolated_with_spot(strike, expiry, spot))
    }

    async fn get_atm_vol(&self, symbol: &str, expiry: i64) -> Option<f64> {
        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");
        state
            .surfaces
            .get(symbol)
            .and_then(|surface| surface.get_atm(expiry))
    }

    async fn is_healthy(&self) -> bool {
        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");

        // Check if connected
        if !state.is_connected {
            return false;
        }

        // Check if data is fresh (within cache TTL)
        if let Some(last_update) = state.last_update_timestamps.values().copied().max() {
            let now = Utc::now().timestamp_millis();
            let age = now - last_update;
            return age < self.config.cache_ttl_ms as i64;
        }

        false
    }

    async fn last_update_timestamp(&self) -> Option<i64> {
        self.state
            .read()
            .expect("block scholes vol oracle state poisoned")
            .last_update_timestamps
            .values()
            .copied()
            .max()
    }
}

impl RiskVolOracle for BlockScholesVolOracle {
    fn get_iv(&self, underlying: &str, strike: f64, expiry_ts: i64) -> Result<f64, VolLookupError> {
        let status = self.status_for(underlying);
        if !status.connected {
            return Err(VolLookupError::UnhealthyProvider {
                underlying: underlying.to_string(),
                provider: VolProviderKind::BlockScholes,
                reason: status
                    .last_error
                    .unwrap_or_else(|| "not connected".to_string()),
            });
        }

        // Skip the ready check — let the lookup itself determine if data exists.
        // The readiness gate was causing false "stale surface" errors when index
        // prices updated the timestamp but no IV data had arrived yet.

        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");
        let spot = state.index_prices.get(underlying).copied();
        let iv = state
            .surfaces
            .get(underlying)
            .and_then(|surface| surface.get_interpolated_with_spot(strike, expiry_ts, spot))
            .ok_or_else(|| VolLookupError::MissingSurface {
                underlying: underlying.to_string(),
                provider: VolProviderKind::BlockScholes,
                strike,
                expiry_ts,
            })?;

        debug!(
            underlying,
            strike,
            expiry_ts,
            iv,
            spot = ?spot,
            provider = VolProviderKind::BlockScholes.as_str(),
            "Backend vol oracle value used"
        );

        Ok(iv)
    }

    fn statuses(&self) -> Vec<VolOracleStatus> {
        self.config
            .symbols
            .iter()
            .map(|symbol| self.status_for(symbol))
            .collect()
    }

    fn get_surface_snapshot(&self, underlying: &str) -> Option<VolSurfaceSnapshot> {
        let state = self
            .state
            .read()
            .expect("block scholes vol oracle state poisoned");
        let surface = state.surfaces.get(underlying)?;
        Some(VolSurfaceSnapshot {
            underlying: underlying.to_string(),
            last_update_ts_ms: state.last_update_timestamps.get(underlying).copied(),
            expiries: surface.expiries().iter().copied().collect(),
            strike_points: surface.export_all_points(),
            delta_curves: surface.export_delta_curves(),
            atm_vols: surface.export_atm_vols(),
            spot_price: state.index_prices.get(underlying).copied(),
        })
    }

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

/// Mock polling oracle for unit testing margin calculations.
///
/// Returns a fixed volatility value for all queries.
pub struct MockVolOracle {
    fixed_vol: f64,
}

impl MockVolOracle {
    /// Create a new mock oracle with a fixed volatility value.
    pub fn new(fixed_vol: f64) -> Self {
        Self { fixed_vol }
    }
}

#[async_trait]
impl PollingVolOracle for MockVolOracle {
    fn start_polling(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async {})
    }

    async fn get_vol(&self, _symbol: &str, _strike: f64, _expiry: i64) -> Option<f64> {
        Some(self.fixed_vol)
    }

    async fn get_atm_vol(&self, _symbol: &str, _expiry: i64) -> Option<f64> {
        Some(self.fixed_vol)
    }

    async fn is_healthy(&self) -> bool {
        true
    }

    async fn last_update_timestamp(&self) -> Option<i64> {
        Some(Utc::now().timestamp_millis())
    }
}

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

    #[test]
    fn test_config_default() {
        let config = BlockScholesVolOracleConfig::default();
        assert_eq!(config.symbols, vec!["BTC", "ETH"]);
        assert_eq!(config.reconnect_delay_ms, DEFAULT_RECONNECT_DELAY_MS);
        assert_eq!(config.max_reconnect_attempts, 0);
        assert!(config.api_key.is_none());
    }

    #[test]
    fn test_oracle_creation() {
        let oracle = BlockScholesVolOracle::with_defaults();
        assert_eq!(oracle.config().symbols, vec!["BTC", "ETH"]);
    }

    #[tokio::test]
    async fn test_set_vol_for_testing() {
        let oracle = BlockScholesVolOracle::with_defaults();

        oracle
            .set_vol_for_testing("BTC", 100000.0, 1735689600, 0.75)
            .await;

        let vol = oracle.get_vol("BTC", 100000.0, 1735689600).await;
        assert_eq!(vol, Some(0.75));
    }

    #[tokio::test]
    async fn test_set_atm_vol_for_testing() {
        let oracle = BlockScholesVolOracle::with_defaults();

        oracle
            .set_atm_vol_for_testing("ETH", 1735689600, 0.80)
            .await;

        let vol = oracle.get_atm_vol("ETH", 1735689600).await;
        assert_eq!(vol, Some(0.80));
    }

    #[tokio::test]
    async fn test_get_vol_not_found() {
        let oracle = BlockScholesVolOracle::with_defaults();

        let vol = oracle.get_vol("BTC", 100000.0, 1735689600).await;
        assert_eq!(vol, None);
    }

    #[tokio::test]
    async fn test_is_healthy_when_not_connected() {
        let oracle = BlockScholesVolOracle::with_defaults();

        assert!(!oracle.is_healthy().await);
    }

    #[tokio::test]
    async fn test_is_healthy_after_test_data() {
        let oracle = BlockScholesVolOracle::with_defaults();

        // Setting test data marks the oracle as connected
        oracle
            .set_vol_for_testing("BTC", 100000.0, 1735689600, 0.75)
            .await;

        assert!(oracle.is_healthy().await);
    }

    #[tokio::test]
    async fn test_mock_vol_oracle() {
        let mock = MockVolOracle::new(0.65);

        assert_eq!(mock.get_vol("ANY", 0.0, 0).await, Some(0.65));
        assert_eq!(mock.get_atm_vol("ANY", 0).await, Some(0.65));
        assert!(mock.is_healthy().await);
        assert!(mock.last_update_timestamp().await.is_some());
    }

    #[tokio::test]
    async fn test_interpolation_through_oracle() {
        let oracle = BlockScholesVolOracle::with_defaults();

        // Set up a 2x2 grid
        oracle
            .set_vol_for_testing("BTC", 40000.0, 1735689600, 0.70)
            .await;
        oracle
            .set_vol_for_testing("BTC", 40000.0, 1735776000, 0.75)
            .await;
        oracle
            .set_vol_for_testing("BTC", 60000.0, 1735689600, 0.80)
            .await;
        oracle
            .set_vol_for_testing("BTC", 60000.0, 1735776000, 0.85)
            .await;

        // Query center point
        let mid_expiry = (1735689600 + 1735776000) / 2;
        let vol = oracle.get_vol("BTC", 50000.0, mid_expiry).await;

        assert!(vol.is_some());
        let v = vol.unwrap();
        // Should be around 0.775 (average of corners)
        assert!(v > 0.70 && v < 0.90, "Expected ~0.775, got {}", v);
    }

    #[tokio::test]
    async fn test_atm_fallback_when_no_strikes() {
        // When only ATM data exists (no delta curves, no strikes), fall back to ATM
        let oracle = BlockScholesVolOracle::with_defaults();

        oracle
            .set_atm_vol_for_testing("BTC", 1735689600, 0.65)
            .await;
        oracle
            .set_atm_vol_for_testing("BTC", 1735776000, 0.70)
            .await;

        // Query an arbitrary strike — should fall back to ATM
        let vol = oracle.get_vol("BTC", 95000.0, 1735689600).await;
        assert_eq!(vol, Some(0.65), "Should fall back to ATM vol");

        // Query between expiries — should interpolate ATM
        let mid = (1735689600 + 1735776000) / 2;
        let vol = oracle.get_vol("BTC", 95000.0, mid).await;
        assert!(vol.is_some());
        let v = vol.unwrap();
        assert!(
            v > 0.64 && v < 0.71,
            "Expected interpolated ATM ~0.675, got {}",
            v
        );
    }
}
// CALL-681: force rebuild for fixed vol catalog change