tradestation_api/streaming/
market_data_streams.rs

1//! Market-data streaming methods (quotes, bars, market depth, options) on
2//! `Client`, split out of the single-file `streaming` module.
3
4use super::*;
5use crate::{Client, Error};
6
7impl Client {
8    /// Stream live quote updates for one or more symbols.
9    ///
10    /// Returns an async stream of [`StreamQuote`] values including
11    /// `StreamStatus` messages (EndSnapshot, GoAway).
12    pub async fn stream_quotes(
13        &mut self,
14        symbols: &[&str],
15    ) -> Result<BoxStream<StreamQuote>, Error> {
16        let symbols_str = symbols.join(",");
17        let path = format!("/v3/marketdata/stream/quotes/{}", symbols_str);
18        let headers = self.auth_headers().await?;
19        let url = format!("{}{}", self.base_url(), &path);
20
21        let resp = self.http.get(&url).headers(headers).send().await?;
22
23        if !resp.status().is_success() {
24            let status = resp.status().as_u16();
25            let body = resp.text().await.unwrap_or_default();
26            return Err(Error::Api {
27                status,
28                message: body,
29            });
30        }
31
32        let stream = async_stream::try_stream! {
33            let mut bytes_stream = resp.bytes_stream();
34            let mut buffer = String::new();
35
36            use futures::StreamExt;
37            while let Some(chunk) = bytes_stream.next().await {
38                let chunk = chunk.map_err(Error::Http)?;
39                buffer.push_str(&String::from_utf8_lossy(&chunk));
40
41                // Process complete JSON lines
42                while let Some(newline_pos) = buffer.find('\n') {
43                    let line = buffer[..newline_pos].trim().to_string();
44                    buffer = buffer[newline_pos + 1..].to_string();
45
46                    if line.is_empty() {
47                        continue;
48                    }
49
50                    match serde_json::from_str::<StreamQuote>(&line) {
51                        Ok(quote) => yield quote,
52                        Err(e) => {
53                            tracing::warn!("Failed to parse stream quote: {e}, line: {line}");
54                        }
55                    }
56                }
57            }
58        };
59
60        Ok(Box::pin(stream))
61    }
62
63    /// Stream live bar updates for a symbol.
64    ///
65    /// # Parameters
66    ///
67    /// - `symbol`: Ticker symbol (e.g., "AAPL")
68    /// - `interval`: Bar interval (e.g., "1", "5")
69    /// - `unit`: Bar unit (e.g., "Minute", "Daily")
70    pub async fn stream_bars(
71        &mut self,
72        symbol: &str,
73        interval: &str,
74        unit: &str,
75    ) -> Result<BoxStream<StreamBar>, Error> {
76        let path = format!(
77            "/v3/marketdata/stream/barcharts/{}?interval={}&unit={}",
78            symbol, interval, unit
79        );
80        let headers = self.auth_headers().await?;
81        let url = format!("{}{}", self.base_url(), &path);
82
83        let resp = self.http.get(&url).headers(headers).send().await?;
84
85        if !resp.status().is_success() {
86            let status = resp.status().as_u16();
87            let body = resp.text().await.unwrap_or_default();
88            return Err(Error::Api {
89                status,
90                message: body,
91            });
92        }
93
94        let stream = async_stream::try_stream! {
95            let mut bytes_stream = resp.bytes_stream();
96            let mut buffer = String::new();
97
98            use futures::StreamExt;
99            while let Some(chunk) = bytes_stream.next().await {
100                let chunk = chunk.map_err(Error::Http)?;
101                buffer.push_str(&String::from_utf8_lossy(&chunk));
102
103                while let Some(newline_pos) = buffer.find('\n') {
104                    let line = buffer[..newline_pos].trim().to_string();
105                    buffer = buffer[newline_pos + 1..].to_string();
106
107                    if line.is_empty() {
108                        continue;
109                    }
110
111                    match serde_json::from_str::<StreamBar>(&line) {
112                        Ok(bar) => yield bar,
113                        Err(e) => {
114                            tracing::warn!("Failed to parse stream bar: {e}, line: {line}");
115                        }
116                    }
117                }
118            }
119        };
120
121        Ok(Box::pin(stream))
122    }
123
124    /// Stream Level 2 market depth quotes for a symbol.
125    pub async fn stream_market_depth_quotes(
126        &mut self,
127        symbol: &str,
128    ) -> Result<BoxStream<StreamMarketDepthQuote>, Error> {
129        let path = format!("/v3/marketdata/stream/marketdepth/quotes/{}", symbol);
130        let headers = self.auth_headers().await?;
131        let url = format!("{}{}", self.base_url(), &path);
132
133        let resp = self.http.get(&url).headers(headers).send().await?;
134
135        if !resp.status().is_success() {
136            let status = resp.status().as_u16();
137            let body = resp.text().await.unwrap_or_default();
138            return Err(Error::Api {
139                status,
140                message: body,
141            });
142        }
143
144        let stream = async_stream::try_stream! {
145            let mut bytes_stream = resp.bytes_stream();
146            let mut buffer = String::new();
147
148            use futures::StreamExt;
149            while let Some(chunk) = bytes_stream.next().await {
150                let chunk = chunk.map_err(Error::Http)?;
151                buffer.push_str(&String::from_utf8_lossy(&chunk));
152
153                while let Some(newline_pos) = buffer.find('\n') {
154                    let line = buffer[..newline_pos].trim().to_string();
155                    buffer = buffer[newline_pos + 1..].to_string();
156
157                    if line.is_empty() {
158                        continue;
159                    }
160
161                    match serde_json::from_str::<StreamMarketDepthQuote>(&line) {
162                        Ok(item) => yield item,
163                        Err(e) => {
164                            tracing::warn!("Failed to parse stream market depth quote: {e}, line: {line}");
165                        }
166                    }
167                }
168            }
169        };
170
171        Ok(Box::pin(stream))
172    }
173
174    /// Stream market depth aggregates for a symbol.
175    pub async fn stream_market_depth_aggregates(
176        &mut self,
177        symbol: &str,
178    ) -> Result<BoxStream<StreamMarketDepthAggregate>, Error> {
179        let path = format!("/v3/marketdata/stream/marketdepth/aggregates/{}", symbol);
180        let headers = self.auth_headers().await?;
181        let url = format!("{}{}", self.base_url(), &path);
182
183        let resp = self.http.get(&url).headers(headers).send().await?;
184
185        if !resp.status().is_success() {
186            let status = resp.status().as_u16();
187            let body = resp.text().await.unwrap_or_default();
188            return Err(Error::Api {
189                status,
190                message: body,
191            });
192        }
193
194        let stream = async_stream::try_stream! {
195            let mut bytes_stream = resp.bytes_stream();
196            let mut buffer = String::new();
197
198            use futures::StreamExt;
199            while let Some(chunk) = bytes_stream.next().await {
200                let chunk = chunk.map_err(Error::Http)?;
201                buffer.push_str(&String::from_utf8_lossy(&chunk));
202
203                while let Some(newline_pos) = buffer.find('\n') {
204                    let line = buffer[..newline_pos].trim().to_string();
205                    buffer = buffer[newline_pos + 1..].to_string();
206
207                    if line.is_empty() {
208                        continue;
209                    }
210
211                    match serde_json::from_str::<StreamMarketDepthAggregate>(&line) {
212                        Ok(item) => yield item,
213                        Err(e) => {
214                            tracing::warn!("Failed to parse stream market depth aggregate: {e}, line: {line}");
215                        }
216                    }
217                }
218            }
219        };
220
221        Ok(Box::pin(stream))
222    }
223
224    /// Stream option chain updates for an underlying symbol.
225    pub async fn stream_option_chains(
226        &mut self,
227        underlying: &str,
228    ) -> Result<BoxStream<StreamOptionChain>, Error> {
229        let path = format!("/v3/marketdata/stream/options/chains/{}", underlying);
230        let headers = self.auth_headers().await?;
231        let url = format!("{}{}", self.base_url(), &path);
232
233        let resp = self.http.get(&url).headers(headers).send().await?;
234
235        if !resp.status().is_success() {
236            let status = resp.status().as_u16();
237            let body = resp.text().await.unwrap_or_default();
238            return Err(Error::Api {
239                status,
240                message: body,
241            });
242        }
243
244        let stream = async_stream::try_stream! {
245            let mut bytes_stream = resp.bytes_stream();
246            let mut buffer = String::new();
247
248            use futures::StreamExt;
249            while let Some(chunk) = bytes_stream.next().await {
250                let chunk = chunk.map_err(Error::Http)?;
251                buffer.push_str(&String::from_utf8_lossy(&chunk));
252
253                while let Some(newline_pos) = buffer.find('\n') {
254                    let line = buffer[..newline_pos].trim().to_string();
255                    buffer = buffer[newline_pos + 1..].to_string();
256
257                    if line.is_empty() {
258                        continue;
259                    }
260
261                    match serde_json::from_str::<StreamOptionChain>(&line) {
262                        Ok(item) => yield item,
263                        Err(e) => {
264                            tracing::warn!("Failed to parse stream option chain: {e}, line: {line}");
265                        }
266                    }
267                }
268            }
269        };
270
271        Ok(Box::pin(stream))
272    }
273
274    /// Stream option quote updates for specific option legs.
275    pub async fn stream_option_quotes(
276        &mut self,
277        legs: &[&str],
278    ) -> Result<BoxStream<StreamOptionQuote>, Error> {
279        let legs_str = legs.join(",");
280        let path = format!("/v3/marketdata/stream/options/quotes/{}", legs_str);
281        let headers = self.auth_headers().await?;
282        let url = format!("{}{}", self.base_url(), &path);
283
284        let resp = self.http.get(&url).headers(headers).send().await?;
285
286        if !resp.status().is_success() {
287            let status = resp.status().as_u16();
288            let body = resp.text().await.unwrap_or_default();
289            return Err(Error::Api {
290                status,
291                message: body,
292            });
293        }
294
295        let stream = async_stream::try_stream! {
296            let mut bytes_stream = resp.bytes_stream();
297            let mut buffer = String::new();
298
299            use futures::StreamExt;
300            while let Some(chunk) = bytes_stream.next().await {
301                let chunk = chunk.map_err(Error::Http)?;
302                buffer.push_str(&String::from_utf8_lossy(&chunk));
303
304                while let Some(newline_pos) = buffer.find('\n') {
305                    let line = buffer[..newline_pos].trim().to_string();
306                    buffer = buffer[newline_pos + 1..].to_string();
307
308                    if line.is_empty() {
309                        continue;
310                    }
311
312                    match serde_json::from_str::<StreamOptionQuote>(&line) {
313                        Ok(item) => yield item,
314                        Err(e) => {
315                            tracing::warn!("Failed to parse stream option quote: {e}, line: {line}");
316                        }
317                    }
318                }
319            }
320        };
321
322        Ok(Box::pin(stream))
323    }
324}
tradestation_api/streaming/market_data_streams.rs

tradestation_api/streaming/
market_data_streams.rs
