feat(rtds): chainlink twap price streams

src/rtds/client.rs

@@ -10,6 +10,8 @@ use crate::Result;
 use crate::auth::state::{Authenticated, State, Unauthenticated};
 use crate::auth::{Credentials, Normal};
 use crate::error::Error;
+use crate::rtds::types::request::ChainlinkTwapWindow;
+use crate::rtds::types::response::ChainlinkTwapPrice;
 use crate::types::Address;
 use crate::ws::ConnectionManager;
 use crate::ws::config::Config;
@@ -122,7 +124,7 @@ impl Client<Unauthenticated> {
     pub fn subscribe_comments(
         &self,
         comment_type: Option<CommentType>,
-    ) -> Result<impl Stream<Item = Result<Comment>>> {
+    ) -> Result<impl Stream<Item = Result<Comment>> + use<>> {
         let subscription = Subscription::comments(comment_type);
         let stream = self.inner.subscriptions.subscribe(subscription)?;
 
@@ -177,7 +179,7 @@ impl<S: State> Client<S> {
     pub fn subscribe_crypto_prices(
         &self,
         symbols: Option<Vec<String>>,
-    ) -> Result<impl Stream<Item = Result<CryptoPrice>>> {
+    ) -> Result<impl Stream<Item = Result<CryptoPrice>> + use<S>> {
         let subscription = Subscription::crypto_prices(symbols);
         let stream = self.inner.subscriptions.subscribe(subscription)?;
 
@@ -193,7 +195,7 @@ impl<S: State> Client<S> {
     pub fn subscribe_chainlink_prices(
         &self,
         symbol: Option<String>,
-    ) -> Result<impl Stream<Item = Result<ChainlinkPrice>>> {
+    ) -> Result<impl Stream<Item = Result<ChainlinkPrice>> + use<S>> {
         let subscription = Subscription::chainlink_prices(symbol);
         let stream = self.inner.subscriptions.subscribe(subscription)?;
 
@@ -205,11 +207,28 @@ impl<S: State> Client<S> {
         }))
     }
 
+    /// Subscribe to Chainlink time-weighted average price (TWAP) feed updates.
+    pub fn subscribe_chainlink_twap_prices(
+        &self,
+        symbol: Option<String>,
+        twap_window: ChainlinkTwapWindow,
+    ) -> Result<impl Stream<Item = Result<ChainlinkTwapPrice>> + use<S>> {
+        let subscription = Subscription::chainlink_twap_prices(symbol, twap_window);
+        let stream = self.inner.subscriptions.subscribe(subscription)?;
+
+        Ok(stream.filter_map(|msg_result| async move {
+            match msg_result {
+                Ok(msg) => msg.as_chainlink_twap_price().map(Ok),
+                Err(e) => Some(Err(e)),
+            }
+        }))
+    }
+
     /// Subscribe to raw RTDS messages for a custom topic/type combination.
     pub fn subscribe_raw(
         &self,
         subscription: Subscription,
-    ) -> Result<impl Stream<Item = Result<RtdsMessage>>> {
+    ) -> Result<impl Stream<Item = Result<RtdsMessage>> + use<S>> {
         self.inner.subscriptions.subscribe(subscription)
     }
 
@@ -273,6 +292,37 @@ impl<S: State> Client<S> {
         self.inner.subscriptions.unsubscribe(&[topic])
     }
 
+    /// Unsubscribe from Chainlink TWAP price feed updates.
+    ///
+    /// This decrements the reference count for the chainlink TWAP topic. Only sends
+    /// an unsubscribe request to the server when no other streams are using this topic.
+    ///
+    /// If `twap_window` is `None`, will unsubscribe from all windows
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the unsubscribe request fails.
+    pub fn unsubscribe_chainlink_twap_prices(
+        &self,
+        twap_window: impl Into<Option<ChainlinkTwapWindow>>,
+    ) -> Result<()> {
+        let topics = if let Some(twap_window) = twap_window.into() {
+            vec![TopicType::new(
+                twap_window.to_topic_string(),
+                "*".to_owned(),
+            )]
+        } else {
+            [
+                ChainlinkTwapWindow::ThirtySeconds,
+                ChainlinkTwapWindow::SixtySeconds,
+            ]
+            .into_iter()
+            .map(|w| TopicType::new(w.to_topic_string(), "*".to_owned()))
+            .collect()
+        };
+        self.inner.subscriptions.unsubscribe(&topics)
+    }
+
     /// Unsubscribe from comment events.
     ///
     /// # Arguments
@@ -299,7 +349,7 @@ impl Client<Authenticated<Normal>> {
     pub fn subscribe_comments(
         &self,
         comment_type: Option<CommentType>,
-    ) -> Result<impl Stream<Item = Result<Comment>>> {
+    ) -> Result<impl Stream<Item = Result<Comment>> + use<>> {
         let subscription = Subscription::comments(comment_type)
             .with_clob_auth(self.inner.state.credentials.clone());
         let stream = self.inner.subscriptions.subscribe(subscription)?;

src/rtds/mod.rs

@@ -47,7 +47,10 @@ pub mod types;
 pub use client::Client;
 pub use error::RtdsError;
 pub use subscription::SubscriptionInfo;
-pub use types::request::{Subscription, SubscriptionAction, SubscriptionRequest};
+pub use types::request::{
+    ChainlinkTwapWindow, Subscription, SubscriptionAction, SubscriptionRequest,
+};
 pub use types::response::{
-    ChainlinkPrice, Comment, CommentProfile, CommentType, CryptoPrice, RtdsMessage,
+    ChainlinkPrice, ChainlinkTwapPrice, Comment, CommentProfile, CommentType, CryptoPrice,
+    RtdsMessage,
 };

src/rtds/subscription.rs

@@ -176,7 +176,7 @@ impl SubscriptionManager {
     pub fn subscribe(
         &self,
         subscription: Subscription,
-    ) -> Result<impl Stream<Item = Result<RtdsMessage>>> {
+    ) -> Result<impl Stream<Item = Result<RtdsMessage>> + use<>> {
         let topic_type = TopicType::new(subscription.topic.clone(), subscription.msg_type.clone());
 
         // Store auth for re-subscription on reconnect.

src/rtds/types/request.rs

@@ -1,3 +1,5 @@
+use std::fmt::Display;
+
 use bon::Builder;
 use secrecy::ExposeSecret as _;
 use serde::Serialize;
@@ -6,6 +8,13 @@ use serde_json::Value;
 use super::response::CommentType;
 use crate::auth::Credentials;
 
+pub(crate) const CHAINLINK_CRYPTO_PRICE_TOPIC: &str = "crypto_prices_chainlink";
+pub(crate) const CHAINLINK_TWAP_TOPIC_PREFIX: &str = "crypto_prices_twap";
+
+fn is_chainlink_topic(topic: &str) -> bool {
+    topic == CHAINLINK_CRYPTO_PRICE_TOPIC || topic.starts_with(CHAINLINK_TWAP_TOPIC_PREFIX)
+}
+
 /// RTDS subscription request message.
 #[non_exhaustive]
 #[derive(Clone, Debug, Serialize, Builder)]
@@ -87,7 +96,19 @@ impl Subscription {
     pub fn chainlink_prices(symbol: Option<String>) -> Self {
         let filters = symbol.map(|s| format!(r#"{{"symbol":"{s}"}}"#));
         Self {
-            topic: "crypto_prices_chainlink".to_owned(),
+            topic: CHAINLINK_CRYPTO_PRICE_TOPIC.to_owned(),
+            msg_type: "*".to_owned(),
+            filters,
+            clob_auth: None,
+        }
+    }
+
+    /// Create a subscription for Chainlink time-weighted average crypto prices (TWAP).
+    #[must_use]
+    pub fn chainlink_twap_prices(symbol: Option<String>, twap_window: ChainlinkTwapWindow) -> Self {
+        let filters = symbol.map(|s| format!(r#"{{"symbol":"{s}"}}"#));
+        Self {
+            topic: twap_window.to_topic_string(),
             msg_type: "*".to_owned(),
             filters,
             clob_auth: None,
@@ -143,7 +164,7 @@ impl Serialize for Subscription {
             // Chainlink endpoint expects filters as a JSON string (escaped),
             // while other endpoints (like Binance crypto_prices) expect raw JSON.
             // See: https://github.com/Polymarket/rs-clob-client/issues/136
-            if self.topic == "crypto_prices_chainlink" {
+            if is_chainlink_topic(&self.topic) {
                 // Chainlink: emit filters as string, e.g. "{\"symbol\":\"btc/usd\"}"
                 map.serialize_entry("filters", filters)?;
             } else if let Ok(json_value) = serde_json::from_str::<Value>(filters) {
@@ -170,6 +191,31 @@ impl Serialize for Subscription {
     }
 }
 
+#[derive(Debug, Clone, Copy)]
+#[non_exhaustive]
+pub enum ChainlinkTwapWindow {
+    ThirtySeconds,
+    SixtySeconds,
+}
+
+impl ChainlinkTwapWindow {
+    #[must_use]
+    pub fn to_topic_string(&self) -> String {
+        format!("{CHAINLINK_TWAP_TOPIC_PREFIX}_{self}")
+    }
+}
+
+impl Display for ChainlinkTwapWindow {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let self_str = match self {
+            Self::ThirtySeconds => "thirty",
+            Self::SixtySeconds => "sixty",
+        };
+
+        f.write_str(self_str)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -203,6 +249,25 @@ mod tests {
         );
     }
 
+    #[test]
+    fn serialize_chainlink_twap_subscription() {
+        let sub = Subscription::chainlink_twap_prices(
+            "eth/usd".to_owned().into(),
+            ChainlinkTwapWindow::SixtySeconds,
+        );
+        let request = SubscriptionRequest::subscribe(vec![sub]);
+
+        let json = serde_json::to_string(&request).unwrap();
+        assert!(json.contains("\"topic\":\"crypto_prices_twap_sixty\""));
+        assert!(json.contains("\"type\":\"*\""));
+        // Chainlink filters should be a JSON string (escaped), not a raw JSON object
+        // See: https://github.com/Polymarket/rs-clob-client/issues/136
+        assert!(
+            json.contains(r#""filters":"{\"symbol\":\"eth/usd\"}""#),
+            "Chainlink filters should be serialized as escaped JSON string, got: {json}"
+        );
+    }
+
     #[test]
     fn serialize_comments_subscription() {
         let sub = Subscription::comments(Some(CommentType::CommentCreated));
@@ -224,6 +289,17 @@ mod tests {
         assert!(!json.contains("\"filters\""));
     }
 
+    #[test]
+    fn serialize_chainlink_twap_without_filters() {
+        // When no symbol is provided, there should be no filters field
+        let sub = Subscription::chainlink_twap_prices(None, ChainlinkTwapWindow::ThirtySeconds);
+        let request = SubscriptionRequest::subscribe(vec![sub]);
+
+        let json = serde_json::to_string(&request).unwrap();
+        assert!(json.contains("\"topic\":\"crypto_prices_twap_thirty\""));
+        assert!(!json.contains("\"filters\""));
+    }
+
     #[test]
     fn serialize_crypto_prices_without_filters() {
         // When no symbols are provided, there should be no filters field
@@ -296,6 +372,20 @@ mod tests {
         assert!(json.contains("\"type\":\"*\""));
     }
 
+    #[test]
+    fn serialize_unsubscribe_chainlink_twap() {
+        let sub = Subscription::chainlink_twap_prices(
+            "btc/usd".to_owned().into(),
+            ChainlinkTwapWindow::SixtySeconds,
+        );
+        let request = SubscriptionRequest::unsubscribe(vec![sub]);
+
+        let json = serde_json::to_string(&request).unwrap();
+        assert!(json.contains("\"action\":\"unsubscribe\""));
+        assert!(json.contains("\"topic\":\"crypto_prices_twap_sixty\""));
+        assert!(json.contains("\"type\":\"*\""));
+    }
+
     #[test]
     fn serialize_unsubscribe_comments() {
         let sub = Subscription::comments(Some(CommentType::CommentCreated));

src/rtds/types/response.rs

@@ -3,6 +3,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+use crate::rtds::types::request::{CHAINLINK_CRYPTO_PRICE_TOPIC, CHAINLINK_TWAP_TOPIC_PREFIX};
 use crate::types::{Address, Decimal};
 
 /// Top-level RTDS message wrapper.
@@ -36,7 +37,17 @@ impl RtdsMessage {
     /// Try to extract the payload as a Chainlink price update.
     #[must_use]
     pub fn as_chainlink_price(&self) -> Option<ChainlinkPrice> {
-        if self.topic == "crypto_prices_chainlink" {
+        if self.topic == CHAINLINK_CRYPTO_PRICE_TOPIC {
+            serde_json::from_value(self.payload.clone()).ok()
+        } else {
+            None
+        }
+    }
+
+    /// Try to extract the payload as a Chainlink price update.
+    #[must_use]
+    pub fn as_chainlink_twap_price(&self) -> Option<ChainlinkTwapPrice> {
+        if self.topic.starts_with(CHAINLINK_TWAP_TOPIC_PREFIX) {
             serde_json::from_value(self.payload.clone()).ok()
         } else {
             None
@@ -78,6 +89,19 @@ pub struct ChainlinkPrice {
     pub value: Decimal,
 }
 
+/// Chainlink price feed update payload.
+#[non_exhaustive]
+#[derive(Debug, Clone, Deserialize, Serialize, Builder)]
+pub struct ChainlinkTwapPrice {
+    /// Trading pair symbol (slash-separated, e.g., "eth/usd", "btc/usd")
+    pub symbol: String,
+    /// Price timestamp in Unix milliseconds
+    pub timestamp: i64,
+    /// Current price value
+    pub value: Decimal,
+    pub window_s: i64,
+}
+
 /// Comment event payload.
 #[non_exhaustive]
 #[derive(Debug, Clone, Deserialize, Serialize, Builder)]
@@ -231,6 +255,34 @@ mod tests {
         assert_eq!(price.value, dec!(3456.78));
     }
 
+    #[test]
+    fn parse_chainlink_twap_price_message() {
+        let json = r#"{
+            "topic": "crypto_prices_twap_thirty",
+            "type": "update",
+            "timestamp": 1785178800123,
+            "payload": {
+                "symbol": "btc/usd",
+                "value": 65000.51234,
+                "full_accuracy_value": "65000512340000000000000",
+                "timestamp": 1785178800000,
+                "window_s": 30
+            }
+        }"#;
+
+        let msgs = parse_messages(json.as_bytes()).unwrap();
+        assert_eq!(msgs.len(), 1);
+
+        let msg = &msgs[0];
+        assert_eq!(msg.topic, "crypto_prices_twap_thirty");
+
+        let price = msg.as_chainlink_twap_price().unwrap();
+        assert_eq!(price.symbol, "btc/usd");
+        assert_eq!(price.value, dec!(65000.51234));
+        assert_eq!(price.window_s, 30);
+        assert_eq!(price.timestamp, 1785178800000);
+    }
+
     #[test]
     fn parse_comment_message() {
         let json = r#"{