hypercall_vol_oracle/

derive_oracle.rs

//! Derive (Lyra) Volatility Oracle implementation.
//!
//! TODO(clients): Move Derive HTTP transport and wire models into a dedicated client crate.
//!
//! Polls Derive's public API for option tickers and populates a volatility
//! surface with exchange-computed IV values. No API key required.
//!
//! API pattern:
//! 1. GET /public/get_instruments?currency=X&instrument_type=option&expired=false
//!    -> list of instruments with expiry timestamps
//! 2. POST /public/get_tickers {currency, instrument_type, expiry_date}
//!    -> bulk ticker data per expiry with IV in option_pricing.i

use std::collections::{BTreeSet, 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_DERIVE_BASE_URL: &str = "https://api.lyra.finance";

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

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

// --- API response types ---

#[derive(Debug, Deserialize)]
struct DeriveInstrumentsResponse {
    result: Vec<DeriveInstrument>,
}

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct DeriveInstrument {
    instrument_name: String,
    option_details: Option<DeriveOptionDetails>,
}

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct DeriveOptionDetails {
    expiry: i64,
    strike: String,
}

/// Bulk tickers response. Keyed by instrument name with abbreviated fields.
#[derive(Debug, Deserialize)]
struct DeriveTickersResponse {
    result: DeriveTickersResult,
}

#[derive(Debug, Deserialize)]
struct DeriveTickersResult {
    tickers: HashMap<String, DeriveTicker>,
}

#[derive(Debug, Deserialize)]
struct DeriveTicker {
    /// Index (spot) price
    #[serde(rename = "I")]
    index_price: Option<String>,
    /// Option pricing with abbreviated keys
    option_pricing: Option<DeriveOptionPricing>,
}

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct DeriveOptionPricing {
    /// Implied volatility as decimal string (e.g., "0.812" = 81.2%)
    #[serde(rename = "i")]
    iv: Option<String>,
    /// Mark price
    #[serde(rename = "m")]
    mark_price: Option<String>,
}

pub struct DeriveVolOracle {
    client: Client,
    config: DeriveVolOracleConfig,
    state: Arc<RwLock<DeriveSurfaceState>>,
    messages_received: AtomicU64,
}

impl DeriveVolOracle {
    pub fn new(config: DeriveVolOracleConfig) -> Self {
        Self {
            client: Client::builder()
                .timeout(Duration::from_secs(15))
                .build()
                .expect("failed to build HTTP client"),
            config,
            state: Arc::new(RwLock::new(DeriveSurfaceState::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!("Derive 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("derive vol oracle state poisoned");
                        // Only swap if new surface is at least 50% the size of the
                        // existing one to avoid replacing a full surface with a
                        // partial one when some per-expiry ticker fetches fail.
                        let existing_points =
                            state.surfaces.get(symbol).map(|s| s.len()).unwrap_or(0);
                        let accepted = point_count > 0
                            && (existing_points == 0 || point_count >= existing_points / 2);
                        if accepted {
                            state.surfaces.insert(symbol.clone(), surface);
                            state.connected.insert(symbol.clone(), true);
                            state.last_update_ts_ms.insert(symbol.clone(), now_ms);
                            state.last_error.remove(symbol);
                        } else if existing_points > 0 {
                            // Partial refresh rejected - don't update timestamp so
                            // staleness gate will eventually fire if this persists.
                            warn!(
                                "Derive partial refresh for {} ({} points vs {} existing), keeping old surface",
                                symbol, point_count, existing_points
                            );
                        }
                    }
                    self.messages_received.fetch_add(1, Ordering::Relaxed);
                    counter!(
                        "ht_vol_oracle_messages_received_total",
                        "provider" => VolProviderKind::Derive.as_str(),
                        "underlying" => symbol.clone()
                    )
                    .increment(1);
                    info!(
                        "Updated Derive vol surface for {} with {} points (spot ${:.2})",
                        symbol, point_count, spot
                    );
                }
                Err(err) => {
                    let message = err.to_string();
                    let mut state = self
                        .state
                        .write()
                        .expect("derive 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!(
                            "Derive vol surface refresh failed for {} (keeping last good data): {}",
                            symbol, message
                        );
                    } else {
                        warn!(
                            "Derive vol surface refresh failed for {}: {}",
                            symbol, message
                        );
                    }
                }
            }
        }

        Ok(())
    }

    /// Fetch the full vol surface for a single underlying from Derive.
    ///
    /// 1. Fetch instruments to discover expiry dates
    /// 2. For each expiry, POST get_tickers to get bulk IV data
    /// 3. Parse instrument names for strike, read IV from option_pricing.i
    async fn fetch_surface(&self, symbol: &str) -> Result<(VolatilitySurface, f64)> {
        // Step 1: Fetch instruments to discover expiries
        let instruments_url = format!(
            "{}/public/get_instruments?instrument_type=option&currency={}&expired=false",
            self.config.base_url, symbol
        );
        let instruments_resp = self
            .client
            .get(&instruments_url)
            .send()
            .await
            .with_context(|| format!("Failed to fetch Derive instruments for {symbol}"))?
            .error_for_status()
            .with_context(|| format!("Derive instruments returned non-success for {symbol}"))?
            .json::<DeriveInstrumentsResponse>()
            .await
            .with_context(|| format!("Failed to decode Derive instruments for {symbol}"))?;

        // Collect unique expiry timestamps
        let now_ts = Utc::now().timestamp();
        let mut expiry_dates: BTreeSet<(i64, String)> = BTreeSet::new();
        for inst in &instruments_resp.result {
            if let Some(ref details) = inst.option_details {
                if details.expiry > now_ts {
                    if let Some(dt) = chrono::DateTime::from_timestamp(details.expiry, 0) {
                        let date_str = dt.format("%Y%m%d").to_string();
                        expiry_dates.insert((details.expiry, date_str));
                    } else {
                        warn!(
                            "Skipping invalid Derive expiry timestamp: {}",
                            details.expiry
                        );
                    }
                }
            }
        }

        if expiry_dates.is_empty() {
            anyhow::bail!("Derive returned no active expiries for {symbol}");
        }

        // Step 2: For each expiry, fetch tickers
        // Use $0.5 strike precision since Derive lists half-dollar strikes
        // (e.g., HYPE at $34.5, $35.0). The default $100 precision would
        // quantize all HYPE strikes into ~2 buckets.
        let mut surface = VolatilitySurface::with_precision(0.5);
        let mut spot_price = 0.0_f64;
        let mut total_inserted = 0_u32;

        let tickers_url = format!("{}/public/get_tickers", self.config.base_url);

        for (expiry_ts, expiry_date) in &expiry_dates {
            let body = serde_json::json!({
                "currency": symbol,
                "instrument_type": "option",
                "expiry_date": expiry_date,
            });

            let resp = match self.client.post(&tickers_url).json(&body).send().await {
                Ok(r) => match r.error_for_status() {
                    Ok(r) => r,
                    Err(e) => {
                        warn!(
                            "Derive get_tickers HTTP error for {} expiry {}: {}",
                            symbol, expiry_date, e
                        );
                        continue;
                    }
                },
                Err(e) => {
                    warn!(
                        "Derive get_tickers failed for {} expiry {}: {}",
                        symbol, expiry_date, e
                    );
                    continue;
                }
            };

            let tickers: DeriveTickersResponse = match resp.json().await {
                Ok(t) => t,
                Err(e) => {
                    warn!(
                        "Derive get_tickers decode failed for {} expiry {}: {}",
                        symbol, expiry_date, e
                    );
                    continue;
                }
            };

            let mut expiry_count = 0_u32;
            for (inst_name, ticker) in &tickers.result.tickers {
                // Parse strike from instrument name: "HYPE-20260410-37-C" -> 37
                // Also handles half-strikes like "HYPE-20260410-34_5-C" -> 34.5
                let strike = match parse_strike_from_instrument(inst_name) {
                    Some(s) => s,
                    None => continue,
                };

                // Get IV
                let iv = match ticker
                    .option_pricing
                    .as_ref()
                    .and_then(|op| op.iv.as_ref())
                    .and_then(|s| s.parse::<f64>().ok())
                {
                    Some(v) if v > 0.01 && v < 5.0 => v,
                    _ => continue,
                };

                // Get spot price
                if let Some(ref idx) = ticker.index_price {
                    if let Ok(p) = idx.parse::<f64>() {
                        if p > 0.0 {
                            spot_price = p;
                        }
                    }
                }

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

                // Set ATM vol if close to spot
                if spot_price > 0.0 {
                    let moneyness = (strike / spot_price - 1.0).abs();
                    if moneyness < 0.03 {
                        surface.set_atm_vol(*expiry_ts, iv);
                    }
                }
            }

            total_inserted += expiry_count;
            debug!(
                "Derive {} expiry {}: {} IV points",
                symbol, expiry_date, expiry_count
            );
        }

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

        // Enforce static no-arbitrage on the ingested surface. Derive quotes
        // HYPE 0DTE/weeklies from a thin order book where sparse strike
        // coverage routinely leaves monotonicity/butterfly violations in the
        // raw mark IVs.
        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::Derive.as_str(),
                    "underlying" => symbol.to_string()
                )
                .increment(clamps as u64);
                info!(
                    underlying = symbol,
                    clamps, "Clamped non-arb-free vol surface points (derive)"
                );
            }
        }

        Ok((surface, spot_price))
    }

    fn status_for(&self, symbol: &str) -> VolOracleStatus {
        let state = self.state.read().expect("derive 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::Derive,
            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 DeriveVolOracle {
    fn get_iv(&self, underlying: &str, strike: f64, expiry_ts: i64) -> Result<f64, VolLookupError> {
        let state = self.state.read().expect("derive 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::Derive,
                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::Derive,
                staleness_seconds: staleness_seconds.unwrap_or(f64::INFINITY),
                threshold_seconds: self.config.staleness_threshold.as_secs_f64(),
            });
        }

        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::Derive,
                strike,
                expiry_ts,
            })?;

        debug!(
            underlying,
            strike,
            expiry_ts,
            iv,
            provider = VolProviderKind::Derive.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("derive 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
    }
}

/// Parse strike price from a Derive instrument name.
///
/// Formats:
/// - "HYPE-20260410-37-C" -> 37.0
/// - "HYPE-20260410-34_5-C" -> 34.5  (underscore = decimal point)
fn parse_strike_from_instrument(name: &str) -> Option<f64> {
    let parts: Vec<&str> = name.split('-').collect();
    if parts.len() < 4 {
        return None;
    }
    // Third part is the strike, with underscore for decimal
    let strike_str = parts[2].replace('_', ".");
    strike_str.parse::<f64>().ok()
}

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

    #[test]
    fn test_parse_strike() {
        assert_eq!(
            parse_strike_from_instrument("HYPE-20260410-37-C"),
            Some(37.0)
        );
        assert_eq!(
            parse_strike_from_instrument("HYPE-20260410-34_5-P"),
            Some(34.5)
        );
        assert_eq!(
            parse_strike_from_instrument("HYPE-20260410-100-C"),
            Some(100.0)
        );
        assert_eq!(parse_strike_from_instrument("bad"), None);
    }

    #[test]
    fn test_oracle_not_ready_before_poll() {
        let config = DeriveVolOracleConfig {
            base_url: DEFAULT_DERIVE_BASE_URL.to_string(),
            poll_interval: Duration::from_secs(30),
            staleness_threshold: Duration::from_secs(120),
            symbols: vec!["HYPE".to_string()],
        };
        let oracle = DeriveVolOracle::new(config);

        let result = oracle.get_iv("HYPE", 35.0, 1800000000);
        assert!(matches!(
            result,
            Err(VolLookupError::UnhealthyProvider { .. })
        ));
    }
}