tradestation-api 0.1.0

//! HTTP chunked-transfer streaming for TradeStation v3.
//!
//! TradeStation uses HTTP streaming (NOT WebSocket):
//! - Content-Type: `application/vnd.tradestation.streams.v3+json`
//! - Newline-delimited JSON objects
//! - `StreamStatus` messages signal snapshot boundaries (`EndSnapshot`) and
//!   reconnect requests (`GoAway`)
//!
//! All stream methods return a [`BoxStream`] of typed updates that can be
//! consumed with `futures::StreamExt`.
//!
//! # Example
//!
//! ```no_run
//! # use tradestation_api::{Client, Credentials};
//! # async fn example(client: &mut Client) -> Result<(), Box<dyn std::error::Error>> {
//! use futures::StreamExt;
//!
//! let mut stream = client.stream_quotes(&["AAPL"]).await?;
//! while let Some(result) = stream.next().await {
//!     let quote = result?;
//!     if !quote.is_status() {
//!         println!("{}: {}", quote.symbol.as_deref().unwrap_or("?"), quote.last.as_deref().unwrap_or("?"));
//!     }
//! }
//! # Ok(())
//! # }
//! ```

use futures::stream::Stream;
use serde::Deserialize;
use std::pin::Pin;

use crate::Client;
use crate::Error;

/// Stream status messages from TradeStation.
///
/// These appear inline in the stream data to signal events like end-of-snapshot
/// or a server-initiated disconnect.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamStatus {
    /// Status code: "EndSnapshot", "GoAway", etc.
    pub status: String,
    /// Optional message with additional details.
    #[serde(default)]
    pub message: Option<String>,
}

/// A streaming quote update.
///
/// Contains real-time quote data or a stream status message. Use [`StreamQuote::is_status`]
/// to distinguish between the two.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamQuote {
    /// Ticker symbol.
    pub symbol: Option<String>,
    /// Last traded price.
    pub last: Option<String>,
    /// Best ask price.
    pub ask: Option<String>,
    /// Best bid price.
    pub bid: Option<String>,
    /// Cumulative volume.
    pub volume: Option<String>,
    /// Time of the last trade.
    #[serde(default)]
    pub trade_time: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamQuote {
    /// Whether this is a status message rather than a quote update.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }

    /// Whether this is a GoAway message indicating reconnection is needed.
    pub fn is_go_away(&self) -> bool {
        self.status.as_deref() == Some("GoAway")
    }
}

/// A streaming bar (OHLCV) update.
///
/// Delivered via [`Client::stream_bars`]. Contains partial or completed bar data.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamBar {
    /// Highest price during the bar period.
    pub high: Option<String>,
    /// Lowest price during the bar period.
    pub low: Option<String>,
    /// Opening price.
    pub open: Option<String>,
    /// Closing price (updates in real time for the current bar).
    pub close: Option<String>,
    /// Bar timestamp.
    pub time_stamp: Option<String>,
    /// Total volume during the bar period.
    pub total_volume: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamBar {
    /// Whether this is a status message rather than bar data.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming market depth (Level 2) quote.
///
/// Delivered via [`Client::stream_market_depth_quotes`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamMarketDepthQuote {
    /// Ticker symbol.
    pub symbol: Option<String>,
    /// Ask price at this depth level.
    pub ask: Option<String>,
    /// Ask size at this depth level.
    pub ask_size: Option<String>,
    /// Bid price at this depth level.
    pub bid: Option<String>,
    /// Bid size at this depth level.
    pub bid_size: Option<String>,
    /// Side of the book ("Ask" or "Bid").
    #[serde(default)]
    pub side: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamMarketDepthQuote {
    /// Whether this is a status message rather than depth data.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming market depth aggregate summary.
///
/// Delivered via [`Client::stream_market_depth_aggregates`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamMarketDepthAggregate {
    /// Ticker symbol.
    pub symbol: Option<String>,
    /// Total ask size across all levels.
    pub total_ask_size: Option<String>,
    /// Total bid size across all levels.
    pub total_bid_size: Option<String>,
    /// Number of price levels.
    #[serde(default)]
    pub levels: Option<u32>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamMarketDepthAggregate {
    /// Whether this is a status message rather than aggregate data.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming option chain update.
///
/// Delivered via [`Client::stream_option_chains`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamOptionChain {
    /// Option symbol.
    pub symbol: Option<String>,
    /// Underlying ticker symbol.
    pub underlying: Option<String>,
    /// Option type ("Call" or "Put").
    #[serde(default, rename = "Type")]
    pub option_type: Option<String>,
    /// Strike price.
    pub strike_price: Option<String>,
    /// Expiration date.
    pub expiration_date: Option<String>,
    /// Best bid price.
    pub bid: Option<String>,
    /// Best ask price.
    pub ask: Option<String>,
    /// Last traded price.
    pub last: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamOptionChain {
    /// Whether this is a status message rather than chain data.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming option quote update.
///
/// Delivered via [`Client::stream_option_quotes`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamOptionQuote {
    /// Option symbol.
    pub symbol: Option<String>,
    /// Best bid price.
    pub bid: Option<String>,
    /// Best ask price.
    pub ask: Option<String>,
    /// Last traded price.
    pub last: Option<String>,
    /// Cumulative volume.
    pub volume: Option<String>,
    /// Open interest.
    #[serde(default)]
    pub open_interest: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamOptionQuote {
    /// Whether this is a status message rather than quote data.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming order status update.
///
/// Delivered via [`Client::stream_orders`] and [`Client::stream_orders_by_id`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamOrder {
    /// Order identifier.
    pub order_id: Option<String>,
    /// Account this order belongs to.
    pub account_id: Option<String>,
    /// Ticker symbol.
    pub symbol: Option<String>,
    /// Ordered quantity.
    pub quantity: Option<String>,
    /// Order type.
    pub order_type: Option<String>,
    /// Current order status.
    #[serde(default)]
    pub order_status: Option<String>,
    /// Filled quantity.
    #[serde(default)]
    pub filled_quantity: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamOrder {
    /// Whether this is a status message rather than an order update.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// A streaming position update.
///
/// Delivered via [`Client::stream_positions`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub struct StreamPosition {
    /// Account holding this position.
    pub account_id: Option<String>,
    /// Ticker symbol.
    pub symbol: Option<String>,
    /// Current position quantity.
    pub quantity: Option<String>,
    /// Average entry price.
    pub average_price: Option<String>,
    /// Last traded price.
    pub last: Option<String>,
    /// Unrealized P&L.
    #[serde(default)]
    pub unrealized_profit_loss: Option<String>,
    /// Stream status (present only for status messages).
    #[serde(default)]
    pub status: Option<String>,
}

impl StreamPosition {
    /// Whether this is a status message rather than a position update.
    pub fn is_status(&self) -> bool {
        self.status.is_some()
    }
}

/// Type alias for a boxed async stream of results.
///
/// All streaming methods return this type. Consume it with
/// `futures::StreamExt::next()`.
pub type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Error>> + Send>>;

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))
    }

    /// Stream order status updates for the specified accounts.
    pub async fn stream_orders(
        &mut self,
        account_ids: &[&str],
    ) -> Result<BoxStream<StreamOrder>, Error> {
        let ids = account_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/orders", ids);
        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::<StreamOrder>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream order updates for specific order IDs.
    pub async fn stream_orders_by_id(
        &mut self,
        account_ids: &[&str],
        order_ids: &[&str],
    ) -> Result<BoxStream<StreamOrder>, Error> {
        let ids = account_ids.join(",");
        let oids = order_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/orders/{}", ids, oids);
        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::<StreamOrder>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream order: {e}, line: {line}");
                        }
                    }
                }
            }
        };

        Ok(Box::pin(stream))
    }

    /// Stream position updates for the specified accounts.
    pub async fn stream_positions(
        &mut self,
        account_ids: &[&str],
    ) -> Result<BoxStream<StreamPosition>, Error> {
        let ids = account_ids.join(",");
        let path = format!("/v3/brokerage/stream/accounts/{}/positions", ids);
        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::<StreamPosition>(&line) {
                        Ok(item) => yield item,
                        Err(e) => {
                            tracing::warn!("Failed to parse stream position: {e}, line: {line}");
                        }
                    }
                }
            }
        };

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