hypercall/price_oracle/

hyperliquid_ws.rs

//! WebSocket-based price feed from Hyperliquid.
//!
//! Maintains a single persistent WebSocket connection to the Hyperliquid public API,
//! subscribing to `activeAssetCtx` for configured coins. This replaces HTTP polling
//! of `metaAndAssetCtxs` which isn't available on self-hosted nodes and hits rate limits.

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

use futures::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, error, info, warn};

/// Default WebSocket URL for Hyperliquid mainnet.
pub const DEFAULT_WS_URL: &str = "wss://api.hyperliquid.xyz/ws";

/// Default WebSocket URL for Hyperliquid testnet.
pub const DEFAULT_TESTNET_WS_URL: &str = "wss://api.hyperliquid-testnet.xyz/ws";

/// A cached oracle price with its timestamp.
#[derive(Debug, Clone)]
struct PriceEntry {
    price: f64,
    prev_day_px: Option<f64>,
    updated_at: Instant,
}

/// Shared WebSocket price feed for Hyperliquid oracle prices.
///
/// One instance serves multiple oracle consumers. Subscribes to `activeAssetCtx`
/// for each configured coin and stores the latest oracle price.
pub struct HyperliquidWsFeed {
    ws_url: String,
    coins: Vec<String>,
    prices: Arc<RwLock<HashMap<String, PriceEntry>>>,
}

impl HyperliquidWsFeed {
    pub fn new(ws_url: String, coins: Vec<String>) -> Self {
        Self {
            ws_url,
            coins,
            prices: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Create a mock feed with default prices for each coin. Used in test mode
    /// so that CatalogManager can create markets without a live WS connection.
    /// Prices are refreshed on every `get_price` call so they never go stale.
    pub fn new_mock(coins: Vec<String>, default_prices: &HashMap<String, f64>) -> Self {
        let mut prices = HashMap::new();
        for coin in &coins {
            if let Some(&price) = default_prices.get(coin) {
                prices.insert(
                    coin.clone(),
                    PriceEntry {
                        price,
                        prev_day_px: Some(price),
                        updated_at: Instant::now(),
                    },
                );
            }
        }
        Self {
            ws_url: String::new(),
            coins,
            prices: Arc::new(RwLock::new(prices)),
        }
    }

    /// Refresh the `updated_at` timestamp on all mock prices so they never go stale.
    /// Called periodically in test mode since `get_price` returns None after 10s.
    pub async fn refresh_mock_timestamps(&self) {
        let mut prices = self.prices.write().await;
        for entry in prices.values_mut() {
            entry.updated_at = Instant::now();
        }
    }

    /// Get the latest oracle price for a coin.
    ///
    /// Returns None if no price has been received or if the price is stale (>10s old).
    pub async fn get_price(&self, coin: &str) -> Option<f64> {
        let prices = self.prices.read().await;
        prices.get(coin).and_then(|entry| {
            if entry.updated_at.elapsed() < Duration::from_secs(10) {
                Some(entry.price)
            } else {
                None
            }
        })
    }

    /// Get the latest previous-day price for a coin from the WS feed.
    pub async fn get_prev_day_price(&self, coin: &str) -> Option<f64> {
        let prices = self.prices.read().await;
        prices.get(coin).and_then(|entry| entry.prev_day_px)
    }

    /// Start the WebSocket feed. Runs until shutdown signal.
    ///
    /// Auto-reconnects with exponential backoff on disconnection.
    pub async fn start(&self, mut shutdown_rx: tokio::sync::broadcast::Receiver<()>) {
        let prices = self.prices.clone();
        let ws_url = self.ws_url.clone();
        let coins = self.coins.clone();

        const BASE_BACKOFF: Duration = Duration::from_secs(1);
        const MAX_BACKOFF: Duration = Duration::from_secs(30);
        let mut consecutive_failures: u32 = 0;

        loop {
            let before = Instant::now();
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("HyperliquidWsFeed: shutdown signal received");
                    return;
                }
                _ = Self::run_connection(&ws_url, &coins, &prices) => {
                    // If the connection stayed up for >60s it was healthy;
                    // reset backoff so the next reconnect is fast.
                    if before.elapsed() > Duration::from_secs(60) {
                        consecutive_failures = 0;
                    }
                    consecutive_failures += 1;
                    let backoff = std::cmp::min(
                        BASE_BACKOFF * 2u32.saturating_pow(consecutive_failures - 1),
                        MAX_BACKOFF,
                    );
                    warn!(
                        "HyperliquidWsFeed: connection lost (failures: {}), reconnecting in {:?}",
                        consecutive_failures, backoff
                    );
                    tokio::select! {
                        _ = shutdown_rx.recv() => {
                            info!("HyperliquidWsFeed: shutdown during reconnect backoff");
                            return;
                        }
                        _ = tokio::time::sleep(backoff) => {}
                    }
                }
            }
        }
    }

    async fn run_connection(
        ws_url: &str,
        coins: &[String],
        prices: &Arc<RwLock<HashMap<String, PriceEntry>>>,
    ) {
        info!("HyperliquidWsFeed: connecting to {}", ws_url);

        let (ws_stream, _) = match tokio_tungstenite::connect_async(ws_url).await {
            Ok(conn) => {
                info!("HyperliquidWsFeed: connected");
                conn
            }
            Err(e) => {
                error!("HyperliquidWsFeed: connection failed: {}", e);
                return;
            }
        };

        let (mut write, mut read) = ws_stream.split();

        // Subscribe to activeAssetCtx for each coin
        for coin in coins {
            let sub = WsSubscription {
                method: "subscribe".to_string(),
                subscription: ActiveAssetCtxSub {
                    sub_type: "activeAssetCtx".to_string(),
                    coin: coin.clone(),
                },
            };
            let msg = sonic_rs::to_string(&sub).expect("serialize subscription");
            if let Err(e) = write.send(Message::Text(msg)).await {
                error!("HyperliquidWsFeed: failed to subscribe to {}: {}", coin, e);
                return;
            }
            info!("HyperliquidWsFeed: subscribed to activeAssetCtx:{}", coin);
        }

        // Read messages with a read timeout to detect hung connections.
        // Without this, a half-open connection (connected but not sending data)
        // blocks read.next() forever, preventing reconnect. This was the root
        // cause of CALL-720 (22h stale spot prices in the market-maker) and
        // CALL-731 (same issue in the API server).
        const WS_READ_TIMEOUT: Duration = Duration::from_secs(60);

        loop {
            match tokio::time::timeout(WS_READ_TIMEOUT, read.next()).await {
                Ok(Some(msg)) => match msg {
                    Ok(Message::Text(text)) => {
                        Self::handle_message(&text, prices).await;
                    }
                    Ok(Message::Ping(data)) => {
                        if let Err(e) = write.send(Message::Pong(data)).await {
                            error!("HyperliquidWsFeed: failed to send pong: {}", e);
                            return;
                        }
                    }
                    Ok(Message::Close(_)) => {
                        info!("HyperliquidWsFeed: server sent close frame");
                        return;
                    }
                    Err(e) => {
                        error!("HyperliquidWsFeed: read error: {}", e);
                        return;
                    }
                    _ => {}
                },
                Ok(None) => {
                    info!("HyperliquidWsFeed: stream ended");
                    return;
                }
                Err(_) => {
                    warn!(
                        "HyperliquidWsFeed: read timeout ({}s) — no messages received, reconnecting",
                        WS_READ_TIMEOUT.as_secs()
                    );
                    return;
                }
            }
        }
    }

    async fn handle_message(text: &str, prices: &Arc<RwLock<HashMap<String, PriceEntry>>>) {
        // Parse the WS message envelope
        let envelope: WsMessage = match sonic_rs::from_str(text) {
            Ok(m) => m,
            Err(_) => {
                // Subscription confirmations and other non-data messages
                debug!(
                    "HyperliquidWsFeed: non-data message: {}",
                    &text[..text.len().min(200)]
                );
                return;
            }
        };

        // We only care about activeAssetCtx channels
        if !envelope.channel.starts_with("activeAssetCtx") {
            return;
        }

        // Extract coin from channel name: "activeAssetCtx:BTC" -> "BTC"
        // The channel is "activeAssetCtx" but coin comes from the data
        let ctx: ActiveAssetCtxData = match sonic_rs::from_value(&envelope.data) {
            Ok(d) => d,
            Err(e) => {
                warn!(
                    "HyperliquidWsFeed: failed to parse activeAssetCtx data: {}",
                    e
                );
                return;
            }
        };

        let coin = ctx.coin;
        let prev_day_px = ctx
            .ctx
            .prev_day_px
            .parse::<f64>()
            .ok()
            .filter(|v| v.is_finite() && *v > 0.0);
        if let Some(oracle_price) = ctx
            .ctx
            .oracle_px
            .parse::<f64>()
            .ok()
            .filter(|v| v.is_finite() && *v > 0.0)
        {
            debug!(
                "HyperliquidWsFeed: {} oracle_px={} mark_px={} prev_day_px={:?}",
                coin, oracle_price, ctx.ctx.mark_px, prev_day_px
            );
            let mut prices = prices.write().await;
            // Preserve existing prev_day_px if the WS message omitted it
            let effective_prev_day_px =
                prev_day_px.or_else(|| prices.get(&coin).and_then(|e| e.prev_day_px));
            prices.insert(
                coin,
                PriceEntry {
                    price: oracle_price,
                    prev_day_px: effective_prev_day_px,
                    updated_at: Instant::now(),
                },
            );
        }
    }
}

// --- Serde types for WS protocol ---

#[derive(Serialize)]
struct WsSubscription {
    method: String,
    subscription: ActiveAssetCtxSub,
}

#[derive(Serialize)]
struct ActiveAssetCtxSub {
    #[serde(rename = "type")]
    sub_type: String,
    coin: String,
}

#[derive(Deserialize)]
struct WsMessage {
    channel: String,
    data: sonic_rs::Value,
}

#[derive(Deserialize)]
struct ActiveAssetCtxData {
    coin: String,
    ctx: AssetCtxWs,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct AssetCtxWs {
    oracle_px: String,
    mark_px: String,
    #[serde(default)]
    prev_day_px: String,
}

#[async_trait::async_trait]
impl crate::shared::service::Service for HyperliquidWsFeed {
    fn name(&self) -> &'static str {
        "HyperliquidWsFeed"
    }

    fn owner(&self) -> crate::shared::service::ServiceOwner {
        crate::shared::service::ServiceOwner::Engine
    }

    async fn run(
        self: std::sync::Arc<Self>,
        shutdown: crate::shared::ShutdownRx,
    ) -> anyhow::Result<()> {
        self.start(shutdown).await;
        Ok(())
    }
}

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

    #[test]
    fn test_parse_active_asset_ctx_message() {
        let json = r#"{
            "channel": "activeAssetCtx:BTC",
            "data": {
                "coin": "BTC",
                "ctx": {
                    "funding": "0.0001",
                    "openInterest": "500000",
                    "prevDayPx": "64000.0",
                    "dayNtlVlm": "1000000000",
                    "premium": "0.001",
                    "oraclePx": "64500.5",
                    "markPx": "64550.0"
                }
            }
        }"#;

        let msg: WsMessage = sonic_rs::from_str(json).unwrap();
        assert_eq!(msg.channel, "activeAssetCtx:BTC");

        let data: ActiveAssetCtxData = sonic_rs::from_value(&msg.data).unwrap();
        assert_eq!(data.coin, "BTC");
        assert_eq!(data.ctx.oracle_px, "64500.5");
        assert_eq!(data.ctx.mark_px, "64550.0");
    }

    #[test]
    fn test_subscription_serialization() {
        let sub = WsSubscription {
            method: "subscribe".to_string(),
            subscription: ActiveAssetCtxSub {
                sub_type: "activeAssetCtx".to_string(),
                coin: "ETH".to_string(),
            },
        };
        let json = sonic_rs::to_string(&sub).unwrap();
        assert!(json.contains("\"method\":\"subscribe\""));
        assert!(json.contains("\"type\":\"activeAssetCtx\""));
        assert!(json.contains("\"coin\":\"ETH\""));
    }
}