From a7ca8e43418838b219634f5a5bdde67e0ad1ad21 Mon Sep 17 00:00:00 2001 From: floor-licker Date: Thu, 29 Jan 2026 18:35:02 -0500 Subject: [PATCH] feat(ws): align auth and event_type parsing --- examples/demo.rs | 107 +++++++++--------- examples/snipe.rs | 120 +++++++++++++------- src/decode.rs | 89 +++++++++------ src/stream.rs | 221 ++++++++++++------------------------- src/types.rs | 274 ++++++++++++++++++++++++++++++++++++++++------ 5 files changed, 506 insertions(+), 305 deletions(-) diff --git a/examples/demo.rs b/examples/demo.rs index c53be6d..d505e81 100644 --- a/examples/demo.rs +++ b/examples/demo.rs @@ -26,7 +26,7 @@ use polyfill_rs::{ types::*, // Utility functions - utils::{address, crypto, math, rate_limit, retry, time, url}, + utils::{address, math, rate_limit, retry, time, url}, // Configuration ClientConfig, @@ -43,7 +43,7 @@ use rust_decimal::Decimal; use rust_decimal_macros::dec; use std::time::Duration; use tokio::time::sleep; -use tracing::{debug, error, info}; +use tracing::{error, info}; /// Demo showcasing polyfill-rs functionality #[allow(dead_code)] @@ -592,10 +592,9 @@ impl PolyfillDemo { initial_dump: Some(true), custom_feature_enabled: None, auth: Some(WssAuth { - address: "0x1234567890123456789012345678901234567890".to_string(), - signature: "mock_signature".to_string(), - timestamp: time::now_secs(), - nonce: crypto::generate_nonce().to_string(), + api_key: "test-api-key".to_string(), + secret: "test-secret".to_string(), + passphrase: "test-passphrase".to_string(), }), }; @@ -603,33 +602,33 @@ impl PolyfillDemo { // Simulate receiving stream messages let messages = vec![ - StreamMessage::Heartbeat { - timestamp: chrono::Utc::now(), - }, - StreamMessage::BookUpdate { - data: OrderDelta { - token_id: "12345".to_string(), - timestamp: chrono::Utc::now(), - side: Side::BUY, + StreamMessage::Book(BookUpdate { + asset_id: "12345".to_string(), + market: "market1".to_string(), + timestamp: time::now_millis(), + bids: vec![OrderSummary { price: dec!(0.75), size: dec!(100.0), - sequence: 1, - }, - }, - StreamMessage::Trade { - data: FillEvent { - id: "fill1".to_string(), - order_id: "order1".to_string(), - token_id: "12345".to_string(), - side: Side::BUY, - price: dec!(0.75), + }], + asks: vec![OrderSummary { + price: dec!(0.76), size: dec!(50.0), - timestamp: chrono::Utc::now(), - maker_address: alloy_primitives::Address::ZERO, - taker_address: alloy_primitives::Address::ZERO, - fee: dec!(0.375), - }, - }, + }], + hash: None, + }), + StreamMessage::Trade(TradeMessage { + id: "fill1".to_string(), + market: "market1".to_string(), + asset_id: "12345".to_string(), + side: Side::BUY, + size: dec!(50.0), + price: dec!(0.75), + status: Some("MATCHED".to_string()), + msg_type: None, + last_update: None, + matchtime: None, + timestamp: None, + }), ]; for message in messages { @@ -638,31 +637,41 @@ impl PolyfillDemo { // Process message based on type match &message { - StreamMessage::BookUpdate { data } => { - info!(" Processing book update for token: {}", data.token_id); - if let Err(e) = self.book_manager.apply_delta(data.clone()) { - error!(" Failed to apply book update: {}", e); - self.stats.errors += 1; + StreamMessage::Book(book) => { + info!(" Processing book update for asset: {}", book.asset_id); + // This is a demo: apply snapshot levels as deltas. + for level in &book.bids { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: book.asset_id.clone(), + timestamp: chrono::Utc::now(), + side: Side::BUY, + price: level.price, + size: level.size, + sequence: book.timestamp, + }); } - }, - StreamMessage::Trade { data } => { + for level in &book.asks { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: book.asset_id.clone(), + timestamp: chrono::Utc::now(), + side: Side::SELL, + price: level.price, + size: level.size, + sequence: book.timestamp, + }); + } + } + StreamMessage::Trade(trade) => { info!( " Processing trade: {} {} @ {}", - data.side.as_str(), - data.size, - data.price + trade.side.as_str(), + trade.size, + trade.price ); - if let Err(e) = self.fill_processor.process_fill(data.clone()) { - error!(" Failed to process fill: {}", e); - self.stats.errors += 1; - } - }, - StreamMessage::Heartbeat { timestamp } => { - debug!(" Received heartbeat at: {}", timestamp); - }, + } _ => { info!(" Unhandled message type"); - }, + } } } diff --git a/examples/snipe.rs b/examples/snipe.rs index 4c8abb7..4fdd583 100644 --- a/examples/snipe.rs +++ b/examples/snipe.rs @@ -101,43 +101,86 @@ impl SnipeStrategy { /// Process a market data update pub fn process_update(&mut self, message: StreamMessage) -> Result<()> { match message { - StreamMessage::BookUpdate { data } => { - if data.token_id == self.token_id { - self.process_book_update(data)?; + StreamMessage::Book(book) => { + if book.asset_id == self.token_id { + self.process_book_update(book)?; } }, - StreamMessage::Trade { data } => { - if data.token_id == self.token_id { - self.process_trade(data)?; + StreamMessage::Trade(trade) => { + if trade.asset_id == self.token_id { + self.process_trade(trade)?; } }, - StreamMessage::Heartbeat { timestamp: _ } => { - self.check_stale_quotes()?; - }, _ => {}, } + + // Opportunistically check for staleness on any incoming update. + self.check_stale_quotes()?; Ok(()) } /// Process order book update - fn process_book_update(&mut self, delta: OrderDelta) -> Result<()> { + fn process_book_update(&mut self, book: BookUpdate) -> Result<()> { // Ensure book exists self.book_manager.get_or_create_book(&self.token_id)?; - // Update local order book - self.book_manager.apply_delta(delta.clone())?; + // Clear the existing book and rebuild from the snapshot. + if let Ok(current) = self.book_manager.get_book(&self.token_id) { + for level in ¤t.bids { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: self.token_id.clone(), + timestamp: chrono::Utc::now(), + side: Side::BUY, + price: level.price, + size: Decimal::ZERO, + sequence: book.timestamp, + }); + } - // Get current book state - let book = self.book_manager.get_book(&self.token_id)?; + for level in ¤t.asks { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: self.token_id.clone(), + timestamp: chrono::Utc::now(), + side: Side::SELL, + price: level.price, + size: Decimal::ZERO, + sequence: book.timestamp, + }); + } + } - // Update best prices - if let Some(best_bid) = book.bids.first() { - self.last_best_bid = Some(best_bid.price); + let ts = chrono::DateTime::from_timestamp( + (book.timestamp / 1000) as i64, + ((book.timestamp % 1000) * 1_000_000) as u32, + ) + .unwrap_or_else(chrono::Utc::now); + + for level in &book.bids { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: self.token_id.clone(), + timestamp: ts, + side: Side::BUY, + price: level.price, + size: level.size, + sequence: book.timestamp, + }); } - if let Some(best_ask) = book.asks.first() { - self.last_best_ask = Some(best_ask.price); + + for level in &book.asks { + let _ = self.book_manager.apply_delta(OrderDelta { + token_id: self.token_id.clone(), + timestamp: ts, + side: Side::SELL, + price: level.price, + size: level.size, + sequence: book.timestamp, + }); } + // Update best prices directly from the snapshot + self.last_best_bid = book.bids.first().map(|l| l.price); + self.last_best_ask = book.asks.first().map(|l| l.price); + self.last_update = time::now_secs(); // Check for trading opportunities @@ -147,17 +190,17 @@ impl SnipeStrategy { } /// Process trade update - fn process_trade(&mut self, fill: FillEvent) -> Result<()> { + fn process_trade(&mut self, trade: TradeMessage) -> Result<()> { info!( "Trade: {} {} @ {} (size: {})", - fill.side.as_str(), - fill.token_id, - fill.price, - fill.size + trade.side.as_str(), + trade.asset_id, + trade.price, + trade.size ); // Update statistics - self.stats.total_volume += fill.size; + self.stats.total_volume += trade.size; // Calculate P&L if this was our trade // (In a real implementation, you'd track your own orders) @@ -327,24 +370,19 @@ impl MockMarketData { let price_change = random_factor * Decimal::from(2) * self.volatility; let new_price = self.base_price * (Decimal::from(1) + price_change); - // Generate order book update - let side = if rand::random::() { - Side::BUY - } else { - Side::SELL - }; + // Generate a simple orderbook snapshot update let size = Decimal::from(rand::random::() % 1000 + 100); + let bid = new_price - dec!(0.01); + let ask = new_price + dec!(0.01); - StreamMessage::BookUpdate { - data: OrderDelta { - token_id: self.token_id.clone(), - timestamp: chrono::Utc::now(), - side, - price: new_price, - size, - sequence: self.sequence, - }, - } + StreamMessage::Book(BookUpdate { + asset_id: self.token_id.clone(), + market: "0xmock".to_string(), + timestamp: time::now_millis(), + bids: vec![OrderSummary { price: bid, size }], + asks: vec![OrderSummary { price: ask, size }], + hash: None, + }) } } diff --git a/src/decode.rs b/src/decode.rs index aabbc73..18c2715 100644 --- a/src/decode.rs +++ b/src/decode.rs @@ -432,44 +432,61 @@ impl Decoder for RawMarketResponse { } } -/// WebSocket message parsing -pub fn parse_stream_message(raw: &str) -> Result { - let value: Value = serde_json::from_str(raw)?; +/// WebSocket message parsing (official `event_type` shape). +/// +/// Polymarket WebSocket servers may send either a single JSON object or a batch array. +/// This parser is tolerant: +/// - Unknown/unsupported `event_type`s are ignored. +/// - Invalid entries inside a batch are skipped (do not fail the whole batch). +pub fn parse_stream_messages(raw: &str) -> Result> { + parse_stream_messages_bytes(raw.as_bytes()) +} - let msg_type = value["type"] - .as_str() - .ok_or_else(|| PolyfillError::parse("Missing message type".to_string(), None))?; +/// See `parse_stream_messages`. +pub fn parse_stream_messages_bytes(bytes: &[u8]) -> Result> { + let value: Value = serde_json::from_slice(bytes)?; - match msg_type { - "book_update" => { - let data = value["data"].clone(); - let delta: OrderDelta = serde_json::from_value(data)?; - Ok(StreamMessage::BookUpdate { data: delta }) - }, - "trade" => { - let data = value["data"].clone(); - let raw_trade: RawTradeResponse = serde_json::from_value(data)?; - let fill = raw_trade.decode()?; - Ok(StreamMessage::Trade { data: fill }) - }, - "order_update" => { - let data = value["data"].clone(); - let raw_order: RawOrderResponse = serde_json::from_value(data)?; - let order = raw_order.decode()?; - Ok(StreamMessage::OrderUpdate { data: order }) - }, - "heartbeat" => { - let timestamp = value["timestamp"] - .as_str() - .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) - .map(|dt| dt.with_timezone(&Utc)) - .unwrap_or_else(Utc::now); - Ok(StreamMessage::Heartbeat { timestamp }) - }, - _ => Err(PolyfillError::parse( - format!("Unknown message type: {}", msg_type), - None, - )), + match value { + Value::Object(map) => { + let event_type = map.get("event_type").and_then(Value::as_str); + match event_type { + None => Ok(vec![]), + Some(_) => { + let msg: StreamMessage = serde_json::from_value(Value::Object(map))?; + match msg { + StreamMessage::Unknown => Ok(vec![]), + other => Ok(vec![other]), + } + } + } + } + Value::Array(arr) => Ok(arr + .into_iter() + .filter_map(|elem| { + let obj = elem.as_object()?; + let event_type = obj.get("event_type").and_then(Value::as_str)?; + // Skip unknown event types early (forward compatibility). + match event_type { + "book" + | "price_change" + | "tick_size_change" + | "last_trade_price" + | "best_bid_ask" + | "new_market" + | "market_resolved" + | "trade" + | "order" => {} + _ => return None, + } + + match serde_json::from_value::(Value::Object(obj.clone())) { + Ok(StreamMessage::Unknown) => None, + Ok(msg) => Some(msg), + Err(_) => None, + } + }) + .collect()), + _ => Ok(vec![]), } } diff --git a/src/stream.rs b/src/stream.rs index 19b7342..dbca228 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -274,12 +274,12 @@ impl WebSocketStream { tokio_tungstenite::tungstenite::Message::Text(text) => { debug!("Received WebSocket message: {}", text); - // Parse the message according to Polymarket's format - let stream_message = self.parse_polymarket_message(&text)?; - - // Send to internal channel - if let Err(e) = self.tx.send(stream_message) { - error!("Failed to send message to internal channel: {}", e); + // Parse the message according to Polymarket's `event_type` format + let stream_messages = crate::decode::parse_stream_messages(&text)?; + for stream_message in stream_messages { + if let Err(e) = self.tx.send(stream_message) { + error!("Failed to send message to internal channel: {}", e); + } } self.stats.messages_received += 1; @@ -313,115 +313,10 @@ impl WebSocketStream { Ok(()) } - /// Parse Polymarket WebSocket message format + /// Parse Polymarket WebSocket message(s) in `event_type` format. #[allow(dead_code)] - fn parse_polymarket_message(&self, text: &str) -> Result { - let value: Value = serde_json::from_str(text).map_err(|e| { - PolyfillError::parse( - format!("Failed to parse WebSocket message: {}", e), - Some(Box::new(e)), - ) - })?; - - // Extract message type - let message_type = value.get("type").and_then(|v| v.as_str()).ok_or_else(|| { - PolyfillError::parse("Missing 'type' field in WebSocket message", None) - })?; - - match message_type { - "book_update" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse book update: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::BookUpdate { data }) - }, - "trade" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse trade: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::Trade { data }) - }, - "order_update" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse order update: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::OrderUpdate { data }) - }, - "user_order_update" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse user order update: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::UserOrderUpdate { data }) - }, - "user_trade" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse user trade: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::UserTrade { data }) - }, - "market_book_update" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse market book update: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::MarketBookUpdate { data }) - }, - "market_trade" => { - let data = - serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone()) - .map_err(|e| { - PolyfillError::parse( - format!("Failed to parse market trade: {}", e), - Some(Box::new(e)), - ) - })?; - Ok(StreamMessage::MarketTrade { data }) - }, - "heartbeat" => { - let timestamp = value - .get("timestamp") - .and_then(|v| v.as_u64()) - .map(|ts| chrono::DateTime::from_timestamp(ts as i64, 0).unwrap_or_default()) - .unwrap_or_else(Utc::now); - Ok(StreamMessage::Heartbeat { timestamp }) - }, - _ => { - warn!("Unknown message type: {}", message_type); - // Return heartbeat as fallback - Ok(StreamMessage::Heartbeat { - timestamp: Utc::now(), - }) - }, - } + fn parse_polymarket_messages(&self, text: &str) -> Result> { + crate::decode::parse_stream_messages(text) } /// Reconnect with exponential backoff @@ -476,33 +371,58 @@ impl Stream for WebSocketStream { type Item = Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // First check internal channel - if let Poll::Ready(Some(message)) = self.rx.poll_recv(cx) { - return Poll::Ready(Some(Ok(message))); - } + loop { + // First drain any parsed messages + if let Poll::Ready(Some(message)) = self.rx.poll_recv(cx) { + return Poll::Ready(Some(Ok(message))); + } + + let Some(connection) = &mut self.connection else { + return Poll::Ready(None); + }; - // Then check WebSocket connection - if let Some(connection) = &mut self.connection { match connection.poll_next_unpin(cx) { - Poll::Ready(Some(Ok(_message))) => { - // Simplified message handling - Poll::Ready(Some(Ok(StreamMessage::Heartbeat { - timestamp: Utc::now(), - }))) + Poll::Pending => return Poll::Pending, + Poll::Ready(Some(Ok(ws_message))) => match ws_message { + tokio_tungstenite::tungstenite::Message::Text(text) => { + match crate::decode::parse_stream_messages(&text) { + Ok(messages) => { + for msg in messages { + let _ = self.tx.send(msg); + } + self.stats.messages_received += 1; + self.stats.last_message_time = Some(Utc::now()); + continue; + } + Err(e) => { + self.stats.errors += 1; + return Poll::Ready(Some(Err(e))); + } + } + } + tokio_tungstenite::tungstenite::Message::Close(_) => { + info!("WebSocket connection closed by server"); + self.connection = None; + return Poll::Ready(None); + } + tokio_tungstenite::tungstenite::Message::Ping(_) => { + // Best-effort: tokio-tungstenite/tungstenite may handle pings internally. + continue; + } + tokio_tungstenite::tungstenite::Message::Pong(_) => continue, + tokio_tungstenite::tungstenite::Message::Binary(_) => continue, + tokio_tungstenite::tungstenite::Message::Frame(_) => continue, }, Poll::Ready(Some(Err(e))) => { error!("WebSocket error: {}", e); self.stats.errors += 1; - Poll::Ready(Some(Err(e.into()))) - }, + return Poll::Ready(Some(Err(e.into()))); + } Poll::Ready(None) => { info!("WebSocket stream ended"); - Poll::Ready(None) - }, - Poll::Pending => Poll::Pending, + return Poll::Ready(None); + } } - } else { - Poll::Ready(None) } } } @@ -655,19 +575,19 @@ mod tests { let mut stream = MockStream::new(); // Add some test messages - stream.add_message(StreamMessage::Heartbeat { - timestamp: Utc::now(), - }); - stream.add_message(StreamMessage::BookUpdate { - data: OrderDelta { - token_id: "test".to_string(), - timestamp: Utc::now(), - side: Side::BUY, - price: rust_decimal_macros::dec!(0.5), - size: rust_decimal_macros::dec!(100), - sequence: 1, - }, - }); + stream.add_message(StreamMessage::Book(BookUpdate { + asset_id: "1".to_string(), + market: "0xabc".to_string(), + timestamp: 1_234_567_890, + bids: vec![], + asks: vec![], + hash: None, + })); + stream.add_message(StreamMessage::PriceChange(PriceChange { + market: "0xabc".to_string(), + timestamp: 1_234_567_891, + price_changes: vec![], + })); assert!(stream.is_connected()); assert_eq!(stream.get_stats().messages_received, 2); @@ -680,9 +600,14 @@ mod tests { manager.add_stream(mock_stream); // Test message broadcasting - let message = StreamMessage::Heartbeat { - timestamp: Utc::now(), - }; + let message = StreamMessage::Book(BookUpdate { + asset_id: "1".to_string(), + market: "0xabc".to_string(), + timestamp: 1_234_567_890, + bids: vec![], + asks: vec![], + hash: None, + }); assert!(manager.broadcast_message(message).is_ok()); } } diff --git a/src/types.rs b/src/types.rs index ca6041c..bd6ed89 100644 --- a/src/types.rs +++ b/src/types.rs @@ -659,18 +659,11 @@ impl Default for ClientConfig { } } -/// WebSocket authentication for Polymarket API -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct WssAuth { - /// User's Ethereum address - pub address: String, - /// EIP-712 signature - pub signature: String, - /// Unix timestamp - pub timestamp: u64, - /// Nonce for replay protection - pub nonce: String, -} +/// WebSocket authentication for Polymarket API user channel. +/// +/// Polymarket's CLOB WebSocket expects the same L2 API credentials used for HTTP calls: +/// `{ apiKey, secret, passphrase }`. +pub type WssAuth = ApiCredentials; /// WebSocket subscription request #[derive(Debug, Clone, Serialize, Deserialize)] @@ -699,28 +692,247 @@ pub struct WssSubscription { pub auth: Option, } -/// WebSocket message types for streaming +/// WebSocket message types for streaming (official Polymarket `event_type` format). #[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "type")] +#[serde(tag = "event_type")] pub enum StreamMessage { - #[serde(rename = "book_update")] - BookUpdate { data: OrderDelta }, + /// Full or incremental orderbook update + #[serde(rename = "book")] + Book(BookUpdate), + /// Price change notification (single or batched) + #[serde(rename = "price_change")] + PriceChange(PriceChange), + /// Tick size change notification + #[serde(rename = "tick_size_change")] + TickSizeChange(TickSizeChange), + /// Last trade price update + #[serde(rename = "last_trade_price")] + LastTradePrice(LastTradePrice), + /// Best bid/ask update (requires `custom_feature_enabled`) + #[serde(rename = "best_bid_ask")] + BestBidAsk(BestBidAsk), + /// New market created (requires `custom_feature_enabled`) + #[serde(rename = "new_market")] + NewMarket(NewMarket), + /// Market resolved (requires `custom_feature_enabled`) + #[serde(rename = "market_resolved")] + MarketResolved(MarketResolved), + /// User trade execution (authenticated channel) #[serde(rename = "trade")] - Trade { data: FillEvent }, - #[serde(rename = "order_update")] - OrderUpdate { data: Order }, - #[serde(rename = "heartbeat")] - Heartbeat { timestamp: DateTime }, - /// User channel events - #[serde(rename = "user_order_update")] - UserOrderUpdate { data: Order }, - #[serde(rename = "user_trade")] - UserTrade { data: FillEvent }, - /// Market channel events - #[serde(rename = "market_book_update")] - MarketBookUpdate { data: OrderDelta }, - #[serde(rename = "market_trade")] - MarketTrade { data: FillEvent }, + Trade(TradeMessage), + /// User order update (authenticated channel) + #[serde(rename = "order")] + Order(OrderMessage), + /// Forward-compatible catch-all for new/unknown event types. + #[serde(other)] + Unknown, +} + +/// Orderbook update message (full snapshot or delta). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BookUpdate { + pub asset_id: String, + pub market: String, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, + #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")] + pub bids: Vec, + #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")] + pub asks: Vec, + #[serde(default)] + pub hash: Option, +} + +/// Unified wire format for `price_change` events. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PriceChange { + pub market: String, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, + #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")] + pub price_changes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PriceChangeEntry { + pub asset_id: String, + pub price: Decimal, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub size: Option, + pub side: Side, + #[serde(default)] + pub hash: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub best_bid: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub best_ask: Option, +} + +/// Tick size change event. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TickSizeChange { + pub asset_id: String, + pub market: String, + pub old_tick_size: Decimal, + pub new_tick_size: Decimal, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, +} + +/// Last trade price update. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LastTradePrice { + pub asset_id: String, + pub market: String, + pub price: Decimal, + #[serde(default)] + pub side: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub size: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub fee_rate_bps: Option, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, +} + +/// Best bid/ask update. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BestBidAsk { + pub market: String, + pub asset_id: String, + pub best_bid: Decimal, + pub best_ask: Decimal, + pub spread: Decimal, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, +} + +/// New market created event. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NewMarket { + pub id: String, + pub question: String, + pub market: String, + pub slug: String, + pub description: String, + #[serde(rename = "assets_ids", alias = "asset_ids")] + pub asset_ids: Vec, + #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")] + pub outcomes: Vec, + #[serde(default)] + pub event_message: Option, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, +} + +/// Market resolved event. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MarketResolved { + pub id: String, + #[serde(default)] + pub question: Option, + pub market: String, + #[serde(default)] + pub slug: Option, + #[serde(default)] + pub description: Option, + #[serde(rename = "assets_ids", alias = "asset_ids")] + pub asset_ids: Vec, + #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")] + pub outcomes: Vec, + pub winning_asset_id: String, + pub winning_outcome: String, + #[serde(default)] + pub event_message: Option, + #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")] + pub timestamp: u64, +} + +/// Event message object for market events. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventMessage { + pub id: String, + pub ticker: String, + pub slug: String, + pub title: String, + pub description: String, +} + +/// User trade execution message. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TradeMessage { + pub id: String, + pub market: String, + pub asset_id: String, + pub side: Side, + pub size: Decimal, + pub price: Decimal, + #[serde(default)] + pub status: Option, + #[serde(rename = "type", default)] + pub msg_type: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_number_from_string" + )] + pub last_update: Option, + #[serde( + default, + alias = "match_time", + deserialize_with = "crate::decode::deserializers::optional_number_from_string" + )] + pub matchtime: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_number_from_string" + )] + pub timestamp: Option, +} + +/// User order update message. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct OrderMessage { + pub id: String, + pub market: String, + pub asset_id: String, + pub side: Side, + pub price: Decimal, + #[serde(rename = "type", default)] + pub msg_type: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub original_size: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_decimal_from_string" + )] + pub size_matched: Option, + #[serde( + default, + deserialize_with = "crate::decode::deserializers::optional_number_from_string" + )] + pub timestamp: Option, + #[serde(default)] + pub associate_trades: Option>, + #[serde(default)] + pub status: Option, } /// Subscription parameters for streaming