hypercall_vol_oracle/

polygon_oracle.rs

// TODO(clients): Move Polygon HTTP transport and wire models into a dedicated client crate.

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::{Datelike, NaiveDate, Utc};
use metrics::{counter, gauge};
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_POLYGON_BASE_URL: &str = "https://api.polygon.io";

#[derive(Debug, Clone)]
pub struct PolygonVolOracleConfig {
    pub api_key: String,
    pub base_url: String,
    pub poll_interval: Duration,
    pub staleness_threshold: Duration,
    pub underlyings: HashMap<String, PolygonUnderlyingConfig>,
}

#[derive(Debug, Clone)]
pub struct PolygonUnderlyingConfig {
    pub ticker: String,
    /// Static scale for local/test routes. Production routes should require live dynamic scaling.
    pub strike_scale: f64,
    pub require_live_strike_scale: bool,
}

/// Shared handle for injecting platform spot prices into the Polygon oracle.
/// The oracle uses these to compute a dynamic strike scale at each refresh.
pub type PlatformSpotPrices = Arc<RwLock<HashMap<String, f64>>>;

#[derive(Debug, Default)]
struct PolygonSurfaceState {
    surfaces: HashMap<String, VolatilitySurface>,
    connected: HashMap<String, bool>,
    last_update_ts_ms: HashMap<String, i64>,
    last_error: HashMap<String, String>,
    /// ETF spot prices extracted from the most recent Polygon response.
    etf_spots: HashMap<String, f64>,
    /// Platform spot used when the current scaled surface was built.
    surface_spots: HashMap<String, f64>,
    /// Last successfully computed dynamic strike scale per underlying.
    /// Exposed for diagnostics only. Production live refreshes still require
    /// current platform and ETF spots when `require_live_strike_scale` is set.
    last_dynamic_scale: HashMap<String, f64>,
}

#[derive(Debug, Deserialize)]
struct PolygonResponse {
    #[serde(default)]
    status: Option<String>,
    #[serde(default)]
    results: Option<Vec<PolygonSnapshot>>,
    #[serde(default)]
    next_url: Option<String>,
}

#[derive(Debug, Deserialize)]
struct PolygonSnapshot {
    #[serde(default)]
    details: Option<PolygonDetails>,
    #[serde(default)]
    implied_volatility: Option<f64>,
    #[serde(default)]
    underlying_asset: Option<PolygonUnderlyingAsset>,
    #[serde(default)]
    last_quote: Option<PolygonMarketTimestamp>,
    #[serde(default)]
    last_trade: Option<PolygonMarketTimestamp>,
    #[serde(default)]
    fmv_last_updated: Option<i64>,
}

#[derive(Debug, Deserialize)]
struct PolygonDetails {
    #[serde(default)]
    strike_price: Option<f64>,
    #[serde(default)]
    expiration_date: Option<String>,
}

#[derive(Debug, Deserialize)]
struct PolygonUnderlyingAsset {
    #[serde(default)]
    price: Option<f64>,
}

#[derive(Debug, Deserialize)]
struct PolygonMarketTimestamp {
    #[serde(default)]
    last_updated: Option<i64>,
    #[serde(default)]
    sip_timestamp: Option<i64>,
    #[serde(default)]
    participant_timestamp: Option<i64>,
}

impl PolygonSnapshot {
    fn source_update_ts_ms(&self) -> Option<i64> {
        [
            self.fmv_last_updated.and_then(polygon_timestamp_to_ms),
            self.last_quote
                .as_ref()
                .and_then(PolygonMarketTimestamp::source_update_ts_ms),
            self.last_trade
                .as_ref()
                .and_then(PolygonMarketTimestamp::source_update_ts_ms),
        ]
        .into_iter()
        .flatten()
        .max()
    }
}

impl PolygonMarketTimestamp {
    fn source_update_ts_ms(&self) -> Option<i64> {
        [
            self.last_updated,
            self.sip_timestamp,
            self.participant_timestamp,
        ]
        .into_iter()
        .flatten()
        .filter_map(polygon_timestamp_to_ms)
        .max()
    }
}

fn polygon_timestamp_to_ms(timestamp: i64) -> Option<i64> {
    if timestamp <= 0 {
        return None;
    }

    // Polygon option snapshots report market-data timestamps in nanoseconds.
    // Be permissive for fixtures or future fields that may already be millis.
    if timestamp >= 1_000_000_000_000_000 {
        Some(timestamp / 1_000_000)
    } else if timestamp >= 1_000_000_000_000 {
        Some(timestamp)
    } else {
        None
    }
}

pub struct PolygonVolOracle {
    client: Client,
    config: PolygonVolOracleConfig,
    state: Arc<RwLock<PolygonSurfaceState>>,
    messages_received: AtomicU64,
    /// Platform spot prices injected externally (e.g. from GreeksCache).
    /// Used to compute dynamic strike_scale = platform_spot / etf_spot.
    platform_spots: PlatformSpotPrices,
}

impl PolygonVolOracle {
    pub fn new(config: PolygonVolOracleConfig, platform_spots: PlatformSpotPrices) -> Self {
        Self {
            client: Client::new(),
            config,
            state: Arc::new(RwLock::new(PolygonSurfaceState::default())),
            messages_received: AtomicU64::new(0),
            platform_spots,
        }
    }

    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!("Polygon vol oracle refresh failed: {err:#}");
                }
            }
        })
    }

    /// Compute the strike scale for an underlying.
    ///
    /// Priority: live dynamic ratio > cached last-good dynamic > static config.
    /// The last-good dynamic scale is stored in state so that transient spot
    /// price gaps after warm-up don't silently revert to a stale config value.
    fn compute_strike_scale(
        &self,
        underlying: &str,
        etf_spot: Option<f64>,
        config: &PolygonUnderlyingConfig,
    ) -> Result<(f64, Option<f64>)> {
        let platform_spot = self
            .platform_spots
            .read()
            .expect("platform_spots poisoned")
            .get(underlying)
            .copied();

        if let (Some(ps), Some(es)) = (platform_spot, etf_spot) {
            if ps > 0.0 && es > 0.0 {
                let dynamic = ps / es;
                debug!(
                    underlying,
                    platform_spot = ps,
                    etf_spot = es,
                    dynamic_scale = dynamic,
                    "Using dynamic strike scale"
                );
                // Cache for reuse during transient gaps.
                self.state
                    .write()
                    .expect("polygon vol oracle state poisoned")
                    .last_dynamic_scale
                    .insert(underlying.to_string(), dynamic);
                return Ok((dynamic, Some(ps)));
            }
        }

        if config.require_live_strike_scale {
            anyhow::bail!(
                "Missing live strike scale inputs for {}: platform_spot={:?}, etf_spot={:?}",
                underlying,
                platform_spot,
                etf_spot
            );
        }

        // Prefer last-good dynamic over static config.
        let cached = self
            .state
            .read()
            .expect("polygon vol oracle state poisoned")
            .last_dynamic_scale
            .get(underlying)
            .copied();

        if let Some(scale) = cached {
            debug!(
                underlying,
                cached_scale = scale,
                "Using cached dynamic strike scale (live spots unavailable)"
            );
            return Ok((scale, None));
        }

        debug!(
            underlying,
            fallback_scale = config.strike_scale,
            "Using static fallback strike scale (no dynamic history)"
        );
        Ok((config.strike_scale, None))
    }

    async fn refresh_all(&self) -> Result<()> {
        let underlyings = self
            .config
            .underlyings
            .iter()
            .map(|(underlying, config)| (underlying.clone(), config.clone()))
            .collect::<Vec<_>>();

        for (underlying, config) in underlyings {
            match self.fetch_surface(&underlying, &config).await {
                Ok((surface, etf_spot, platform_spot, source_update_ts_ms)) => {
                    let point_count = surface.len();
                    {
                        let mut state = self
                            .state
                            .write()
                            .expect("polygon vol oracle state poisoned");
                        state.connected.insert(underlying.clone(), true);
                        state
                            .last_update_ts_ms
                            .insert(underlying.clone(), source_update_ts_ms);
                        state.last_error.remove(&underlying);
                        if let Some(spot) = etf_spot {
                            state.etf_spots.insert(underlying.clone(), spot);
                        }
                        if let Some(spot) = platform_spot {
                            state.surface_spots.insert(underlying.clone(), spot);
                        } else {
                            state.surface_spots.remove(&underlying);
                        }
                        state.surfaces.insert(underlying.clone(), surface);
                    }
                    self.messages_received.fetch_add(1, Ordering::Relaxed);
                    counter!(
                        "ht_vol_oracle_messages_received_total",
                        "provider" => VolProviderKind::Polygon.as_str(),
                        "underlying" => underlying.clone()
                    )
                    .increment(1);

                    // Vol sanity check: warn if ATM vol looks unreasonable.
                    self.check_atm_vol_sanity(&underlying);

                    info!(
                        "Updated Polygon vol surface for {} with {} points",
                        underlying, point_count
                    );
                }
                Err(err) => {
                    let message = err.to_string();
                    let has_existing = self
                        .state
                        .read()
                        .expect("polygon vol oracle state poisoned")
                        .surfaces
                        .contains_key(&underlying);
                    if has_existing {
                        warn!(
                            "Polygon vol surface refresh failed for {} (keeping last good data): {}",
                            underlying, message
                        );
                    } else {
                        warn!(
                            "Polygon vol surface refresh failed for {}: {}",
                            underlying, message
                        );
                        let mut state = self
                            .state
                            .write()
                            .expect("polygon vol oracle state poisoned");
                        state.connected.insert(underlying.clone(), false);
                    }
                    let mut state = self
                        .state
                        .write()
                        .expect("polygon vol oracle state poisoned");
                    state.last_error.insert(underlying, message);
                }
            }
        }

        Ok(())
    }

    /// Fetch vol surface from Polygon, returning the surface and the ETF spot price.
    async fn fetch_surface(
        &self,
        underlying: &str,
        config: &PolygonUnderlyingConfig,
    ) -> Result<(VolatilitySurface, Option<f64>, Option<f64>, i64)> {
        let mut raw_points: Vec<(f64, i64, f64, i64)> = Vec::new();
        let mut etf_spot: Option<f64> = None;
        let mut source_update_ts_ms: Option<i64> = None;
        let mut url = format!(
            "{}/v3/snapshot/options/{}?limit=250&apiKey={}",
            self.config.base_url, config.ticker, self.config.api_key
        );
        let now_ts = Utc::now().timestamp();
        let mut page = 0_u8;

        while page < 6 {
            page += 1;
            let response = self
                .client
                .get(&url)
                .send()
                .await
                .with_context(|| format!("Failed to fetch Polygon page {page} for {underlying}"))?
                .error_for_status()
                .with_context(|| {
                    format!("Polygon returned non-success status on page {page} for {underlying}")
                })?
                .json::<PolygonResponse>()
                .await
                .with_context(|| {
                    format!("Failed to decode Polygon response on page {page} for {underlying}")
                })?;

            if response.status.as_deref() != Some("OK") {
                anyhow::bail!(
                    "Polygon response status was {:?} on page {} for {}",
                    response.status,
                    page,
                    underlying
                );
            }

            for snapshot in response.results.unwrap_or_default() {
                // Extract ETF spot from the first snapshot that has it.
                if etf_spot.is_none() {
                    if let Some(ref ua) = snapshot.underlying_asset {
                        if let Some(price) = ua.price {
                            if price > 0.0 {
                                etf_spot = Some(price);
                            }
                        }
                    }
                }

                let Some(details) = snapshot.details.as_ref() else {
                    continue;
                };
                let Some(strike) = details.strike_price else {
                    continue;
                };
                let Some(expiration_date) = details.expiration_date.as_deref() else {
                    continue;
                };
                let Some(iv) = snapshot.implied_volatility else {
                    continue;
                };
                if !(0.01..5.0).contains(&iv) {
                    continue;
                }
                let expiry_date = NaiveDate::parse_from_str(&expiration_date, "%Y-%m-%d")
                    .with_context(|| {
                        format!("Invalid Polygon expiry {expiration_date} for {underlying}")
                    })?;
                let expiry_code = expiry_date.year() as u64 * 10_000
                    + expiry_date.month() as u64 * 100
                    + expiry_date.day() as u64;
                let expiry_ts =
                    hypercall_types::expiry_date_to_timestamp_checked(underlying, expiry_code)
                        .with_context(|| {
                            format!("Invalid Polygon expiry {expiration_date} for {underlying}")
                        })? as i64;
                if expiry_ts <= now_ts {
                    continue;
                }
                let Some(point_update_ts_ms) = snapshot.source_update_ts_ms() else {
                    continue;
                };
                source_update_ts_ms = Some(
                    source_update_ts_ms
                        .unwrap_or(point_update_ts_ms)
                        .max(point_update_ts_ms),
                );

                raw_points.push((strike, expiry_ts, iv, point_update_ts_ms));
            }

            let Some(next_url) = response.next_url else {
                break;
            };
            url = if next_url.contains("apiKey=") {
                next_url
            } else {
                format!("{}&apiKey={}", next_url, self.config.api_key)
            };
        }

        if raw_points.is_empty() {
            anyhow::bail!(
                "Polygon returned no usable surface points for {underlying} (market may be closed)"
            );
        }
        let source_update_ts_ms = source_update_ts_ms.ok_or_else(|| {
            anyhow::anyhow!(
                "Polygon returned no source market timestamps for usable surface points for {underlying}"
            )
        })?;

        // Compute the strike scale now that we know the ETF spot.
        let (scale, platform_spot_at_build) =
            self.compute_strike_scale(underlying, etf_spot, config)?;

        let mut surface = VolatilitySurface::new();
        for (strike, expiry_ts, iv, _point_update_ts_ms) in raw_points {
            surface.insert(strike * scale, expiry_ts, iv);
        }

        // Polygon's per-instrument IVs are solved from exchange mids and can
        // carry cross-model artifacts on the OTM wings (e.g. a jump/tail fit
        // that makes a higher-strike call more expensive than a lower-strike
        // call in the same expiry). Enforce static no-arbitrage in the surface
        // itself — monotone and convex call prices per expiry — before anyone
        // reads from it. Requires platform spot (surface strikes are already in
        // platform space via `scale`).
        let platform_spot = self
            .platform_spots
            .read()
            .expect("platform_spots poisoned")
            .get(underlying)
            .copied();
        if let Some(spot) = platform_spot {
            let clamps = surface.sanitize_arb_free(spot, 0.0);
            if clamps > 0 {
                counter!(
                    "ht_vol_surface_arb_clamps_total",
                    "provider" => VolProviderKind::Polygon.as_str(),
                    "underlying" => underlying.to_string()
                )
                .increment(clamps as u64);
                info!(
                    underlying,
                    clamps, "Clamped non-arb-free vol surface points (polygon)"
                );
            }
        }

        Ok((
            surface,
            etf_spot,
            platform_spot_at_build,
            source_update_ts_ms,
        ))
    }

    /// Emit a warning and metric if the ATM vol for an underlying looks unreasonable.
    fn check_atm_vol_sanity(&self, underlying: &str) {
        let atm_gauge = gauge!(
            "ht_vol_oracle_atm_iv",
            "provider" => VolProviderKind::Polygon.as_str(),
            "underlying" => underlying.to_string()
        );

        let state = self
            .state
            .read()
            .expect("polygon vol oracle state poisoned");
        let Some(surface) = state.surfaces.get(underlying) else {
            atm_gauge.set(-1.0);
            return;
        };

        let now_ts = Utc::now().timestamp();
        let nearest_expiry = surface.expiries().iter().find(|&&e| e > now_ts).copied();
        let Some(expiry) = nearest_expiry else {
            atm_gauge.set(-1.0);
            return;
        };

        let platform_spot = self
            .platform_spots
            .read()
            .expect("platform_spots poisoned")
            .get(underlying)
            .copied();
        let Some(spot) = platform_spot else {
            atm_gauge.set(-1.0);
            return;
        };

        if let Some(atm_iv) = surface.get_interpolated(spot, expiry) {
            atm_gauge.set(atm_iv);

            if atm_iv > 2.0 {
                warn!(
                    underlying,
                    atm_iv,
                    expiry,
                    "Polygon ATM vol exceeds 200% — possible strike scale mismatch or stale data"
                );
            }
        }
    }

    fn status_for(&self, underlying: &str) -> VolOracleStatus {
        let state = self
            .state
            .read()
            .expect("polygon vol oracle state poisoned");
        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);
        let surface_points = state
            .surfaces
            .get(underlying)
            .map(VolatilitySurface::len)
            .unwrap_or(0);

        VolOracleStatus {
            underlying: underlying.to_string(),
            provider: VolProviderKind::Polygon,
            route_facing: true,
            connected: state.connected.get(underlying).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(underlying).cloned(),
        }
    }
}

impl RiskVolOracle for PolygonVolOracle {
    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::Polygon,
                reason: status
                    .last_error
                    .unwrap_or_else(|| "not connected".to_string()),
            });
        }

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

        let state = self
            .state
            .read()
            .expect("polygon vol oracle state poisoned");
        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::Polygon,
                strike,
                expiry_ts,
            })?;

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

        Ok(iv)
    }

    fn statuses(&self) -> Vec<VolOracleStatus> {
        self.config
            .underlyings
            .keys()
            .map(|underlying| self.status_for(underlying))
            .collect()
    }

    fn get_surface_snapshot(&self, underlying: &str) -> Option<VolSurfaceSnapshot> {
        let state = self
            .state
            .read()
            .expect("polygon 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(),
            // Return the platform spot from the refresh that produced this
            // scaled surface. Reading the live spot here would erase the
            // moneyness shift needed by closed-session transforms.
            spot_price: state.surface_spots.get(underlying).copied(),
        })
    }

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

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

    #[test]
    fn source_update_ts_uses_market_data_timestamps() {
        let snapshot = PolygonSnapshot {
            details: None,
            implied_volatility: None,
            underlying_asset: None,
            last_quote: Some(PolygonMarketTimestamp {
                last_updated: Some(1_700_000_000_123_456_789),
                sip_timestamp: None,
                participant_timestamp: None,
            }),
            last_trade: Some(PolygonMarketTimestamp {
                last_updated: None,
                sip_timestamp: Some(1_700_000_001_000_000_000),
                participant_timestamp: None,
            }),
            fmv_last_updated: Some(1_699_999_999_000_000_000),
        };

        assert_eq!(snapshot.source_update_ts_ms(), Some(1_700_000_001_000));
    }

    #[test]
    fn source_update_ts_rejects_missing_or_invalid_timestamps() {
        let snapshot = PolygonSnapshot {
            details: None,
            implied_volatility: None,
            underlying_asset: None,
            last_quote: Some(PolygonMarketTimestamp {
                last_updated: Some(0),
                sip_timestamp: Some(-1),
                participant_timestamp: None,
            }),
            last_trade: None,
            fmv_last_updated: None,
        };

        assert_eq!(snapshot.source_update_ts_ms(), None);
    }

    #[test]
    fn snapshot_uses_surface_build_spot_not_current_platform_spot() {
        let platform_spots = Arc::new(RwLock::new(HashMap::from([("OIL".to_string(), 120.0)])));
        let oracle = PolygonVolOracle::new(
            PolygonVolOracleConfig {
                api_key: "test".to_string(),
                base_url: "http://127.0.0.1:9".to_string(),
                poll_interval: Duration::from_secs(60),
                staleness_threshold: Duration::from_secs(180),
                underlyings: HashMap::from([(
                    "OIL".to_string(),
                    PolygonUnderlyingConfig {
                        ticker: "OIL".to_string(),
                        strike_scale: 1.0,
                        require_live_strike_scale: true,
                    },
                )]),
            },
            platform_spots.clone(),
        );
        let expiry = Utc::now().timestamp() + 30 * 86_400;
        let mut surface = VolatilitySurface::new();
        surface.insert(100.0, expiry, 0.50);
        {
            let mut state = oracle
                .state
                .write()
                .expect("polygon vol oracle state poisoned");
            state.surfaces.insert("OIL".to_string(), surface);
            state
                .last_update_ts_ms
                .insert("OIL".to_string(), Utc::now().timestamp_millis());
            state.surface_spots.insert("OIL".to_string(), 100.0);
        }
        platform_spots
            .write()
            .expect("platform_spots poisoned")
            .insert("OIL".to_string(), 140.0);

        let snapshot = oracle
            .get_surface_snapshot("OIL")
            .expect("snapshot should exist");
        assert_eq!(snapshot.spot_price, Some(100.0));
    }
}