strata_sdk/
market_stream.rs

1use futures_util::{SinkExt, StreamExt};
2use tokio::net::TcpStream;
3use tokio_tungstenite::tungstenite::Message;
4use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
5
6use super::*;
7
8type PlatformSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
9
10#[derive(Clone, Debug)]
11struct BookCursor {
12    stream_id: String,
13    sequence: u64,
14}
15
16/// One sequenced connection to Strata's product-level market stream.
17///
18/// The stream returns only public book, best-price, trade, and market-status
19/// events. A malformed frame or sequence gap returns an error and closes the
20/// connection; reconnecting obtains a fresh snapshot.
21pub struct MarketDataStream {
22    market_id: String,
23    socket: PlatformSocket,
24    book: Option<BookCursor>,
25}
26
27impl std::fmt::Debug for MarketDataStream {
28    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        formatter
30            .debug_struct("MarketDataStream")
31            .field("market_id", &self.market_id)
32            .field("has_book_snapshot", &self.book.is_some())
33            .finish_non_exhaustive()
34    }
35}
36
37impl MarketDataStream {
38    pub(crate) async fn connect(client: &StrataClient, market_id: &str) -> Result<Self, SdkError> {
39        let market_id = validate_platform_market_id(market_id)?;
40        let url =
41            platform_websocket_url(&client.base_url, &format!("v2/markets/{market_id}/stream"))?;
42        let (socket, _) = tokio_tungstenite::connect_async(url.as_str())
43            .await
44            .map_err(|error| SdkError::Stream(error.to_string()))?;
45        Ok(Self {
46            market_id,
47            socket,
48            book: None,
49        })
50    }
51
52    pub fn market_id(&self) -> &str {
53        &self.market_id
54    }
55
56    /// Receive and validate the next public event. `Ok(None)` means the peer
57    /// closed cleanly. Callers should reconnect after any error or close.
58    pub async fn next_event(&mut self) -> Result<Option<PlatformMarketDataEvent>, SdkError> {
59        loop {
60            let Some(frame) = self.socket.next().await else {
61                return Ok(None);
62            };
63            let frame = frame.map_err(|error| SdkError::Stream(error.to_string()))?;
64            match frame {
65                Message::Text(text) => {
66                    let event: PlatformMarketDataEvent = serde_json::from_str(&text)
67                        .map_err(|error| SdkError::InvalidResponse(error.to_string()))?;
68                    if let Err(error) = self.validate_event(&event) {
69                        let _ = self.socket.close(None).await;
70                        return Err(error);
71                    }
72                    return Ok(Some(event));
73                }
74                Message::Ping(payload) => {
75                    self.socket
76                        .send(Message::Pong(payload))
77                        .await
78                        .map_err(|error| SdkError::Stream(error.to_string()))?;
79                }
80                Message::Pong(_) => {}
81                Message::Close(_) => return Ok(None),
82                _ => {
83                    let _ = self.socket.close(None).await;
84                    return Err(SdkError::InvalidResponse(
85                        "market stream sent a non-text data frame".to_owned(),
86                    ));
87                }
88            }
89        }
90    }
91
92    pub async fn close(&mut self) -> Result<(), SdkError> {
93        self.socket
94            .close(None)
95            .await
96            .map_err(|error| SdkError::Stream(error.to_string()))
97    }
98
99    fn validate_event(&mut self, event: &PlatformMarketDataEvent) -> Result<(), SdkError> {
100        match event {
101            PlatformMarketDataEvent::BookSnapshot {
102                schema_version,
103                contract_version,
104                market_id,
105                stream_id,
106                sequence,
107                snapshot_id,
108                bids,
109                asks,
110                ..
111            } => {
112                self.validate_identity(*schema_version, contract_version, market_id)?;
113                let sequence = validate_response_atoms(sequence, "sequence", false)?;
114                if stream_id.trim().is_empty() || snapshot_id.trim().is_empty() {
115                    return Err(SdkError::InvalidResponse(
116                        "market stream snapshot identity is invalid".to_owned(),
117                    ));
118                }
119                validate_book_levels(bids, asks)?;
120                self.book = Some(BookCursor {
121                    stream_id: stream_id.clone(),
122                    sequence,
123                });
124            }
125            PlatformMarketDataEvent::BookDelta {
126                schema_version,
127                contract_version,
128                market_id,
129                stream_id,
130                sequence,
131                previous_sequence,
132                changes,
133                ..
134            } => {
135                self.validate_identity(*schema_version, contract_version, market_id)?;
136                let next = validate_response_atoms(sequence, "sequence", false)?;
137                let previous =
138                    validate_response_atoms(previous_sequence, "previous_sequence", false)?;
139                let cursor = self.book.as_mut().ok_or_else(|| {
140                    SdkError::InvalidResponse(
141                        "book delta arrived before a recoverable snapshot".to_owned(),
142                    )
143                })?;
144                if &cursor.stream_id != stream_id
145                    || previous != cursor.sequence
146                    || next != previous.saturating_add(1)
147                {
148                    return Err(SdkError::InvalidResponse(
149                        "market stream sequence gap detected".to_owned(),
150                    ));
151                }
152                for change in changes {
153                    validate_response_atoms(&change.price_atoms, "price_atoms", false)?;
154                    validate_response_atoms(&change.size_atoms, "size_atoms", true)?;
155                }
156                cursor.sequence = next;
157            }
158            PlatformMarketDataEvent::BestBidAsk {
159                schema_version,
160                contract_version,
161                market_id,
162                sequence,
163                best_bid,
164                best_ask,
165                ..
166            } => {
167                self.validate_identity(*schema_version, contract_version, market_id)?;
168                validate_response_atoms(sequence, "sequence", false)?;
169                if let Some(level) = best_bid {
170                    validate_book_level(level)?;
171                }
172                if let Some(level) = best_ask {
173                    validate_book_level(level)?;
174                }
175            }
176            PlatformMarketDataEvent::Trade {
177                schema_version,
178                contract_version,
179                market_id,
180                trade,
181                ..
182            } => {
183                self.validate_identity(*schema_version, contract_version, market_id)?;
184                if trade.trade_id.trim().is_empty() {
185                    return Err(SdkError::InvalidResponse(
186                        "market stream trade identity is invalid".to_owned(),
187                    ));
188                }
189                validate_response_atoms(&trade.price_atoms, "price_atoms", false)?;
190                validate_response_atoms(&trade.size_atoms, "size_atoms", false)?;
191            }
192            PlatformMarketDataEvent::MarketStatus {
193                schema_version,
194                contract_version,
195                market_id,
196                ..
197            }
198            | PlatformMarketDataEvent::Heartbeat {
199                schema_version,
200                contract_version,
201                market_id,
202                ..
203            } => self.validate_identity(*schema_version, contract_version, market_id)?,
204        }
205        Ok(())
206    }
207
208    fn validate_identity(
209        &self,
210        schema_version: u16,
211        contract_version: &str,
212        market_id: &str,
213    ) -> Result<(), SdkError> {
214        validate_platform_market_response(
215            schema_version,
216            contract_version,
217            market_id,
218            &self.market_id,
219        )
220    }
221}
222
223pub(crate) fn platform_websocket_url(base_url: &Url, path: &str) -> Result<Url, SdkError> {
224    let mut url = base_url.clone();
225    let scheme = match url.scheme() {
226        "https" => "wss",
227        "http" => "ws",
228        _ => {
229            return Err(SdkError::InvalidBaseUrl(
230                "stream URL must use http or https".to_owned(),
231            ))
232        }
233    };
234    url.set_scheme(scheme)
235        .map_err(|_| SdkError::InvalidBaseUrl("could not select WebSocket scheme".to_owned()))?;
236    let base_path = url.path().trim_end_matches('/');
237    url.set_path(&format!("{base_path}/{path}"));
238    url.set_query(None);
239    url.set_fragment(None);
240    Ok(url)
241}
strata_sdk/market_stream.rs

strata_sdk/
market_stream.rs
