hypercall/price_oracle/

hyperliquid_oracle.rs

//! Hyperliquid-based Mark Price Oracle implementation.
//!
//! This module provides a production-grade oracle that:
//! - Fetches oracle/index prices from Hyperliquid's `metaAndAssetCtxs` endpoint
//! - Computes forward prices using `spot * e^(r * t)`
//! - Supports TWAP settlement with median-of-means and 5% trimming
//! - Persists samples to PostgreSQL for reproducibility

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use async_trait::async_trait;
use chrono::Utc;
use metrics::counter;
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

use super::hydromancer_client::HydromancerClient;
use super::hyperliquid_types::{MetaAndAssetCtxsRequest, MetaAndAssetCtxsResponse};
use super::hyperliquid_ws::HyperliquidWsFeed;
use crate::shared::traits::{MarkPriceOracle, TxHash, H256};
use crate::snapshot::SyncStatus;
use hypercall_db::OracleWriter;

/// Default risk-free rate (5%)
pub const DEFAULT_RISK_FREE_RATE: f64 = 0.05;

/// Default polling interval in milliseconds (2 seconds)
pub const DEFAULT_POLL_INTERVAL_MS: u64 = 2000;

/// Default TWAP window in seconds (30 minutes)
pub const DEFAULT_TWAP_WINDOW_SECONDS: u32 = 1800;

/// Default maximum samples to keep in memory per settlement
pub const DEFAULT_MAX_MEMORY_SAMPLES: usize = 2000;

/// Default minimum samples required for settlement (local or Hydromancer)
pub const DEFAULT_MIN_SETTLEMENT_SAMPLES: usize = 500;

/// Hard floor for MIN_SETTLEMENT_SAMPLES
const MIN_SETTLEMENT_SAMPLES_FLOOR: usize = 50;

/// TTL for cached Hydromancer failures before retry
const FALLBACK_FAILURE_TTL: Duration = Duration::from_secs(300);

/// Configuration for the Hyperliquid Mark Price Oracle.
#[derive(Clone)]
pub struct HyperliquidOracleConfig {
    /// Base API URL for Hyperliquid info endpoint (HTTP fallback)
    pub api_url: String,
    /// Polling interval in milliseconds
    pub poll_interval_ms: u64,
    /// Risk-free rate for forward price calculation (e.g., 0.05 = 5%)
    pub risk_free_rate: f64,
    /// Symbol to track (e.g., "BTC", "ETH")
    pub symbol: String,
    /// TWAP window duration in seconds
    pub twap_window_seconds: u32,
    /// Oracle persistence writer (optional, for DB persistence of samples/settlements)
    pub oracle_writer: Option<Arc<dyn OracleWriter>>,
    /// Maximum samples to keep in memory per settlement
    pub max_memory_samples: usize,
    /// Shared WebSocket price feed (preferred over HTTP polling when set)
    pub ws_feed: Option<Arc<HyperliquidWsFeed>>,
    /// Optional notifier fired whenever the spot price updates
    pub price_notify: Option<Arc<tokio::sync::Notify>>,
    /// Minimum number of samples required for settlement (local or Hydromancer)
    pub min_settlement_samples: usize,
}

impl std::fmt::Debug for HyperliquidOracleConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HyperliquidOracleConfig")
            .field("api_url", &self.api_url)
            .field("poll_interval_ms", &self.poll_interval_ms)
            .field("risk_free_rate", &self.risk_free_rate)
            .field("symbol", &self.symbol)
            .field("twap_window_seconds", &self.twap_window_seconds)
            .field("max_memory_samples", &self.max_memory_samples)
            .field("min_settlement_samples", &self.min_settlement_samples)
            .field("ws_feed", &self.ws_feed.is_some())
            .finish()
    }
}

impl Default for HyperliquidOracleConfig {
    fn default() -> Self {
        Self {
            api_url: "https://api.hyperliquid.xyz/info".to_string(),
            poll_interval_ms: DEFAULT_POLL_INTERVAL_MS,
            risk_free_rate: DEFAULT_RISK_FREE_RATE,
            // Symbol must be explicitly set - no default to prevent accidental misconfiguration
            symbol: String::new(),
            twap_window_seconds: DEFAULT_TWAP_WINDOW_SECONDS,
            oracle_writer: None,
            max_memory_samples: DEFAULT_MAX_MEMORY_SAMPLES,
            ws_feed: None,
            price_notify: None,
            min_settlement_samples: DEFAULT_MIN_SETTLEMENT_SAMPLES
                .max(MIN_SETTLEMENT_SAMPLES_FLOOR),
        }
    }
}

impl HyperliquidOracleConfig {
    pub fn from_pricing_config(config: &hypercall_config::PricingConfig) -> Self {
        Self {
            api_url: config.hyperliquid_info_url.clone(),
            poll_interval_ms: DEFAULT_POLL_INTERVAL_MS,
            risk_free_rate: DEFAULT_RISK_FREE_RATE,
            symbol: config.oracle_symbol.clone(),
            twap_window_seconds: DEFAULT_TWAP_WINDOW_SECONDS,
            oracle_writer: None,
            max_memory_samples: DEFAULT_MAX_MEMORY_SAMPLES,
            ws_feed: None,
            price_notify: None,
            min_settlement_samples: config
                .min_settlement_samples
                .max(MIN_SETTLEMENT_SAMPLES_FLOOR),
        }
    }

    /// Validate the configuration, returning an error if invalid.
    fn validate(&self) -> Result<()> {
        if self.symbol.is_empty() {
            anyhow::bail!(
                "HyperliquidOracleConfig: symbol must be specified (e.g., \"BTC\", \"ETH\")"
            );
        }
        Ok(())
    }
}

/// A single price sample for TWAP computation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PriceSample {
    /// Unix timestamp in milliseconds when sample was taken
    pub timestamp_ms: i64,
    /// Oracle/index price from Hyperliquid
    pub price: f64,
    /// Source identifier (for multi-source redundancy in future)
    pub source: String,
}

/// Settlement window tracking for TWAP computation.
#[derive(Debug, Clone)]
pub struct SettlementWindow {
    /// Expiry timestamp (Unix seconds)
    pub expiry_timestamp: i64,
    /// TWAP window starts at: expiry - twap_seconds
    pub window_start: i64,
    /// TWAP window ends at: expiry
    pub window_end: i64,
    /// Collected samples during window
    pub samples: Vec<PriceSample>,
    /// Finalized settlement price (None until computed)
    pub settlement_price: Option<f64>,
    /// Whether this settlement has been persisted to DB
    pub persisted: bool,
}

/// Result of an oracle price fetch from Hyperliquid.
pub struct OracleFetchResult {
    pub oracle_price: f64,
    pub prev_day_px: Option<f64>,
}

/// Current state of the oracle.
#[derive(Debug, Default)]
struct OracleState {
    /// Latest spot/oracle price
    current_spot: Option<f64>,
    /// Previous day's price (from Hyperliquid perp context)
    prev_day_px: Option<f64>,
    /// Last successful fetch timestamp (milliseconds)
    last_fetch_timestamp: Option<i64>,
    /// Active settlement windows (expiry_timestamp -> window)
    settlements: HashMap<i64, SettlementWindow>,
    /// Price sequence number (monotonic counter)
    price_seq: u64,
    /// Whether the oracle is healthy (receiving updates)
    is_healthy: bool,
    /// When true, WS/HTTP price updates are ignored (set by set_spot_price_for_testing)
    test_price_override: bool,
}

/// Cached result from a Hydromancer fallback attempt.
#[derive(Debug, Clone)]
enum CachedFallbackResult {
    /// Successful settlement price (cached permanently)
    Success(f64),
    /// Failed attempt (cached with TTL)
    Failure(Instant),
}

/// The main Hyperliquid Mark Price Oracle.
///
/// This oracle fetches oracle/index prices from Hyperliquid and computes
/// forward prices for options settlement.
pub struct HyperliquidMarkPriceOracle {
    config: HyperliquidOracleConfig,
    client: reqwest::Client,
    state: Arc<RwLock<OracleState>>,
    /// Sync status for readiness tracking
    sync_status: Arc<SyncStatus>,
    /// Hydromancer fallback client (None if not configured)
    hydromancer_client: Option<Arc<HydromancerClient>>,
    /// Cache of Hydromancer fallback results keyed by (symbol, expiry_timestamp)
    fallback_cache: Arc<RwLock<HashMap<(String, i64), CachedFallbackResult>>>,
}

impl HyperliquidMarkPriceOracle {
    /// Create a new oracle with the given configuration.
    ///
    /// Returns an error if the configuration is invalid (e.g., missing symbol).
    pub fn new(config: HyperliquidOracleConfig) -> Result<Self> {
        config.validate()?;

        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()
            .expect("Failed to create HTTP client");

        Ok(Self {
            config,
            client,
            state: Arc::new(RwLock::new(OracleState::default())),
            sync_status: Arc::new(SyncStatus::new()),
            hydromancer_client: None,
            fallback_cache: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Risk-free rate used by this oracle when converting spot to forward.
    pub fn risk_free_rate(&self) -> f64 {
        self.config.risk_free_rate
    }

    /// Create with async initialization (fetches initial price).
    ///
    /// If a WebSocket feed is configured, waits for it to deliver a price first.
    /// Falls back to HTTP if WS doesn't deliver within a few seconds.
    /// Sets sync status to Ready after successful initial price fetch.
    pub async fn new_with_init(config: HyperliquidOracleConfig) -> Result<Self> {
        let oracle = Self::new(config)?;

        // If WS feed is configured, wait for it to deliver a price (it may still be connecting)
        if let Some(ref ws_feed) = oracle.config.ws_feed {
            const WS_WAIT_ATTEMPTS: u32 = 15; // 15 * 500ms = 7.5s max wait
            for attempt in 0..WS_WAIT_ATTEMPTS {
                if let Some(price) = ws_feed.get_price(&oracle.config.symbol).await {
                    let prev_day_px = ws_feed.get_prev_day_price(&oracle.config.symbol).await;
                    let mut state = oracle.state.write().await;
                    state.current_spot = Some(price);
                    state.prev_day_px = prev_day_px;
                    state.last_fetch_timestamp = Some(Utc::now().timestamp_millis());
                    state.is_healthy = true;
                    drop(state);

                    oracle.sync_status.set_ready();
                    info!(
                        "Oracle initialized for {} via WebSocket: initial_price={} prev_day_px={:?}",
                        oracle.config.symbol, price, prev_day_px
                    );
                    return Ok(oracle);
                }
                if attempt < WS_WAIT_ATTEMPTS - 1 {
                    debug!(
                        "Oracle {} waiting for WS price ({}/{})",
                        oracle.config.symbol,
                        attempt + 1,
                        WS_WAIT_ATTEMPTS
                    );
                    tokio::time::sleep(Duration::from_millis(500)).await;
                }
            }
            warn!(
                "Oracle {} WS feed didn't deliver price in time, falling back to HTTP",
                oracle.config.symbol
            );
        }

        // Fallback: fetch via HTTP with exponential backoff
        const MAX_RETRIES: u32 = 5;
        const INITIAL_DELAY_MS: u64 = 1000;

        let mut last_error = None;
        for attempt in 0..MAX_RETRIES {
            match oracle.fetch_oracle_price().await {
                Ok(result) => {
                    {
                        let mut state = oracle.state.write().await;
                        state.current_spot = Some(result.oracle_price);
                        state.prev_day_px = result.prev_day_px;
                        state.last_fetch_timestamp = Some(Utc::now().timestamp_millis());
                        state.is_healthy = true;
                    }

                    oracle.sync_status.set_ready();

                    info!(
                        "Oracle initialized for {} via HTTP: initial_price={}{}",
                        oracle.config.symbol,
                        result.oracle_price,
                        if attempt > 0 {
                            format!(" (after {} retries)", attempt)
                        } else {
                            String::new()
                        }
                    );

                    return Ok(oracle);
                }
                Err(e) => {
                    // "Symbol not found in Hyperliquid universe" is a
                    // structural config error — the symbol isn't going to
                    // materialize on the next HTTP attempt. Retrying burns
                    // ~31 seconds of exponential backoff per missing
                    // symbol, which was starving the integration test
                    // harness startup budget (~80s total for USOIL + GOLD
                    // in testnet where neither symbol exists). Fail fast
                    // for this error class.
                    let err_text = e.to_string();
                    if err_text.contains("not found in Hyperliquid universe") {
                        warn!(
                            "Oracle {} symbol unknown to Hyperliquid universe, \
                             not retrying: {}",
                            oracle.config.symbol, e
                        );
                        return Err(e);
                    }
                    let delay_ms = INITIAL_DELAY_MS * 2u64.pow(attempt);
                    warn!(
                        "Oracle {} HTTP init attempt {}/{} failed: {}. Retrying in {}ms...",
                        oracle.config.symbol,
                        attempt + 1,
                        MAX_RETRIES,
                        e,
                        delay_ms
                    );
                    last_error = Some(e);
                    tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                }
            }
        }

        Err(last_error
            .unwrap_or_else(|| anyhow::anyhow!("Oracle init failed after {} retries", MAX_RETRIES)))
    }

    /// Get the sync status for readiness checks.
    pub fn sync_status(&self) -> Arc<SyncStatus> {
        self.sync_status.clone()
    }

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

    /// Check if the oracle is healthy (receiving updates).
    pub async fn is_healthy(&self) -> bool {
        let state = self.state.read().await;
        state.is_healthy
    }

    /// Get the last fetch timestamp in milliseconds (if any).
    /// Returns None if no price has been fetched yet.
    pub async fn get_last_fetch_timestamp_ms(&self) -> Option<i64> {
        let state = self.state.read().await;
        state.last_fetch_timestamp
    }

    /// Get the staleness of the price in seconds.
    /// Returns None if no price has been fetched yet.
    pub async fn get_staleness_seconds(&self) -> Option<f64> {
        let state = self.state.read().await;
        state.last_fetch_timestamp.map(|ts| {
            let now_ms = chrono::Utc::now().timestamp_millis();
            (now_ms - ts) as f64 / 1000.0
        })
    }

    /// Fetch oracle price (and prev day price) from Hyperliquid metaAndAssetCtxs endpoint.
    pub async fn fetch_oracle_price(&self) -> Result<OracleFetchResult> {
        let request = MetaAndAssetCtxsRequest::new();

        let response = self
            .client
            .post(&self.config.api_url)
            .json(&request)
            .send()
            .await
            .context("Failed to send request to Hyperliquid API")?;

        if !response.status().is_success() {
            let status = response.status();
            let error_text = response.text().await.unwrap_or_default();
            anyhow::bail!("API request failed with status {}: {}", status, error_text);
        }

        let data: MetaAndAssetCtxsResponse = response
            .json()
            .await
            .context("Failed to parse metaAndAssetCtxs response")?;

        // Find index of our symbol in universe
        let symbol_index = data
            .meta
            .universe
            .iter()
            .position(|asset| asset.name == self.config.symbol)
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "Symbol {} not found in Hyperliquid universe",
                    self.config.symbol
                )
            })?;

        // Get corresponding asset context
        let ctx = data.asset_ctxs.get(symbol_index).ok_or_else(|| {
            anyhow::anyhow!("Asset context not found for symbol {}", self.config.symbol)
        })?;

        let oracle_price = ctx.oracle_price().ok_or_else(|| {
            anyhow::anyhow!("Failed to parse oracle price for {}", self.config.symbol)
        })?;

        Ok(OracleFetchResult {
            oracle_price,
            prev_day_px: ctx.prev_day_price(),
        })
    }

    /// Start the background polling task (legacy, no shutdown support).
    ///
    /// Returns a JoinHandle that can be awaited or aborted.
    /// For graceful shutdown support, use `start_polling_with_shutdown` instead.
    pub fn start_polling(self: &Arc<Self>) -> JoinHandle<()> {
        // Create a dummy shutdown channel that never fires.
        // IMPORTANT: We must keep the sender alive, otherwise recv() returns Closed immediately.
        let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
        std::mem::forget(tx);
        self.start_polling_with_shutdown(rx)
    }

    /// Start the background polling task with shutdown support.
    ///
    /// Reads prices exclusively from the WebSocket feed. If the WS price is
    /// stale or unavailable (during reconnect), marks the oracle unhealthy and
    /// skips the tick — the WS feed's own reconnect loop handles recovery.
    /// Returns a JoinHandle that can be awaited or added to a TaskGroup.
    pub fn start_polling_with_shutdown(
        self: &Arc<Self>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> JoinHandle<()> {
        let oracle = Arc::clone(self);

        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(Duration::from_millis(oracle.config.poll_interval_ms));

            let ws_feed =
                oracle.config.ws_feed.as_ref().expect(
                    "Oracle polling requires a WebSocket feed — ws_feed must be set in config",
                );
            info!(
                "Oracle {} using WebSocket feed (WS-only, no HTTP fallback)",
                oracle.config.symbol
            );

            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("Oracle {} received shutdown signal", oracle.config.symbol);
                        break;
                    }
                    _ = interval.tick() => {
                        match ws_feed.get_price(&oracle.config.symbol).await {
                            Some(price) => {
                                let now_ms = Utc::now().timestamp_millis();
                                let now_secs = now_ms / 1000;
                                // Read prev_day_px before taking the state write lock
                                let prev_day_px = ws_feed.get_prev_day_price(&oracle.config.symbol).await;

                                let mut state = oracle.state.write().await;

                                // Skip price update if a test override is active
                                if state.test_price_override {
                                    drop(state);
                                    continue;
                                }

                                state.current_spot = Some(price);
                                if let Some(pdp) = prev_day_px {
                                    state.prev_day_px = Some(pdp);
                                }
                                state.last_fetch_timestamp = Some(now_ms);
                                state.price_seq += 1;
                                state.is_healthy = true;

                                debug!(
                                    "Oracle updated: {} = {} (seq={})",
                                    oracle.config.symbol, price, state.price_seq
                                );

                                // Process active settlement windows
                                oracle
                                    .process_settlement_samples(&mut state, price, now_ms, now_secs)
                                    .await;

                                if let Some(ref notify) = oracle.config.price_notify {
                                    notify.notify_waiters();
                                }
                            }
                            None => {
                                // WS price stale or unavailable — waiting for reconnect
                                let mut state = oracle.state.write().await;
                                if state.test_price_override {
                                    continue;
                                }
                                warn!(
                                    "Oracle {}: WS price unavailable, waiting for reconnect",
                                    oracle.config.symbol
                                );
                                state.is_healthy = false;
                            }
                        }
                    }
                }
            }
            info!("Oracle {} polling stopped", oracle.config.symbol);
        })
    }

    /// Process settlement samples for active windows.
    async fn process_settlement_samples(
        &self,
        state: &mut OracleState,
        price: f64,
        now_ms: i64,
        now_secs: i64,
    ) {
        let sample = PriceSample {
            timestamp_ms: now_ms,
            price,
            source: "hyperliquid".to_string(),
        };

        // Collect expiries that need TWAP computation
        let mut expiries_to_compute: Vec<i64> = Vec::new();

        for (&expiry, window) in state.settlements.iter_mut() {
            // Skip if already finalized
            if window.settlement_price.is_some() {
                continue;
            }

            // Check if within settlement window
            if now_secs >= window.window_start && now_secs <= window.window_end {
                // Within settlement window - collect sample
                window.samples.push(sample.clone());

                // Trim if exceeding memory limit
                if window.samples.len() > self.config.max_memory_samples {
                    let excess = window.samples.len() - self.config.max_memory_samples;
                    window.samples.drain(0..excess);
                }

                debug!(
                    "Collected settlement sample: expiry={}, samples={}",
                    expiry,
                    window.samples.len()
                );
            }

            // Check if window has ended - need to compute TWAP
            if now_secs > window.window_end && window.settlement_price.is_none() {
                expiries_to_compute.push(expiry);
            }
        }

        // Compute TWAP for windows that have ended
        let mut expiries_to_remove: Vec<i64> = Vec::new();

        for expiry in expiries_to_compute {
            if let Some(window) = state.settlements.get_mut(&expiry) {
                if !window.samples.is_empty() {
                    let twap = Self::compute_twap_median_of_means(&window.samples, 0.05);
                    window.settlement_price = Some(twap);

                    info!(
                        "Settlement computed: expiry={}, twap={}, samples={}",
                        expiry,
                        twap,
                        window.samples.len()
                    );

                    // Persist to database if configured
                    if let Some(ref writer) = self.config.oracle_writer {
                        let input = hypercall_db::NewOracleSettlementPriceInput {
                            symbol: self.config.symbol.clone(),
                            expiry_timestamp: window.expiry_timestamp,
                            settlement_price: twap,
                            sample_count: window.samples.len() as i32,
                            window_start: window.window_start,
                            window_end: window.window_end,
                            algorithm: "median_of_means_5pct_trim".to_string(),
                        };
                        if let Err(e) = writer.save_oracle_settlement_price_sync(&input) {
                            error!("Failed to persist settlement: {}", e);
                        } else {
                            window.persisted = true;
                            // Mark for removal from memory after successful persistence
                            expiries_to_remove.push(expiry);
                        }
                    }
                } else {
                    warn!("Settlement window ended with no samples: expiry={}", expiry);
                    // Also remove empty settlements to prevent memory accumulation
                    expiries_to_remove.push(expiry);
                }
            }
        }

        // Clean up persisted settlements from memory to prevent unbounded growth.
        // Settlement prices can be retrieved from the database via load_settlement_from_db.
        for expiry in expiries_to_remove {
            if let Some(window) = state.settlements.remove(&expiry) {
                debug!(
                    "Removed persisted settlement from memory: expiry={}, price={:?}",
                    expiry, window.settlement_price
                );
            }
        }
    }

    /// Compute forward price: spot * e^(r * t).
    ///
    /// - `spot`: Current spot/oracle price
    /// - `expiry_timestamp`: Unix timestamp (seconds) of expiry
    /// - `risk_free_rate`: Annual risk-free rate (e.g., 0.05 for 5%)
    ///
    /// Returns error if expiry is in the past.
    pub fn compute_forward_price(
        spot: f64,
        expiry_timestamp: i64,
        risk_free_rate: f64,
    ) -> Result<f64> {
        let now = Utc::now().timestamp();
        let time_to_expiry_secs = expiry_timestamp - now;

        if time_to_expiry_secs <= 0 {
            anyhow::bail!("Expiry {} is in the past (now={})", expiry_timestamp, now);
        }

        // Convert seconds to years (365.25 days to account for leap years)
        let time_to_expiry_years = time_to_expiry_secs as f64 / (365.25 * 24.0 * 60.0 * 60.0);

        // Forward = Spot * e^(r * t)
        let forward = spot * (risk_free_rate * time_to_expiry_years).exp();

        Ok(forward)
    }

    /// Compute TWAP using median-of-means with trimming.
    ///
    /// Algorithm:
    /// 1. Sort all samples by price
    /// 2. Trim `trim_pct` from each tail (removes outliers/manipulation)
    /// 3. Divide remaining samples into sqrt(n) buckets
    /// 4. Compute mean of each bucket
    /// 5. Return median of bucket means
    ///
    /// This is more robust than simple mean for manipulation resistance.
    pub fn compute_twap_median_of_means(samples: &[PriceSample], trim_pct: f64) -> f64 {
        if samples.is_empty() {
            return 0.0;
        }

        if samples.len() == 1 {
            return samples[0].price;
        }

        // Step 1: Extract and sort prices
        let mut prices: Vec<f64> = samples.iter().map(|s| s.price).collect();
        prices.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

        // Step 2: Apply trimming from each end
        let trim_count = (prices.len() as f64 * trim_pct) as usize;
        let trimmed = if trim_count * 2 >= prices.len() {
            // If trimming would remove all samples, just use the middle one
            &prices[prices.len() / 2..=prices.len() / 2]
        } else {
            &prices[trim_count..prices.len() - trim_count]
        };

        if trimmed.is_empty() {
            return prices[prices.len() / 2]; // Fallback to median
        }

        // Step 3: Divide into k buckets (sqrt(n) buckets is a common choice)
        let k = (trimmed.len() as f64).sqrt().ceil() as usize;
        let bucket_size = trimmed.len().div_ceil(k); // Ceiling division

        // Step 4: Compute mean of each bucket
        let mut bucket_means: Vec<f64> = Vec::with_capacity(k);

        for chunk in trimmed.chunks(bucket_size) {
            let sum: f64 = chunk.iter().sum();
            let mean = sum / chunk.len() as f64;
            bucket_means.push(mean);
        }

        // Step 5: Return median of bucket means
        bucket_means.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

        if bucket_means.len().is_multiple_of(2) && bucket_means.len() >= 2 {
            // Even number of buckets - average the two middle values
            let mid = bucket_means.len() / 2;
            (bucket_means[mid - 1] + bucket_means[mid]) / 2.0
        } else {
            // Odd number of buckets - take the middle value
            bucket_means[bucket_means.len() / 2]
        }
    }

    /// Persist price samples to database via the OracleWriter trait.
    fn persist_samples_via_writer(
        writer: &dyn OracleWriter,
        symbol: &str,
        expiry_timestamp: i64,
        samples: &[PriceSample],
    ) -> Result<()> {
        if samples.is_empty() {
            return Ok(());
        }

        let inputs: Vec<hypercall_db::NewOraclePriceSampleInput> = samples
            .iter()
            .map(|s| hypercall_db::NewOraclePriceSampleInput {
                symbol: symbol.to_string(),
                expiry_timestamp,
                sample_timestamp_ms: s.timestamp_ms,
                price: s.price,
                source: s.source.clone(),
            })
            .collect();

        writer.save_oracle_price_samples_sync(&inputs)?;

        debug!(
            "Persisted {} samples for symbol={}, expiry={}",
            samples.len(),
            symbol,
            expiry_timestamp
        );

        Ok(())
    }

    /// Set the Hydromancer fallback client.
    pub fn set_hydromancer_client(&mut self, client: Arc<HydromancerClient>) {
        self.hydromancer_client = Some(client);
    }

    /// Persist a Hydromancer-computed settlement price via the OracleWriter trait.
    fn persist_hydromancer_settlement_via_writer(
        writer: &dyn OracleWriter,
        symbol: &str,
        expiry_timestamp: i64,
        settlement_price: f64,
        sample_count: usize,
        window_start: i64,
        window_end: i64,
    ) -> Result<()> {
        let input = hypercall_db::NewOracleSettlementPriceInput {
            symbol: symbol.to_string(),
            expiry_timestamp,
            settlement_price,
            sample_count: sample_count as i32,
            window_start,
            window_end,
            algorithm: "hydromancer_median_of_means_5pct_trim".to_string(),
        };

        writer.save_oracle_settlement_price_sync(&input)?;

        info!(
            "Persisted Hydromancer settlement: symbol={}, expiry={}, price={}, samples={}",
            symbol, expiry_timestamp, settlement_price, sample_count
        );

        Ok(())
    }

    /// Attempt to settle an expiry using Hydromancer as a fallback oracle source.
    ///
    /// Returns Some(price) on success, None if Hydromancer cannot provide sufficient data.
    /// Results are cached to avoid repeated API calls.
    pub async fn attempt_hydromancer_fallback(&self, expiry_timestamp: i64) -> Option<f64> {
        let hydromancer = match &self.hydromancer_client {
            Some(c) => c.clone(),
            None => return None,
        };

        let symbol = &self.config.symbol;
        let cache_key = (symbol.clone(), expiry_timestamp);

        // Check cache first
        {
            let cache = self.fallback_cache.read().await;
            if let Some(cached) = cache.get(&cache_key) {
                match cached {
                    CachedFallbackResult::Success(price) => return Some(*price),
                    CachedFallbackResult::Failure(ts) => {
                        if ts.elapsed() < FALLBACK_FAILURE_TTL {
                            return None;
                        }
                        // TTL expired, will retry below
                    }
                }
            }
        }

        warn!(
            "Attempting Hydromancer fallback for symbol={}, expiry={}",
            symbol, expiry_timestamp
        );

        let twap_window = self.config.twap_window_seconds as i64;
        let window_start_ms = (expiry_timestamp - twap_window) * 1000;
        let window_end_ms = expiry_timestamp * 1000;

        // Fetch from Hydromancer
        let records = match hydromancer
            .fetch_oracle_price_history(symbol, window_start_ms, window_end_ms)
            .await
        {
            Ok(r) => r,
            Err(e) => {
                error!(
                    "Hydromancer API error for symbol={}, expiry={}: {}",
                    symbol, expiry_timestamp, e
                );
                counter!("ht_oracle_settlement_fallback_total", "underlying" => symbol.clone(), "result" => "api_error")
                    .increment(1);
                let mut cache = self.fallback_cache.write().await;
                cache.insert(cache_key, CachedFallbackResult::Failure(Instant::now()));
                return None;
            }
        };

        // Convert to PriceSamples with validation
        let samples =
            HydromancerClient::records_to_price_samples(records, window_start_ms, window_end_ms);

        // Validate sufficient samples
        if samples.len() < self.config.min_settlement_samples {
            warn!(
                "Hydromancer returned insufficient samples for symbol={}, expiry={}: {} < {}",
                symbol,
                expiry_timestamp,
                samples.len(),
                self.config.min_settlement_samples
            );
            counter!("ht_oracle_settlement_fallback_total", "underlying" => symbol.clone(), "result" => "insufficient_samples")
                .increment(1);
            let mut cache = self.fallback_cache.write().await;
            cache.insert(cache_key, CachedFallbackResult::Failure(Instant::now()));
            return None;
        }

        // Compute TWAP
        let settlement_price = Self::compute_twap_median_of_means(&samples, 0.05);
        let sample_count = samples.len();

        info!(
            "Hydromancer fallback computed settlement: symbol={}, expiry={}, price={}, samples={}",
            symbol, expiry_timestamp, settlement_price, sample_count
        );

        // Persist settlement and samples
        if let Some(ref writer) = self.config.oracle_writer {
            let window_start_secs = expiry_timestamp - twap_window;

            if let Err(e) = Self::persist_hydromancer_settlement_via_writer(
                writer.as_ref(),
                symbol,
                expiry_timestamp,
                settlement_price,
                sample_count,
                window_start_secs,
                expiry_timestamp,
            ) {
                error!(
                    "Failed to persist Hydromancer settlement for symbol={}, expiry={}: {}",
                    symbol, expiry_timestamp, e
                );
            }

            if let Err(e) = Self::persist_samples_via_writer(
                writer.as_ref(),
                symbol,
                expiry_timestamp,
                &samples,
            ) {
                error!(
                    "Failed to persist Hydromancer samples for symbol={}, expiry={}: {}",
                    symbol, expiry_timestamp, e
                );
            }
        }

        counter!("ht_oracle_settlement_fallback_total", "underlying" => symbol.clone(), "result" => "success")
            .increment(1);

        // Cache success
        let mut cache = self.fallback_cache.write().await;
        cache.insert(cache_key, CachedFallbackResult::Success(settlement_price));

        Some(settlement_price)
    }

    /// Set spot price directly (for testing only).
    ///
    /// This allows tests to inject spot prices without making network requests.
    /// ONLY use this in test environments.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn set_spot_price_for_testing(&self, price: f64) {
        let mut state = self.state.write().await;
        state.current_spot = Some(price);
        state.last_fetch_timestamp = Some(chrono::Utc::now().timestamp_millis());
        state.is_healthy = true;
        state.test_price_override = true;
        state.price_seq += 1;
        drop(state);
        self.sync_status.set_ready();
        if let Some(ref notify) = self.config.price_notify {
            notify.notify_waiters();
        }
        info!(
            "Test mode: Set spot price for {} to ${:.2} (WS/HTTP updates frozen)",
            self.config.symbol, price
        );
    }
}

#[async_trait]
impl MarkPriceOracle for HyperliquidMarkPriceOracle {
    async fn get_spot_price(&self) -> Option<f64> {
        let state = self.state.read().await;
        state.current_spot
    }

    async fn get_prev_day_price(&self) -> Option<f64> {
        let state = self.state.read().await;
        state.prev_day_px
    }

    async fn get_mark_price(&self, expiry_timestamp: i64) -> Result<f64> {
        let state = self.state.read().await;

        let spot = state
            .current_spot
            .ok_or_else(|| anyhow::anyhow!("No spot price available"))?;

        Self::compute_forward_price(spot, expiry_timestamp, self.config.risk_free_rate)
    }

    async fn register_settlement(&self, expiry_timestamp: i64, twap_seconds: u32) {
        let mut state = self.state.write().await;

        // Check if already registered
        if state.settlements.contains_key(&expiry_timestamp) {
            warn!(
                "Settlement already registered for expiry {}",
                expiry_timestamp
            );
            return;
        }

        let window_start = expiry_timestamp - twap_seconds as i64;

        let window = SettlementWindow {
            expiry_timestamp,
            window_start,
            window_end: expiry_timestamp,
            samples: Vec::with_capacity(twap_seconds as usize / 2), // ~2s intervals
            settlement_price: None,
            persisted: false,
        };

        state.settlements.insert(expiry_timestamp, window);

        info!(
            "Registered settlement: expiry={}, window_start={}, window_end={}",
            expiry_timestamp, window_start, expiry_timestamp
        );
    }

    async fn get_settlement_price(&self, expiry_timestamp: i64) -> Option<f64> {
        // First check in-memory cache
        {
            let state = self.state.read().await;
            if let Some(price) = state
                .settlements
                .get(&expiry_timestamp)
                .and_then(|w| w.settlement_price)
            {
                return Some(price);
            }
        }

        // Fall back to database lookup if not in memory
        if let Some(ref writer) = self.config.oracle_writer {
            match writer.get_oracle_settlement_price_sync(&self.config.symbol, expiry_timestamp) {
                Ok(Some(price)) => return Some(price),
                Ok(None) => {}
                Err(e) => {
                    warn!(
                        "Failed to load settlement from DB for expiry {}: {}",
                        expiry_timestamp, e
                    );
                }
            }
        }

        // Hydromancer fallback: if window has ended and client is configured
        let now = Utc::now().timestamp();
        if now > expiry_timestamp && self.hydromancer_client.is_some() {
            return self.attempt_hydromancer_fallback(expiry_timestamp).await;
        }

        None
    }

    async fn commit_price(&self, root: H256) -> Result<TxHash> {
        let state = self.state.read().await;
        let price_seq = state.price_seq;

        info!(
            "Committing price root: {} (seq={})",
            hex::encode(root),
            price_seq
        );

        // TODO: Integrate with transaction_submitter module for actual L1 submission
        // This would call something like:
        // self.tx_submitter.submit_price_root(root, price_seq).await

        Ok(format!("0x{}", hex::encode(root)))
    }

    async fn get_price_seq(&self) -> u64 {
        let state = self.state.read().await;
        state.price_seq
    }
}

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

    #[test]
    fn test_forward_price_calculation() {
        // Test with 1 year to expiry
        let spot = 100.0;
        let now = Utc::now().timestamp();
        let one_year_secs = 365.25 * 24.0 * 60.0 * 60.0;
        let expiry = now + one_year_secs as i64;
        let r = 0.05;

        let forward = HyperliquidMarkPriceOracle::compute_forward_price(spot, expiry, r).unwrap();

        // e^0.05 ≈ 1.05127
        let expected = spot * (r * 1.0).exp();
        assert!((forward - expected).abs() < 0.01);
    }

    #[test]
    fn test_forward_price_expired() {
        let spot = 100.0;
        let expiry = Utc::now().timestamp() - 1000; // In the past
        let r = 0.05;

        let result = HyperliquidMarkPriceOracle::compute_forward_price(spot, expiry, r);
        assert!(result.is_err());
    }

    #[test]
    fn test_twap_simple() {
        let samples = vec![
            PriceSample {
                timestamp_ms: 1000,
                price: 100.0,
                source: "test".to_string(),
            },
            PriceSample {
                timestamp_ms: 2000,
                price: 102.0,
                source: "test".to_string(),
            },
            PriceSample {
                timestamp_ms: 3000,
                price: 101.0,
                source: "test".to_string(),
            },
        ];

        let twap = HyperliquidMarkPriceOracle::compute_twap_median_of_means(&samples, 0.0);
        // With no trimming and 3 samples, should be close to mean
        assert!((twap - 101.0).abs() < 1.0);
    }

    #[test]
    fn test_twap_with_outlier() {
        let mut samples: Vec<PriceSample> = (0..100)
            .map(|i| PriceSample {
                timestamp_ms: i * 1000,
                price: 100.0 + (i as f64 * 0.1), // Gradual increase from 100 to ~110
                source: "test".to_string(),
            })
            .collect();

        // Add extreme outliers
        samples.push(PriceSample {
            timestamp_ms: 100000,
            price: 1000.0, // Extreme high
            source: "test".to_string(),
        });
        samples.push(PriceSample {
            timestamp_ms: 101000,
            price: 10.0, // Extreme low
            source: "test".to_string(),
        });

        let twap_trimmed = HyperliquidMarkPriceOracle::compute_twap_median_of_means(&samples, 0.05);
        let twap_untrimmed =
            HyperliquidMarkPriceOracle::compute_twap_median_of_means(&samples, 0.0);

        // True mean of 0-99 range is ~105 (100 + 0.1 * 49.5)
        // Trimmed TWAP should be close to the true mean
        // The exact behavior depends on bucket division, so we just verify:
        // 1. Trimmed TWAP is reasonable (between 100 and 115)
        // 2. Both algorithms produce valid results
        assert!(
            twap_trimmed > 100.0 && twap_trimmed < 115.0,
            "Trimmed TWAP {} should be in reasonable range",
            twap_trimmed
        );
        assert!(twap_untrimmed > 0.0, "Untrimmed TWAP should be positive");
        println!(
            "TWAP test: trimmed={:.2}, untrimmed={:.2}",
            twap_trimmed, twap_untrimmed
        );
    }

    #[test]
    fn test_twap_empty() {
        let samples: Vec<PriceSample> = vec![];
        let twap = HyperliquidMarkPriceOracle::compute_twap_median_of_means(&samples, 0.05);
        assert_eq!(twap, 0.0);
    }

    #[test]
    fn test_twap_single_sample() {
        let samples = vec![PriceSample {
            timestamp_ms: 1000,
            price: 42.5,
            source: "test".to_string(),
        }];
        let twap = HyperliquidMarkPriceOracle::compute_twap_median_of_means(&samples, 0.05);
        assert_eq!(twap, 42.5);
    }

    #[tokio::test]
    async fn test_settlement_registration() {
        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            ..Default::default()
        };
        let oracle = HyperliquidMarkPriceOracle::new(config).expect("Failed to create oracle");

        let expiry = Utc::now().timestamp() + 3600; // 1 hour from now
        oracle.register_settlement(expiry, 1800).await;

        // Verify settlement was registered
        let state = oracle.state.read().await;
        assert!(state.settlements.contains_key(&expiry));

        let window = state.settlements.get(&expiry).unwrap();
        assert_eq!(window.expiry_timestamp, expiry);
        assert_eq!(window.window_start, expiry - 1800);
        assert_eq!(window.window_end, expiry);
        assert!(window.settlement_price.is_none());
    }

    #[test]
    fn test_config_requires_symbol() {
        // Default config has empty symbol - should fail
        let config = HyperliquidOracleConfig::default();
        let result = HyperliquidMarkPriceOracle::new(config);
        assert!(result.is_err());
        let err = result.err().unwrap();
        assert!(
            err.to_string().contains("symbol must be specified"),
            "Expected error about missing symbol, got: {}",
            err
        );
    }

    #[tokio::test]
    async fn test_no_fallback_when_no_client() {
        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            ..Default::default()
        };
        let oracle = HyperliquidMarkPriceOracle::new(config).unwrap();

        // Expiry in the past, but no hydromancer client
        let past_expiry = Utc::now().timestamp() - 3600;
        let result = oracle.get_settlement_price(past_expiry).await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_no_fallback_when_memory_hit() {
        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            ..Default::default()
        };
        let mut oracle = HyperliquidMarkPriceOracle::new(config).unwrap();
        let mock_client = Arc::new(HydromancerClient::with_base_url(
            "http://should-not-be-called".to_string(),
            "key".to_string(),
        ));
        oracle.set_hydromancer_client(mock_client);

        // Set up in-memory settlement
        let past_expiry = Utc::now().timestamp() - 3600;
        {
            let mut state = oracle.state.write().await;
            state.settlements.insert(
                past_expiry,
                SettlementWindow {
                    expiry_timestamp: past_expiry,
                    window_start: past_expiry - 1800,
                    window_end: past_expiry,
                    samples: vec![],
                    settlement_price: Some(70000.0),
                    persisted: true,
                },
            );
        }

        let result = oracle.get_settlement_price(past_expiry).await;
        assert_eq!(result, Some(70000.0));
    }

    #[tokio::test]
    async fn test_no_fallback_before_window_end() {
        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            ..Default::default()
        };
        let mut oracle = HyperliquidMarkPriceOracle::new(config).unwrap();
        let mock_client = Arc::new(HydromancerClient::with_base_url(
            "http://should-not-be-called".to_string(),
            "key".to_string(),
        ));
        oracle.set_hydromancer_client(mock_client);

        // Expiry in the future
        let future_expiry = Utc::now().timestamp() + 3600;
        let result = oracle.get_settlement_price(future_expiry).await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_fallback_triggers_on_miss() {
        use mockito::Server;

        let mut server = Server::new_async().await;

        // Use a past expiry with a known window
        let past_expiry = 100000_i64; // well in the past
        let twap_window = 1800_i64;
        let window_start_ms = (past_expiry - twap_window) * 1000;

        // Create 600 sample records within the window
        let samples: Vec<String> = (0..600)
            .map(|i| {
                let ts = window_start_ms + i * 3000; // 3s intervals within 1800s window
                format!(
                    r#"{{"time":{},"dex":"Hyperliquid","coin":"BTC","oraclePx":"70000.0","markPx":"70000.0","extPerpPx":null}}"#,
                    ts
                )
            })
            .collect();

        server
            .mock("POST", "/")
            .with_status(200)
            .with_body(format!("[{}]", samples.join(",")))
            .create_async()
            .await;

        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            min_settlement_samples: 50,
            twap_window_seconds: twap_window as u32,
            ..Default::default()
        };
        let mut oracle = HyperliquidMarkPriceOracle::new(config).unwrap();
        let client = Arc::new(HydromancerClient::with_base_url(
            server.url(),
            "test-key".to_string(),
        ));
        oracle.set_hydromancer_client(client);

        let result = oracle.get_settlement_price(past_expiry).await;
        assert!(result.is_some());
        assert_eq!(result.unwrap(), 70000.0);
    }

    #[tokio::test]
    async fn test_cache_prevents_repeat_calls() {
        use mockito::Server;

        let mut server = Server::new_async().await;

        let past_expiry = 100000_i64;
        let twap_window = 1800_i64;
        let window_start_ms = (past_expiry - twap_window) * 1000;

        let samples: Vec<String> = (0..100)
            .map(|i| {
                let ts = window_start_ms + i * 3000;
                format!(
                    r#"{{"time":{},"dex":"Hyperliquid","coin":"BTC","oraclePx":"70000.0","markPx":"70000.0","extPerpPx":null}}"#,
                    ts
                )
            })
            .collect();

        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_body(format!("[{}]", samples.join(",")))
            .expect(1) // Should only be called once
            .create_async()
            .await;

        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            min_settlement_samples: 50,
            twap_window_seconds: twap_window as u32,
            ..Default::default()
        };
        let mut oracle = HyperliquidMarkPriceOracle::new(config).unwrap();
        let client = Arc::new(HydromancerClient::with_base_url(
            server.url(),
            "key".to_string(),
        ));
        oracle.set_hydromancer_client(client);

        // First call hits the API
        let result1 = oracle.attempt_hydromancer_fallback(past_expiry).await;
        assert!(result1.is_some());

        // Second call returns cached
        let result2 = oracle.attempt_hydromancer_fallback(past_expiry).await;
        assert_eq!(result1, result2);

        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_insufficient_samples_returns_none() {
        use mockito::Server;

        let mut server = Server::new_async().await;

        let past_expiry = 100000_i64;
        let twap_window = 1800_i64;
        let window_start_ms = (past_expiry - twap_window) * 1000;

        // Only 10 samples, below min_settlement_samples
        let samples: Vec<String> = (0..10)
            .map(|i| {
                let ts = window_start_ms + i * 3000;
                format!(
                    r#"{{"time":{},"dex":"Hyperliquid","coin":"BTC","oraclePx":"70000.0","markPx":"70000.0","extPerpPx":null}}"#,
                    ts
                )
            })
            .collect();

        server
            .mock("POST", "/")
            .with_status(200)
            .with_body(format!("[{}]", samples.join(",")))
            .create_async()
            .await;

        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            min_settlement_samples: 500, // require 500, only get 10
            twap_window_seconds: twap_window as u32,
            ..Default::default()
        };
        let mut oracle = HyperliquidMarkPriceOracle::new(config).unwrap();
        let client = Arc::new(HydromancerClient::with_base_url(
            server.url(),
            "key".to_string(),
        ));
        oracle.set_hydromancer_client(client);

        let result = oracle.attempt_hydromancer_fallback(past_expiry).await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_min_settlement_samples_floor() {
        // Verify the hard floor of 50 is enforced
        let config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            min_settlement_samples: 10, // below floor
            ..Default::default()
        };
        // Floor is enforced at Default level, not at struct level,
        // so setting it directly bypasses the floor. The floor applies
        // only when reading from env var in Default::default().
        // Direct struct construction can set any value.
        assert_eq!(config.min_settlement_samples, 10);

        // But Default enforces the floor
        let default_config = HyperliquidOracleConfig {
            symbol: "BTC".to_string(),
            ..Default::default()
        };
        assert!(default_config.min_settlement_samples >= MIN_SETTLEMENT_SAMPLES_FLOOR);
    }
}