predict_fun_sdk/

ws.rs

//! Predict.fun WebSocket client for real-time market data.
//!
//! Connects to `wss://ws.predict.fun/ws` and provides:
//! - Orderbook snapshots (`predictOrderbook/{marketId}`)
//! - Asset price updates (`assetPriceUpdate/{feedId}`)
//! - Cross-venue chance data (`polymarketChance/{marketId}`, `kalshiChance/{marketId}`)
//! - Wallet event notifications (`predictWalletEvents/{jwt}`)
//!
//! # Protocol
//!
//! Custom JSON RPC over WebSocket:
//! ```text
//! → {"requestId": 0, "method": "subscribe", "params": ["predictOrderbook/123"]}
//! ← {"type": "R", "requestId": 0, "success": true}
//! ← {"type": "M", "topic": "predictOrderbook/123", "data": {...}}
//! ```
//!
//! Message types: `R` = Response (to a request), `M` = Push message (subscription data).
//!
//! # Example
//!
//! ```rust,no_run
//! use predict_fun_sdk::ws::{PredictWsClient, PredictWsMessage, Topic};
//!
//! # async fn example() -> anyhow::Result<()> {
//! let (client, mut rx) = PredictWsClient::connect_mainnet().await?;
//!
//! client.subscribe(Topic::Orderbook { market_id: 45532 }).await?;
//! client.subscribe(Topic::AssetPrice { feed_id: 1 }).await?;
//!
//! while let Some(msg) = rx.recv().await {
//!     match msg {
//!         PredictWsMessage::Orderbook(ob) => {
//!             println!("OB market={}: {} bids, {} asks",
//!                 ob.market_id, ob.bids.len(), ob.asks.len());
//!         }
//!         PredictWsMessage::AssetPrice(p) => {
//!             println!("Price feed {}: ${:.2}", p.feed_id, p.price);
//!         }
//!         _ => {}
//!     }
//! }
//! # Ok(())
//! # }
//! ```

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use anyhow::{anyhow, Context, Result};
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio::time::{self, Duration};
use tokio_tungstenite::tungstenite::Message as WsMessage;
use tracing::{debug, error, info, warn};

/// WebSocket endpoints.
pub const PREDICT_WS_MAINNET: &str = "wss://ws.predict.fun/ws";
pub const PREDICT_WS_TESTNET: &str = "wss://ws.bnb.predict.fail/ws";

/// GraphQL endpoints (for reference / future use).
pub const PREDICT_GQL_MAINNET: &str = "https://graphql.predict.fun/graphql";
pub const PREDICT_GQL_TESTNET: &str = "https://graphql.bnb.predict.fail/graphql";

// ── Topic types ──

/// Subscription topic for the predict.fun WebSocket feed.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Topic {
    /// Real-time orderbook snapshots for a market.
    /// Topic string: `predictOrderbook/{market_id}`
    Orderbook { market_id: i64 },

    /// Asset price updates from oracle feeds.
    /// Topic string: `assetPriceUpdate/{feed_id}`
    /// Known feeds: 1=BTC, 4=ETH
    AssetPrice { feed_id: i64 },

    /// Polymarket chance/probability for a market (cross-venue reference).
    /// Topic string: `polymarketChance/{market_id}`
    PolymarketChance { market_id: i64 },

    /// Kalshi chance/probability for a market (cross-venue reference).
    /// Topic string: `kalshiChance/{market_id}`
    KalshiChance { market_id: i64 },

    /// Wallet events (fills, settlements). Requires JWT auth token.
    /// Topic string: `predictWalletEvents/{jwt}`
    WalletEvents { jwt: String },

    /// Raw topic string for undocumented/custom topics.
    Raw(String),
}

impl Topic {
    /// Convert to the wire-format topic string.
    pub fn to_topic_string(&self) -> String {
        match self {
            Topic::Orderbook { market_id } => format!("predictOrderbook/{}", market_id),
            Topic::AssetPrice { feed_id } => format!("assetPriceUpdate/{}", feed_id),
            Topic::PolymarketChance { market_id } => format!("polymarketChance/{}", market_id),
            Topic::KalshiChance { market_id } => format!("kalshiChance/{}", market_id),
            Topic::WalletEvents { jwt } => format!("predictWalletEvents/{}", jwt),
            Topic::Raw(s) => s.clone(),
        }
    }

    /// Parse a wire-format topic string back into a Topic.
    pub fn from_topic_string(s: &str) -> Self {
        if let Some(rest) = s.strip_prefix("predictOrderbook/") {
            if let Ok(id) = rest.parse::<i64>() {
                return Topic::Orderbook { market_id: id };
            }
        }
        if let Some(rest) = s.strip_prefix("assetPriceUpdate/") {
            if let Ok(id) = rest.parse::<i64>() {
                return Topic::AssetPrice { feed_id: id };
            }
        }
        if let Some(rest) = s.strip_prefix("polymarketChance/") {
            if let Ok(id) = rest.parse::<i64>() {
                return Topic::PolymarketChance { market_id: id };
            }
        }
        if let Some(rest) = s.strip_prefix("kalshiChance/") {
            if let Ok(id) = rest.parse::<i64>() {
                return Topic::KalshiChance { market_id: id };
            }
        }
        if let Some(rest) = s.strip_prefix("predictWalletEvents/") {
            return Topic::WalletEvents {
                jwt: rest.to_string(),
            };
        }
        Topic::Raw(s.to_string())
    }
}

impl std::fmt::Display for Topic {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.to_topic_string())
    }
}

// ── Parsed message types ──

/// Orderbook level: `[price, size]`.
pub type Level = (f64, f64);

/// Last settled order info from the orderbook snapshot.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LastOrderSettled {
    pub id: String,
    pub kind: String,
    #[serde(rename = "marketId")]
    pub market_id: i64,
    pub outcome: String,
    pub price: String,
    pub side: String,
}

/// Full orderbook snapshot pushed on every change.
#[derive(Debug, Clone)]
pub struct OrderbookSnapshot {
    pub market_id: i64,
    pub bids: Vec<Level>,
    pub asks: Vec<Level>,
    pub version: u64,
    pub update_timestamp_ms: u64,
    pub order_count: u64,
    pub last_order_settled: Option<LastOrderSettled>,
}

impl OrderbookSnapshot {
    /// Best bid price, or None if empty.
    pub fn best_bid(&self) -> Option<f64> {
        self.bids.first().map(|(p, _)| *p)
    }

    /// Best ask price, or None if empty.
    pub fn best_ask(&self) -> Option<f64> {
        self.asks.first().map(|(p, _)| *p)
    }

    /// Midpoint price, or None if either side is empty.
    pub fn mid(&self) -> Option<f64> {
        match (self.best_bid(), self.best_ask()) {
            (Some(b), Some(a)) => Some((b + a) / 2.0),
            _ => None,
        }
    }

    /// Spread (ask - bid), or None if either side is empty.
    pub fn spread(&self) -> Option<f64> {
        match (self.best_bid(), self.best_ask()) {
            (Some(b), Some(a)) => Some(a - b),
            _ => None,
        }
    }
}

/// Asset price update from oracle feed.
#[derive(Debug, Clone)]
pub struct AssetPriceUpdate {
    pub feed_id: i64,
    pub price: f64,
    pub publish_time: u64,
    pub timestamp: u64,
}

/// Cross-venue chance data (Polymarket or Kalshi).
#[derive(Debug, Clone)]
pub struct CrossVenueChance {
    pub source: CrossVenueSource,
    pub market_id: i64,
    pub data: Value,
}

/// Source of cross-venue data.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CrossVenueSource {
    Polymarket,
    Kalshi,
}

/// Wallet event notification.
#[derive(Debug, Clone)]
pub struct WalletEvent {
    pub data: Value,
}

/// Parsed WebSocket message from predict.fun.
#[derive(Debug, Clone)]
pub enum PredictWsMessage {
    /// Full orderbook snapshot.
    Orderbook(OrderbookSnapshot),
    /// Asset price update from oracle.
    AssetPrice(AssetPriceUpdate),
    /// Cross-venue chance data.
    CrossVenueChance(CrossVenueChance),
    /// Wallet event (fills, settlements).
    WalletEvent(WalletEvent),
    /// Unparsed push message (unknown topic or parse failure).
    Raw { topic: String, data: Value },
}

// ── Wire protocol types ──

#[derive(Serialize)]
struct WsRequest {
    #[serde(rename = "requestId")]
    request_id: u64,
    method: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    params: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    data: Option<Value>,
}

#[derive(Deserialize)]
struct WsRawMessage {
    #[serde(rename = "type")]
    msg_type: String,
    #[serde(rename = "requestId")]
    request_id: Option<i64>,
    success: Option<bool>,
    error: Option<WsError>,
    topic: Option<String>,
    data: Option<Value>,
}

#[derive(Deserialize, Debug, Clone)]
struct WsError {
    code: String,
    message: Option<String>,
}

impl std::fmt::Display for WsError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.code)?;
        if let Some(msg) = &self.message {
            write!(f, ": {}", msg)?;
        }
        Ok(())
    }
}

// ── Client ──

type WsSink = futures_util::stream::SplitSink<
    tokio_tungstenite::WebSocketStream<
        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
    >,
    WsMessage,
>;

/// Pending subscribe/unsubscribe response channel.
type PendingResponse = oneshot::Sender<Result<()>>;

/// Configuration for the WebSocket client.
#[derive(Debug, Clone)]
pub struct PredictWsConfig {
    /// WebSocket URL. Defaults to mainnet.
    pub url: String,
    /// Channel buffer size for outgoing messages. Default: 1024.
    pub channel_buffer: usize,
    /// Heartbeat timeout in seconds. Default: 60.
    pub heartbeat_timeout_secs: u64,
    /// Maximum reconnection attempts. 0 = infinite. Default: 0.
    pub max_reconnect_attempts: u32,
    /// Maximum reconnection backoff in seconds. Default: 15.
    pub max_reconnect_backoff_secs: u64,
}

impl Default for PredictWsConfig {
    fn default() -> Self {
        Self {
            url: PREDICT_WS_MAINNET.to_string(),
            channel_buffer: 1024,
            heartbeat_timeout_secs: 60,
            max_reconnect_attempts: 0,
            max_reconnect_backoff_secs: 15,
        }
    }
}

impl PredictWsConfig {
    pub fn mainnet() -> Self {
        Self::default()
    }

    pub fn testnet() -> Self {
        Self {
            url: PREDICT_WS_TESTNET.to_string(),
            ..Self::default()
        }
    }
}

/// Handle for interacting with the WebSocket connection.
///
/// Subscribe/unsubscribe to topics. Messages are received on the `mpsc::Receiver`
/// returned from [`PredictWsClient::connect`].
#[derive(Clone)]
pub struct PredictWsClient {
    sink: Arc<Mutex<WsSink>>,
    request_id: Arc<AtomicU64>,
    pending: Arc<Mutex<HashMap<u64, PendingResponse>>>,
    active_topics: Arc<Mutex<Vec<String>>>,
    config: PredictWsConfig,
}

impl PredictWsClient {
    /// Connect to mainnet WebSocket and return (client_handle, message_receiver).
    pub async fn connect_mainnet() -> Result<(Self, mpsc::Receiver<PredictWsMessage>)> {
        Self::connect(PredictWsConfig::mainnet()).await
    }

    /// Connect to testnet WebSocket and return (client_handle, message_receiver).
    pub async fn connect_testnet() -> Result<(Self, mpsc::Receiver<PredictWsMessage>)> {
        Self::connect(PredictWsConfig::testnet()).await
    }

    /// Connect with custom configuration.
    pub async fn connect(
        config: PredictWsConfig,
    ) -> Result<(Self, mpsc::Receiver<PredictWsMessage>)> {
        let (ws_stream, _) = tokio_tungstenite::connect_async(&config.url)
            .await
            .with_context(|| format!("failed to connect to {}", config.url))?;

        info!("Connected to {}", config.url);

        let (sink, stream) = ws_stream.split();
        let (tx, rx) = mpsc::channel(config.channel_buffer);

        let client = Self {
            sink: Arc::new(Mutex::new(sink)),
            request_id: Arc::new(AtomicU64::new(0)),
            pending: Arc::new(Mutex::new(HashMap::new())),
            active_topics: Arc::new(Mutex::new(Vec::new())),
            config: config.clone(),
        };

        // Spawn the read loop
        let client_clone = client.clone();
        tokio::spawn(async move {
            client_clone.read_loop(stream, tx).await
        });

        Ok((client, rx))
    }

    /// Subscribe to a topic. Returns Ok(()) when the server acknowledges.
    pub async fn subscribe(&self, topic: Topic) -> Result<()> {
        let topic_str = topic.to_topic_string();
        let request_id = self.request_id.fetch_add(1, Ordering::Relaxed);

        let (resp_tx, resp_rx) = oneshot::channel();
        {
            let mut pending = self.pending.lock().await;
            pending.insert(request_id, resp_tx);
        }

        let msg = WsRequest {
            request_id,
            method: "subscribe".to_string(),
            params: Some(vec![topic_str.clone()]),
            data: None,
        };

        self.send_raw(&msg).await?;
        debug!("Subscribing to {} (requestId={})", topic_str, request_id);

        // Wait for server response
        let result = tokio::time::timeout(Duration::from_secs(10), resp_rx)
            .await
            .map_err(|_| anyhow!("subscribe timeout for {}", topic_str))?
            .map_err(|_| anyhow!("subscribe channel closed for {}", topic_str))??;

        // Track active topic for reconnection
        {
            let mut topics = self.active_topics.lock().await;
            if !topics.contains(&topic_str) {
                topics.push(topic_str.clone());
            }
        }

        info!("Subscribed to {}", topic_str);
        Ok(result)
    }

    /// Unsubscribe from a topic.
    pub async fn unsubscribe(&self, topic: Topic) -> Result<()> {
        let topic_str = topic.to_topic_string();
        let request_id = self.request_id.fetch_add(1, Ordering::Relaxed);

        let (resp_tx, resp_rx) = oneshot::channel();
        {
            let mut pending = self.pending.lock().await;
            pending.insert(request_id, resp_tx);
        }

        let msg = WsRequest {
            request_id,
            method: "unsubscribe".to_string(),
            params: Some(vec![topic_str.clone()]),
            data: None,
        };

        self.send_raw(&msg).await?;

        tokio::time::timeout(Duration::from_secs(10), resp_rx)
            .await
            .map_err(|_| anyhow!("unsubscribe timeout for {}", topic_str))?
            .map_err(|_| anyhow!("unsubscribe channel closed for {}", topic_str))??;

        // Remove from active topics
        {
            let mut topics = self.active_topics.lock().await;
            topics.retain(|t| t != &topic_str);
        }

        info!("Unsubscribed from {}", topic_str);
        Ok(())
    }

    /// Get list of currently subscribed topics.
    pub async fn active_topics(&self) -> Vec<String> {
        self.active_topics.lock().await.clone()
    }

    /// Send a raw heartbeat response (called internally by the read loop).
    async fn send_heartbeat(&self, data: &Value) -> Result<()> {
        let msg = WsRequest {
            request_id: self.request_id.fetch_add(1, Ordering::Relaxed),
            method: "heartbeat".to_string(),
            params: None,
            data: Some(data.clone()),
        };
        self.send_raw(&msg).await
    }

    async fn send_raw(&self, msg: &WsRequest) -> Result<()> {
        let text = serde_json::to_string(msg).context("failed to serialize WS message")?;
        let mut sink = self.sink.lock().await;
        sink.send(WsMessage::Text(text))
            .await
            .context("failed to send WS message")?;
        Ok(())
    }

    /// Main read loop — processes incoming messages, dispatches to channel.
    fn read_loop(
        &self,
        mut stream: futures_util::stream::SplitStream<
            tokio_tungstenite::WebSocketStream<
                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
            >,
        >,
        tx: mpsc::Sender<PredictWsMessage>,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
        let heartbeat_timeout = Duration::from_secs(self.config.heartbeat_timeout_secs);
        let mut last_heartbeat = time::Instant::now();

        loop {
            tokio::select! {
                msg = stream.next() => {
                    match msg {
                        Some(Ok(WsMessage::Text(text))) => {
                            match serde_json::from_str::<WsRawMessage>(&text) {
                                Ok(raw) => self.handle_message(raw, &tx, &mut last_heartbeat).await,
                                Err(e) => warn!("Failed to parse WS message: {} — raw: {}", e, &text[..text.len().min(200)]),
                            }
                        }
                        Some(Ok(WsMessage::Ping(data))) => {
                            let mut sink = self.sink.lock().await;
                            let _ = sink.send(WsMessage::Pong(data)).await;
                        }
                        Some(Ok(WsMessage::Close(frame))) => {
                            info!("WebSocket closed by server: {:?}", frame);
                            break;
                        }
                        Some(Err(e)) => {
                            error!("WebSocket error: {}", e);
                            break;
                        }
                        None => {
                            info!("WebSocket stream ended");
                            break;
                        }
                        _ => {} // Binary, Pong, Frame — ignore
                    }
                }
                _ = time::sleep(heartbeat_timeout) => {
                    if last_heartbeat.elapsed() > heartbeat_timeout {
                        warn!("Heartbeat timeout ({}s), closing connection",
                            self.config.heartbeat_timeout_secs);
                        break;
                    }
                }
            }
        }

        // Attempt reconnection
        self.try_reconnect(tx).await;
        }) // end Box::pin
    }

    async fn handle_message(
        &self,
        raw: WsRawMessage,
        tx: &mpsc::Sender<PredictWsMessage>,
        last_heartbeat: &mut time::Instant,
    ) {
        match raw.msg_type.as_str() {
            // Response to a request (subscribe/unsubscribe)
            "R" => {
                if let Some(req_id) = raw.request_id {
                    let mut pending = self.pending.lock().await;
                    if let Some(resp_tx) = pending.remove(&(req_id as u64)) {
                        let result = if raw.success.unwrap_or(false) {
                            Ok(())
                        } else {
                            let err_msg = raw
                                .error
                                .map(|e| e.to_string())
                                .unwrap_or_else(|| "unknown error".to_string());
                            Err(anyhow!("subscribe failed: {}", err_msg))
                        };
                        let _ = resp_tx.send(result);
                    }
                }
            }
            // Push message (subscription data)
            "M" => {
                let topic_str = match &raw.topic {
                    Some(t) => t.as_str(),
                    None => return,
                };

                // Handle heartbeat
                if topic_str == "heartbeat" {
                    *last_heartbeat = time::Instant::now();
                    if let Some(data) = &raw.data {
                        if let Err(e) = self.send_heartbeat(data).await {
                            warn!("Failed to send heartbeat response: {}", e);
                        }
                    }
                    return;
                }

                // Parse and dispatch
                let data = match raw.data {
                    Some(d) => d,
                    None => return,
                };

                let parsed = parse_push_message(topic_str, data);
                if tx.try_send(parsed).is_err() {
                    warn!("Message channel full, dropping message for topic {}", topic_str);
                }
            }
            other => {
                debug!("Unknown WS message type: {}", other);
            }
        }
    }

    fn try_reconnect(
        &self,
        tx: mpsc::Sender<PredictWsMessage>,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
        let max_attempts = self.config.max_reconnect_attempts;
        let max_backoff = self.config.max_reconnect_backoff_secs;
        let mut attempt = 0u32;

        loop {
            if max_attempts > 0 && attempt >= max_attempts {
                error!(
                    "Max reconnection attempts ({}) reached, giving up",
                    max_attempts
                );
                return;
            }

            let backoff = Duration::from_secs((2u64.pow(attempt.min(10))).min(max_backoff));
            warn!(
                "Reconnecting in {:?} (attempt {})",
                backoff,
                attempt + 1
            );
            time::sleep(backoff).await;
            attempt += 1;

            match tokio_tungstenite::connect_async(&self.config.url).await {
                Ok((ws_stream, _)) => {
                    info!("Reconnected to {}", self.config.url);
                    let (new_sink, new_stream) = ws_stream.split();

                    // Replace sink
                    {
                        let mut sink = self.sink.lock().await;
                        *sink = new_sink;
                    }

                    // Resubscribe to all active topics
                    let topics = self.active_topics.lock().await.clone();
                    for topic_str in &topics {
                        let request_id = self.request_id.fetch_add(1, Ordering::Relaxed);
                        let msg = WsRequest {
                            request_id,
                            method: "subscribe".to_string(),
                            params: Some(vec![topic_str.clone()]),
                            data: None,
                        };
                        if let Err(e) = self.send_raw(&msg).await {
                            warn!("Failed to resubscribe to {}: {}", topic_str, e);
                        } else {
                            debug!("Resubscribed to {}", topic_str);
                        }
                    }

                    // Restart read loop
                    self.read_loop(new_stream, tx).await;
                    return;
                }
                Err(e) => {
                    error!("Reconnection failed: {}", e);
                }
            }
        }
        }) // end Box::pin
    }
}

// ── Message parsing ──

fn parse_push_message(topic: &str, data: Value) -> PredictWsMessage {
    if topic.starts_with("predictOrderbook/") {
        if let Some(ob) = parse_orderbook(topic, &data) {
            return PredictWsMessage::Orderbook(ob);
        }
    }

    if topic.starts_with("assetPriceUpdate/") {
        if let Some(price) = parse_asset_price(topic, &data) {
            return PredictWsMessage::AssetPrice(price);
        }
    }

    if topic.starts_with("polymarketChance/") {
        if let Ok(id) = topic.strip_prefix("polymarketChance/").unwrap_or("").parse::<i64>() {
            return PredictWsMessage::CrossVenueChance(CrossVenueChance {
                source: CrossVenueSource::Polymarket,
                market_id: id,
                data,
            });
        }
    }

    if topic.starts_with("kalshiChance/") {
        if let Ok(id) = topic.strip_prefix("kalshiChance/").unwrap_or("").parse::<i64>() {
            return PredictWsMessage::CrossVenueChance(CrossVenueChance {
                source: CrossVenueSource::Kalshi,
                market_id: id,
                data,
            });
        }
    }

    if topic.starts_with("predictWalletEvents/") {
        return PredictWsMessage::WalletEvent(WalletEvent { data });
    }

    PredictWsMessage::Raw {
        topic: topic.to_string(),
        data,
    }
}

fn parse_levels(val: &Value) -> Vec<Level> {
    val.as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|lvl| {
                    let price = lvl.get(0).and_then(|v| v.as_f64())?;
                    let size = lvl.get(1).and_then(|v| v.as_f64())?;
                    Some((price, size))
                })
                .collect()
        })
        .unwrap_or_default()
}

fn parse_orderbook(topic: &str, data: &Value) -> Option<OrderbookSnapshot> {
    let market_id = topic
        .strip_prefix("predictOrderbook/")?
        .parse::<i64>()
        .ok()?;

    let bids = parse_levels(data.get("bids")?);
    let asks = parse_levels(data.get("asks")?);
    let version = data.get("version").and_then(|v| v.as_u64()).unwrap_or(0);
    let update_timestamp_ms = data
        .get("updateTimestampMs")
        .and_then(|v| v.as_u64())
        .unwrap_or(0);
    let order_count = data
        .get("orderCount")
        .and_then(|v| v.as_u64())
        .unwrap_or(0);
    let last_order_settled = data
        .get("lastOrderSettled")
        .and_then(|v| serde_json::from_value(v.clone()).ok());

    Some(OrderbookSnapshot {
        market_id,
        bids,
        asks,
        version,
        update_timestamp_ms,
        order_count,
        last_order_settled,
    })
}

fn parse_asset_price(topic: &str, data: &Value) -> Option<AssetPriceUpdate> {
    let feed_id = topic
        .strip_prefix("assetPriceUpdate/")?
        .parse::<i64>()
        .ok()?;

    let price = data.get("price").and_then(|v| v.as_f64())?;
    let publish_time = data.get("publishTime").and_then(|v| v.as_u64()).unwrap_or(0);
    let timestamp = data.get("timestamp").and_then(|v| v.as_u64()).unwrap_or(0);

    Some(AssetPriceUpdate {
        feed_id,
        price,
        publish_time,
        timestamp,
    })
}

// ── Known asset feed IDs ──

/// Known asset price feed IDs. Use with `Topic::AssetPrice { feed_id }`.
pub mod feeds {
    /// Bitcoin price feed.
    pub const BTC: i64 = 1;
    /// Ethereum price feed.
    pub const ETH: i64 = 4;
    /// BNB price feed (tentative — needs confirmation).
    pub const BNB: i64 = 2;
}

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

    #[test]
    fn topic_roundtrip() {
        let topics = vec![
            Topic::Orderbook { market_id: 123 },
            Topic::AssetPrice { feed_id: 1 },
            Topic::PolymarketChance { market_id: 456 },
            Topic::KalshiChance { market_id: 789 },
            Topic::WalletEvents {
                jwt: "abc123".to_string(),
            },
            Topic::Raw("custom/topic".to_string()),
        ];

        for topic in topics {
            let s = topic.to_topic_string();
            let parsed = Topic::from_topic_string(&s);
            assert_eq!(topic, parsed, "Roundtrip failed for {}", s);
        }
    }

    #[test]
    fn topic_display() {
        assert_eq!(
            Topic::Orderbook { market_id: 42 }.to_string(),
            "predictOrderbook/42"
        );
        assert_eq!(
            Topic::AssetPrice { feed_id: 1 }.to_string(),
            "assetPriceUpdate/1"
        );
    }

    #[test]
    fn parse_orderbook_snapshot() {
        let data = serde_json::json!({
            "asks": [[0.72, 15.0], [0.83, 5.88]],
            "bids": [[0.57, 15.0], [0.38, 2.63]],
            "marketId": 45532,
            "version": 1,
            "updateTimestampMs": 1772898630219u64,
            "orderCount": 13,
            "lastOrderSettled": {
                "id": "20035648",
                "kind": "LIMIT",
                "marketId": 45532,
                "outcome": "No",
                "price": "0.60",
                "side": "Bid"
            }
        });

        let ob = parse_orderbook("predictOrderbook/45532", &data).unwrap();
        assert_eq!(ob.market_id, 45532);
        assert_eq!(ob.bids.len(), 2);
        assert_eq!(ob.asks.len(), 2);
        assert!((ob.bids[0].0 - 0.57).abs() < 1e-10);
        assert!((ob.asks[0].0 - 0.72).abs() < 1e-10);
        assert_eq!(ob.version, 1);
        assert_eq!(ob.order_count, 13);
        assert!(ob.last_order_settled.is_some());
        assert!((ob.best_bid().unwrap() - 0.57).abs() < 1e-10);
        assert!((ob.best_ask().unwrap() - 0.72).abs() < 1e-10);
        assert!((ob.mid().unwrap() - 0.645).abs() < 1e-10);
        assert!((ob.spread().unwrap() - 0.15).abs() < 1e-10);
    }

    #[test]
    fn parse_asset_price_update() {
        let data = serde_json::json!({
            "price": 67853.57751504,
            "publishTime": 1772898632u64,
            "timestamp": 1772898633u64
        });

        let price = parse_asset_price("assetPriceUpdate/1", &data).unwrap();
        assert_eq!(price.feed_id, 1);
        assert!((price.price - 67853.577).abs() < 1.0);
        assert_eq!(price.publish_time, 1772898632);
        assert_eq!(price.timestamp, 1772898633);
    }

    #[test]
    fn parse_push_message_dispatches_correctly() {
        // Orderbook
        let ob_data = serde_json::json!({"asks": [], "bids": [], "version": 1, "updateTimestampMs": 0, "orderCount": 0});
        assert!(matches!(
            parse_push_message("predictOrderbook/123", ob_data),
            PredictWsMessage::Orderbook(_)
        ));

        // Asset price
        let price_data = serde_json::json!({"price": 100.0, "publishTime": 0, "timestamp": 0});
        assert!(matches!(
            parse_push_message("assetPriceUpdate/1", price_data),
            PredictWsMessage::AssetPrice(_)
        ));

        // Polymarket chance
        let chance_data = serde_json::json!({"chance": 0.5});
        assert!(matches!(
            parse_push_message("polymarketChance/456", chance_data),
            PredictWsMessage::CrossVenueChance(_)
        ));

        // Kalshi chance
        let kalshi_data = serde_json::json!({"chance": 0.3});
        assert!(matches!(
            parse_push_message("kalshiChance/789", kalshi_data),
            PredictWsMessage::CrossVenueChance(_)
        ));

        // Wallet events
        let wallet_data = serde_json::json!({"event": "fill"});
        assert!(matches!(
            parse_push_message("predictWalletEvents/jwt123", wallet_data),
            PredictWsMessage::WalletEvent(_)
        ));

        // Unknown
        let unknown_data = serde_json::json!({"foo": "bar"});
        assert!(matches!(
            parse_push_message("unknown/topic", unknown_data),
            PredictWsMessage::Raw { .. }
        ));
    }

    #[test]
    fn orderbook_snapshot_helpers_empty() {
        let ob = OrderbookSnapshot {
            market_id: 1,
            bids: vec![],
            asks: vec![],
            version: 0,
            update_timestamp_ms: 0,
            order_count: 0,
            last_order_settled: None,
        };
        assert!(ob.best_bid().is_none());
        assert!(ob.best_ask().is_none());
        assert!(ob.mid().is_none());
        assert!(ob.spread().is_none());
    }

    #[test]
    fn feed_id_constants() {
        assert_eq!(feeds::BTC, 1);
        assert_eq!(feeds::ETH, 4);
        assert_eq!(feeds::BNB, 2);
    }

    #[test]
    fn ws_endpoint_constants() {
        assert_eq!(PREDICT_WS_MAINNET, "wss://ws.predict.fun/ws");
        assert_eq!(PREDICT_WS_TESTNET, "wss://ws.bnb.predict.fail/ws");
    }

    #[test]
    fn config_defaults() {
        let config = PredictWsConfig::default();
        assert_eq!(config.url, PREDICT_WS_MAINNET);
        assert_eq!(config.channel_buffer, 1024);
        assert_eq!(config.heartbeat_timeout_secs, 60);
        assert_eq!(config.max_reconnect_attempts, 0);
    }

    #[test]
    fn config_testnet() {
        let config = PredictWsConfig::testnet();
        assert_eq!(config.url, PREDICT_WS_TESTNET);
    }
}