hypercall_vol_oracle/

deribit_oracle.rs

//! Deribit Volatility Oracle implementation.
//!
//! TODO(clients): Move Deribit HTTP transport and wire models into a dedicated client crate.
//!
//! Polls Deribit's public API for option book summaries and populates a
//! volatility surface with exchange-computed `mark_iv` values. No API key
//! required - uses only public endpoints.
//!
//! Follows the same HTTP polling pattern as `PolygonVolOracle`.

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

use anyhow::{Context, Result};
use chrono::Utc;
use metrics::counter;
use reqwest::Client;
use serde::Deserialize;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

use super::risk_oracle::{
    RiskVolOracle, VolLookupError, VolOracleStatus, VolProviderKind, VolSurfaceSnapshot,
};
use super::vol_surface_cache::VolatilitySurface;

pub const DEFAULT_DERIBIT_BASE_URL: &str = "https://www.deribit.com/api/v2";

#[derive(Debug, Clone)]
pub struct DeribitVolOracleConfig {
    pub base_url: String,
    pub poll_interval: Duration,
    pub staleness_threshold: Duration,
    /// Underlyings to poll (e.g., ["BTC", "ETH"]).
    pub symbols: Vec<String>,
}

#[derive(Debug, Default)]
struct DeribitSurfaceState {
    surfaces: HashMap<String, VolatilitySurface>,
    connected: HashMap<String, bool>,
    last_update_ts_ms: HashMap<String, i64>,
    last_error: HashMap<String, String>,
}

/// Deribit book summary response.
#[derive(Debug, Deserialize)]
struct DeribitResponse {
    result: Vec<BookSummary>,
}

#[derive(Debug, Deserialize)]
struct BookSummary {
    instrument_name: String,
    /// Mark IV as a percentage (e.g., 65.5 = 65.5%)
    mark_iv: Option<f64>,
    underlying_price: Option<f64>,
}

/// Deribit instruments response (for fetching expiry timestamps).
#[derive(Debug, Deserialize)]
struct DeribitInstrumentsResponse {
    result: Vec<DeribitInstrument>,
}

#[derive(Debug, Deserialize)]
struct DeribitInstrument {
    instrument_name: String,
    strike: f64,
    /// Expiry timestamp in milliseconds
    expiration_timestamp: i64,
}

pub struct DeribitVolOracle {
    client: Client,
    config: DeribitVolOracleConfig,
    state: Arc<RwLock<DeribitSurfaceState>>,
    messages_received: AtomicU64,
}

impl DeribitVolOracle {
    pub fn new(config: DeribitVolOracleConfig) -> Self {
        Self {
            client: Client::builder()
                .timeout(Duration::from_secs(15))
                .build()
                .expect("failed to build HTTP client"),
            config,
            state: Arc::new(RwLock::new(DeribitSurfaceState::default())),
            messages_received: AtomicU64::new(0),
        }
    }

    pub fn start_polling(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(self.config.poll_interval);
            loop {
                interval.tick().await;
                if let Err(err) = self.refresh_all().await {
                    error!("Deribit vol oracle refresh failed: {err:#}");
                }
            }
        })
    }

    async fn refresh_all(&self) -> Result<()> {
        for symbol in &self.config.symbols {
            match self.fetch_surface(symbol).await {
                Ok((surface, spot)) => {
                    let now_ms = Utc::now().timestamp_millis();
                    let point_count = surface.len();
                    {
                        let mut state = self
                            .state
                            .write()
                            .expect("deribit vol oracle state poisoned");
                        state.connected.insert(symbol.clone(), true);
                        state.last_update_ts_ms.insert(symbol.clone(), now_ms);
                        state.last_error.remove(symbol);
                        state.surfaces.insert(symbol.clone(), surface);
                    }
                    self.messages_received.fetch_add(1, Ordering::Relaxed);
                    counter!(
                        "ht_vol_oracle_messages_received_total",
                        "provider" => VolProviderKind::Deribit.as_str(),
                        "underlying" => symbol.clone()
                    )
                    .increment(1);
                    info!(
                        "Updated Deribit vol surface for {} with {} points (spot ${:.0})",
                        symbol, point_count, spot
                    );
                }
                Err(err) => {
                    let message = err.to_string();
                    let mut state = self
                        .state
                        .write()
                        .expect("deribit vol oracle state poisoned");
                    state.connected.insert(symbol.clone(), false);
                    state.last_error.insert(symbol.clone(), message.clone());
                    if state.surfaces.contains_key(symbol) {
                        warn!(
                            "Deribit vol surface refresh failed for {} (keeping last good data): {}",
                            symbol, message
                        );
                    } else {
                        warn!(
                            "Deribit vol surface refresh failed for {}: {}",
                            symbol, message
                        );
                    }
                }
            }
        }

        Ok(())
    }

    /// Fetch the full vol surface for a single underlying from Deribit.
    ///
    /// Uses `get_book_summary_by_currency` which returns mark_iv for every
    /// listed option in a single request (no pagination needed).
    async fn fetch_surface(&self, symbol: &str) -> Result<(VolatilitySurface, f64)> {
        // Step 1: Fetch instruments to get strike/expiry mappings.
        // The book summary only has instrument_name, not structured strike/expiry.
        let instruments = self.fetch_instruments(symbol).await?;

        // Build a lookup from instrument_name -> (strike, expiry_ts)
        let mut instrument_map: HashMap<String, (f64, i64)> = HashMap::new();
        for inst in &instruments {
            // expiration_timestamp is in milliseconds
            instrument_map.insert(
                inst.instrument_name.clone(),
                (inst.strike, inst.expiration_timestamp / 1000),
            );
        }

        // Step 2: Fetch book summaries with mark_iv
        let url = format!(
            "{}/public/get_book_summary_by_currency",
            self.config.base_url
        );
        let resp = self
            .client
            .get(&url)
            .query(&[("currency", symbol), ("kind", "option")])
            .send()
            .await
            .with_context(|| format!("Failed to fetch Deribit book summary for {symbol}"))?
            .error_for_status()
            .with_context(|| format!("Deribit returned non-success status for {symbol}"))?
            .json::<DeribitResponse>()
            .await
            .with_context(|| format!("Failed to decode Deribit book summary for {symbol}"))?;

        // Use $1 strike precision to avoid collapsing nearby strikes.
        // ETH has $25-$50 strike spacing which would merge in the default
        // $100 buckets. BTC's $1000 spacing works fine at $1 precision.
        let mut surface = VolatilitySurface::with_precision(1.0);
        let now_ts = Utc::now().timestamp();
        let mut spot_price = 0.0_f64;
        let mut inserted = 0_u32;

        for summary in &resp.result {
            // Grab spot price from any entry that has it
            if let Some(price) = summary.underlying_price {
                if price > 0.0 {
                    spot_price = price;
                }
            }

            let Some(mark_iv_pct) = summary.mark_iv else {
                continue;
            };
            // Deribit returns IV as percentage (e.g., 65.5 = 65.5%)
            let iv = mark_iv_pct / 100.0;

            // Sanity: IV between 1% and 500%
            if !(0.01..5.0).contains(&iv) {
                debug!(
                    "Skipping {} with IV {:.1}% (out of range)",
                    summary.instrument_name, mark_iv_pct
                );
                continue;
            }

            // Look up strike and expiry from instruments
            let Some(&(strike, expiry_ts)) = instrument_map.get(&summary.instrument_name) else {
                continue;
            };

            // Skip expired
            if expiry_ts <= now_ts {
                continue;
            }

            surface.insert(strike, expiry_ts, iv);
            inserted += 1;

            // Also set ATM vol for this expiry if this is close to ATM
            if spot_price > 0.0 {
                let moneyness = (strike / spot_price - 1.0).abs();
                if moneyness < 0.02 {
                    surface.set_atm_vol(expiry_ts, iv);
                }
            }
        }

        if inserted == 0 {
            anyhow::bail!("Deribit returned no usable IV points for {symbol}");
        }

        debug!(
            "Deribit {} surface: {} points from {} book summaries, {} instruments",
            symbol,
            inserted,
            resp.result.len(),
            instruments.len()
        );

        // Enforce static no-arbitrage on the ingested surface. Deribit mark IVs
        // are usually well-behaved, but sparse strikes on short tenors can
        // still leave butterflies or monotonicity violations.
        if spot_price > 0.0 {
            let clamps = surface.sanitize_arb_free(spot_price, 0.0);
            if clamps > 0 {
                counter!(
                    "ht_vol_surface_arb_clamps_total",
                    "provider" => VolProviderKind::Deribit.as_str(),
                    "underlying" => symbol.to_string()
                )
                .increment(clamps as u64);
                info!(
                    underlying = symbol,
                    clamps, "Clamped non-arb-free vol surface points (deribit)"
                );
            }
        }

        Ok((surface, spot_price))
    }

    /// Fetch active option instruments for a currency.
    async fn fetch_instruments(&self, symbol: &str) -> Result<Vec<DeribitInstrument>> {
        let url = format!("{}/public/get_instruments", self.config.base_url);
        let resp = self
            .client
            .get(&url)
            .query(&[
                ("currency", symbol),
                ("kind", "option"),
                ("expired", "false"),
            ])
            .send()
            .await
            .with_context(|| format!("Failed to fetch Deribit instruments for {symbol}"))?
            .error_for_status()
            .with_context(|| {
                format!("Deribit instruments returned non-success status for {symbol}")
            })?
            .json::<DeribitInstrumentsResponse>()
            .await
            .with_context(|| format!("Failed to decode Deribit instruments for {symbol}"))?;

        Ok(resp.result)
    }

    fn status_for(&self, symbol: &str) -> VolOracleStatus {
        let state = self
            .state
            .read()
            .expect("deribit vol oracle state poisoned");
        let last_update_ts_ms = state.last_update_ts_ms.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 ready = staleness_seconds
            .map(|age| age <= self.config.staleness_threshold.as_secs_f64())
            .unwrap_or(false);
        let surface_points = state
            .surfaces
            .get(symbol)
            .map(VolatilitySurface::len)
            .unwrap_or(0);

        VolOracleStatus {
            underlying: symbol.to_string(),
            provider: VolProviderKind::Deribit,
            route_facing: true,
            connected: state.connected.get(symbol).copied().unwrap_or(false),
            ready,
            last_update_ts_ms,
            staleness_seconds,
            staleness_threshold_seconds: Some(self.config.staleness_threshold.as_secs_f64()),
            surface_points,
            messages_received: self.messages_received.load(Ordering::Relaxed),
            last_error: state.last_error.get(symbol).cloned(),
        }
    }
}

impl RiskVolOracle for DeribitVolOracle {
    fn get_iv(&self, underlying: &str, strike: f64, expiry_ts: i64) -> Result<f64, VolLookupError> {
        let state = self
            .state
            .read()
            .expect("deribit vol oracle state poisoned");
        let has_surface = state.surfaces.contains_key(underlying);

        // If disconnected and no cached surface, fail immediately
        let connected = state.connected.get(underlying).copied().unwrap_or(false);
        if !connected && !has_surface {
            return Err(VolLookupError::UnhealthyProvider {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Deribit,
                reason: state
                    .last_error
                    .get(underlying)
                    .cloned()
                    .unwrap_or_else(|| "not connected".to_string()),
            });
        }

        // Check staleness (applies whether connected or not)
        let last_update_ts_ms = state.last_update_ts_ms.get(underlying).copied();
        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);

        if !ready {
            return Err(VolLookupError::StaleSurface {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Deribit,
                staleness_seconds: staleness_seconds.unwrap_or(f64::INFINITY),
                threshold_seconds: self.config.staleness_threshold.as_secs_f64(),
            });
        }

        // Surface exists and is fresh -- serve it
        let iv = state
            .surfaces
            .get(underlying)
            .and_then(|surface| surface.get_interpolated(strike, expiry_ts))
            .ok_or_else(|| VolLookupError::MissingSurface {
                underlying: underlying.to_string(),
                provider: VolProviderKind::Deribit,
                strike,
                expiry_ts,
            })?;

        debug!(
            underlying,
            strike,
            expiry_ts,
            iv,
            provider = VolProviderKind::Deribit.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("deribit vol oracle state poisoned");
        let surface = state.surfaces.get(underlying)?;
        Some(VolSurfaceSnapshot {
            underlying: underlying.to_string(),
            last_update_ts_ms: state.last_update_ts_ms.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: None,
        })
    }

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

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

    #[test]
    fn test_oracle_not_ready_before_poll() {
        let config = DeribitVolOracleConfig {
            base_url: DEFAULT_DERIBIT_BASE_URL.to_string(),
            poll_interval: Duration::from_secs(15),
            staleness_threshold: Duration::from_secs(120),
            symbols: vec!["BTC".to_string()],
        };
        let oracle = DeribitVolOracle::new(config);

        // Should fail with unhealthy before any data arrives
        let result = oracle.get_iv("BTC", 85000.0, 1800000000);
        assert!(matches!(
            result,
            Err(VolLookupError::UnhealthyProvider { .. })
        ));
    }
}