tradestation-api 0.1.1

//! Market-data streaming methods (quotes, bars, market depth, options) on
//! `Client`, split out of the single-file `streaming` module.

use super::*;
use crate::{Client, Error};

impl Client {
    /// Stream live quote updates for one or more symbols.
    ///
    /// Returns an async stream of [`StreamQuote`] values including
    /// `StreamStatus` messages (EndSnapshot, GoAway).
    pub async fn stream_quotes(
        &mut self,
        symbols: &[&str],
    ) -> Result<BoxStream<StreamQuote>, Error> {
        let symbols_str = symbols.join(",");
        let path = format!("/v3/marketdata/stream/quotes/{}", symbols_str);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                // Process complete JSON lines
                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamQuote>(&line) {
                        Ok(quote) => yield quote,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream quote: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream live bar updates for a symbol.
    ///
    /// # Parameters
    ///
    /// - `symbol`: Ticker symbol (e.g., "AAPL")
    /// - `interval`: Bar interval (e.g., "1", "5")
    /// - `unit`: Bar unit (e.g., "Minute", "Daily")
    pub async fn stream_bars(
        &mut self,
        symbol: &str,
        interval: &str,
        unit: &str,
    ) -> Result<BoxStream<StreamBar>, Error> {
        let path = format!(
            "/v3/marketdata/stream/barcharts/{}?interval={}&unit={}",
            symbol, interval, unit
        );
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamBar>(&line) {
                        Ok(bar) => yield bar,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream bar: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream Level 2 market depth quotes for a symbol.
    pub async fn stream_market_depth_quotes(
        &mut self,
        symbol: &str,
    ) -> Result<BoxStream<StreamMarketDepthQuote>, Error> {
        let path = format!("/v3/marketdata/stream/marketdepth/quotes/{}", symbol);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamMarketDepthQuote>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream market depth quote: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream market depth aggregates for a symbol.
    pub async fn stream_market_depth_aggregates(
        &mut self,
        symbol: &str,
    ) -> Result<BoxStream<StreamMarketDepthAggregate>, Error> {
        let path = format!("/v3/marketdata/stream/marketdepth/aggregates/{}", symbol);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamMarketDepthAggregate>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream market depth aggregate: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream option chain updates for an underlying symbol.
    pub async fn stream_option_chains(
        &mut self,
        underlying: &str,
    ) -> Result<BoxStream<StreamOptionChain>, Error> {
        let path = format!("/v3/marketdata/stream/options/chains/{}", underlying);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamOptionChain>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream option chain: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream option quote updates for specific option legs.
    pub async fn stream_option_quotes(
        &mut self,
        legs: &[&str],
    ) -> Result<BoxStream<StreamOptionQuote>, Error> {
        let legs_str = legs.join(",");
        let path = format!("/v3/marketdata/stream/options/quotes/{}", legs_str);
        let headers = self.auth_headers().await?;
        let url = format!("{}{}", self.base_url(), &path);

        let resp = self.http.get(&url).headers(headers).send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status,
                message: body,
            });
        }

        let stream = async_stream::try_stream! {
            let mut bytes_stream = resp.bytes_stream();
            let mut buffer = String::new();

            use futures::StreamExt;
            while let Some(chunk) = bytes_stream.next().await {
                let chunk = chunk.map_err(Error::Http)?;
                buffer.push_str(&String::from_utf8_lossy(&chunk));

                while let Some(newline_pos) = buffer.find('\n') {
                    let line = buffer[..newline_pos].trim().to_string();
                    buffer = buffer[newline_pos + 1..].to_string();

                    if line.is_empty() {
                        continue;
                    }

                    match serde_json::from_str::<StreamOptionQuote>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream option quote: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }
}