hypercall_admin/monitoring/

oracles.rs

//! Oracle health + vol surface admin endpoints.

use axum::{extract::State, http::StatusCode, response::IntoResponse};
use serde::{Deserialize, Serialize};
use sonic_rs::json;
use std::collections::HashMap;
use tracing::warn;

use crate::state::AdminState;
use hypercall_runtime_api::sonic_json::SonicJson;

#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct VolOracleStatusResponse {
    pub underlying: String,
    pub provider: String,
    pub route_facing: bool,
    pub connected: bool,
    pub ready: bool,
    pub last_update_ts_ms: Option<i64>,
    pub staleness_seconds: Option<f64>,
    pub staleness_threshold_seconds: Option<f64>,
    pub surface_points: usize,
    pub messages_received: u64,
    pub last_error: Option<String>,
}

#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct VolOracleStatusesResponse {
    pub statuses: Vec<VolOracleStatusResponse>,
}

#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct PriceOracleStatusResponse {
    pub underlying: String,
    pub connected: bool,
    pub ready: bool,
    pub spot_price: Option<f64>,
    pub prev_day_price: Option<f64>,
    pub staleness_seconds: Option<f64>,
}

#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct PriceOracleStatusesResponse {
    pub statuses: Vec<PriceOracleStatusResponse>,
}

/// List vol oracle health and freshness by underlying.
#[utoipa::path(
    get,
    path = "/monitoring/vol-oracles",
    responses(
        (status = 200, description = "Vol oracle status", body = VolOracleStatusesResponse),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn vol_oracles(State(app_state): State<AdminState>) -> impl IntoResponse {
    let statuses = app_state
        .risk_vol_oracle
        .statuses()
        .into_iter()
        .map(|status| VolOracleStatusResponse {
            underlying: status.underlying,
            provider: status.provider.as_str().to_string(),
            route_facing: status.route_facing,
            connected: status.connected,
            ready: status.ready,
            last_update_ts_ms: status.last_update_ts_ms,
            staleness_seconds: status.staleness_seconds,
            staleness_threshold_seconds: status.staleness_threshold_seconds,
            surface_points: status.surface_points,
            messages_received: status.messages_received,
            last_error: status.last_error,
        })
        .collect();

    SonicJson(VolOracleStatusesResponse { statuses })
}

/// List mark/spot price oracle health and freshness by underlying.
#[utoipa::path(
    get,
    path = "/monitoring/price-oracles",
    responses(
        (status = 200, description = "Price oracle status", body = PriceOracleStatusesResponse),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn price_oracles(State(app_state): State<AdminState>) -> impl IntoResponse {
    let spot_prices = app_state.greeks_cache.get_all_spot_prices_snapshot().await;
    let prev_day_prices = app_state
        .greeks_cache
        .get_all_prev_day_prices_snapshot()
        .await;
    let staleness = app_state.greeks_cache.get_spot_price_staleness().await;
    let unhealthy = app_state.greeks_cache.get_unhealthy_oracles().await;

    let statuses = price_oracle_statuses_from_snapshots(
        app_state.greeks_cache.get_configured_underlyings(),
        spot_prices,
        prev_day_prices,
        staleness,
        unhealthy,
    );

    SonicJson(PriceOracleStatusesResponse { statuses })
}

fn price_oracle_statuses_from_snapshots(
    configured_underlyings: Vec<String>,
    spot_prices: HashMap<String, f64>,
    prev_day_prices: HashMap<String, f64>,
    staleness: HashMap<String, Option<f64>>,
    unhealthy: Vec<String>,
) -> Vec<PriceOracleStatusResponse> {
    let mut statuses: Vec<PriceOracleStatusResponse> = configured_underlyings
        .into_iter()
        .map(|underlying| {
            let spot_price = spot_prices.get(&underlying).copied();
            let is_unhealthy = unhealthy.contains(&underlying);
            PriceOracleStatusResponse {
                prev_day_price: prev_day_prices.get(&underlying).copied(),
                staleness_seconds: staleness.get(&underlying).copied().flatten(),
                connected: !is_unhealthy,
                ready: spot_price.is_some() && !is_unhealthy,
                spot_price,
                underlying,
            }
        })
        .collect();

    statuses.sort_by(|a, b| a.underlying.cmp(&b.underlying));
    statuses
}

// --- Vol Surface Visualization Handlers ---

#[derive(Debug, Deserialize, utoipa::IntoParams)]
pub struct VolSurfaceQuery {
    /// Underlying asset (e.g. "BTC", "ETH"). If omitted, returns all.
    pub underlying: Option<String>,
}

#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct VolSurfaceApiResponse {
    pub surfaces: Vec<hypercall_vol_oracle::VolSurfaceSnapshot>,
}

/// GET /monitoring/vol-surface - Live vol surface snapshot(s).
#[utoipa::path(
    get,
    path = "/monitoring/vol-surface",
    params(VolSurfaceQuery),
    responses(
        (status = 200, description = "Vol surface snapshot", body = VolSurfaceApiResponse),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn get_vol_surface(
    State(app_state): State<AdminState>,
    axum::extract::Query(query): axum::extract::Query<VolSurfaceQuery>,
) -> impl IntoResponse {
    use std::collections::HashSet;

    let statuses = app_state.risk_vol_oracle.statuses();
    let underlyings: Vec<String> = if let Some(ref u) = query.underlying {
        vec![u.clone()]
    } else {
        let mut seen = HashSet::new();
        statuses
            .iter()
            .filter(|s| seen.insert(s.underlying.clone()))
            .map(|s| s.underlying.clone())
            .collect()
    };

    let surfaces: Vec<hypercall_vol_oracle::VolSurfaceSnapshot> = underlyings
        .iter()
        .filter_map(|u| app_state.risk_vol_oracle.get_surface_snapshot(u))
        .collect();

    SonicJson(VolSurfaceApiResponse { surfaces })
}

#[derive(Debug, Deserialize, utoipa::IntoParams)]
pub struct VolSurfaceHistoryQuery {
    pub underlying: String,
    /// Interval in milliseconds (300000 = 5m, 3600000 = 1h, 86400000 = 1d).
    pub interval_ms: i64,
    /// Maximum number of points to return.
    #[serde(default = "default_history_limit")]
    pub limit: usize,
}

fn default_history_limit() -> usize {
    500
}

#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct VolSurfaceHistoryResponse {
    pub underlying: String,
    pub interval_ms: i64,
    pub points: Vec<VolSurfaceHistoryPoint>,
}

#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct VolSurfaceHistoryPoint {
    pub timestamp_ms: i64,
    pub surface_json: serde_json::Value,
}

/// GET /monitoring/vol-surface/history - Historical vol surface snapshots.
#[utoipa::path(
    get,
    path = "/monitoring/vol-surface/history",
    params(VolSurfaceHistoryQuery),
    responses(
        (status = 200, description = "Vol surface history", body = VolSurfaceHistoryResponse),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn get_vol_surface_history(
    State(app_state): State<AdminState>,
    axum::extract::Query(query): axum::extract::Query<VolSurfaceHistoryQuery>,
) -> impl IntoResponse {
    let rows = match app_state
        .db
        .get_vol_surface_history(&query.underlying, query.interval_ms, query.limit)
        .await
    {
        Ok(rows) => rows,
        Err(err) => {
            warn!("Vol surface history query failed: {err:#}");
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                SonicJson(json!({ "error": err.to_string() })),
            )
                .into_response();
        }
    };

    let points: Vec<VolSurfaceHistoryPoint> = rows
        .into_iter()
        .map(|row| VolSurfaceHistoryPoint {
            timestamp_ms: row.timestamp_ms,
            surface_json: row.surface_json,
        })
        .collect();

    SonicJson(VolSurfaceHistoryResponse {
        underlying: query.underlying,
        interval_ms: query.interval_ms,
        points,
    })
    .into_response()
}

#[derive(Debug, Deserialize, utoipa::IntoParams)]
pub struct VolSurfaceSymbolQuery {
    /// Option symbol, e.g. "BTC-20260530-100000-C".
    pub symbol: String,
    /// Interval in milliseconds (300000 = 5m, 3600000 = 1h, 86400000 = 1d).
    #[serde(default = "default_symbol_interval_ms")]
    pub interval_ms: i64,
    /// Maximum number of points to return.
    #[serde(default = "default_history_limit")]
    pub limit: usize,
}

fn default_symbol_interval_ms() -> i64 {
    hypercall_types::HISTORICAL_THEO_INTERVAL_5M_MS
}

#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct VolSurfaceSymbolResponse {
    pub symbol: String,
    pub underlying: String,
    pub strike: f64,
    pub expiry: i64,
    pub interval_ms: i64,
    pub points: Vec<SymbolIvPoint>,
}

#[derive(Debug, Serialize, utoipa::ToSchema)]
pub struct SymbolIvPoint {
    pub timestamp_ms: i64,
    pub iv: Option<f64>,
}

/// GET /monitoring/vol-surface/symbol - IV time series for a single option symbol.
///
/// Reads persisted `vol_surface_snapshots` (not the live oracle) and extracts the
/// IV for the requested (strike, expiry) at each captured timestamp.
#[utoipa::path(
    get,
    path = "/monitoring/vol-surface/symbol",
    params(VolSurfaceSymbolQuery),
    responses(
        (status = 200, description = "Symbol IV time series", body = VolSurfaceSymbolResponse),
        (status = 400, description = "Invalid symbol or interval"),
        (status = 401, description = "Invalid or missing X-Admin-Key header")
    ),
    tag = "Monitoring",
    security(("admin_key" = []))
)]
pub async fn get_vol_surface_symbol(
    State(app_state): State<AdminState>,
    axum::extract::Query(query): axum::extract::Query<VolSurfaceSymbolQuery>,
) -> impl IntoResponse {
    use hypercall_types::ParsedOptionSymbol;
    use rust_decimal::prelude::ToPrimitive;

    let parsed = match ParsedOptionSymbol::from_symbol(&query.symbol) {
        Ok(p) => p,
        Err(reason) => {
            return (
                StatusCode::BAD_REQUEST,
                SonicJson(
                    json!({ "error": format!("Invalid symbol {}: {}", query.symbol, reason) }),
                ),
            )
                .into_response();
        }
    };

    let expiry_ts_secs =
        hypercall_types::expiry_date_to_timestamp(&parsed.underlying, parsed.expiry);
    // Never fall back to a made-up strike: a wrong strike would silently match
    // the wrong vol surface point. Propagate the conversion failure instead.
    let strike = match parsed.strike.to_f64() {
        Some(strike) => strike,
        None => {
            return (
                StatusCode::BAD_REQUEST,
                SonicJson(json!({
                    "error": format!("Strike {} cannot be represented as f64", parsed.strike)
                })),
            )
                .into_response();
        }
    };

    let rows = match app_state
        .db
        .get_vol_surface_history(&parsed.underlying, query.interval_ms, query.limit)
        .await
    {
        Ok(rows) => rows,
        Err(err) => {
            warn!("Vol surface symbol query failed: {err:#}");
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                SonicJson(json!({ "error": err.to_string() })),
            )
                .into_response();
        }
    };

    // Match strike to within $0.5 — vol surface stores raw f64 strikes that may
    // differ slightly from the integer strike encoded in the symbol.
    const STRIKE_TOL: f64 = 0.5;

    let points: Vec<SymbolIvPoint> = rows
        .into_iter()
        .map(|row| {
            let iv = serde_json::from_value::<hypercall_vol_oracle::VolSurfaceSnapshot>(
                row.surface_json,
            )
            .ok()
            .and_then(|snap| {
                snap.strike_points.into_iter().find_map(|p| {
                    if p.expiry == expiry_ts_secs && (p.strike - strike).abs() < STRIKE_TOL {
                        Some(p.iv)
                    } else {
                        None
                    }
                })
            });
            SymbolIvPoint {
                timestamp_ms: row.timestamp_ms,
                iv,
            }
        })
        .collect();

    SonicJson(VolSurfaceSymbolResponse {
        symbol: query.symbol,
        underlying: parsed.underlying,
        strike,
        expiry: expiry_ts_secs,
        interval_ms: query.interval_ms,
        points,
    })
    .into_response()
}

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

    #[test]
    fn price_oracle_statuses_are_sorted_and_mark_unready_or_unhealthy() {
        let statuses = price_oracle_statuses_from_snapshots(
            vec!["ETH".to_string(), "BTC".to_string(), "HYPE".to_string()],
            HashMap::from([("BTC".to_string(), 100_000.0), ("ETH".to_string(), 3_000.0)]),
            HashMap::from([("BTC".to_string(), 99_000.0)]),
            HashMap::from([("BTC".to_string(), Some(12.5)), ("ETH".to_string(), None)]),
            vec!["ETH".to_string()],
        );

        assert_eq!(
            statuses
                .iter()
                .map(|status| status.underlying.as_str())
                .collect::<Vec<_>>(),
            vec!["BTC", "ETH", "HYPE"]
        );

        let btc = &statuses[0];
        assert!(btc.connected);
        assert!(btc.ready);
        assert_eq!(btc.spot_price, Some(100_000.0));
        assert_eq!(btc.prev_day_price, Some(99_000.0));
        assert_eq!(btc.staleness_seconds, Some(12.5));

        let eth = &statuses[1];
        assert!(!eth.connected);
        assert!(!eth.ready);
        assert_eq!(eth.spot_price, Some(3_000.0));
        assert_eq!(eth.staleness_seconds, None);

        let hype = &statuses[2];
        assert!(hype.connected);
        assert!(!hype.ready);
        assert_eq!(hype.spot_price, None);
    }
}