finance_query/streaming/
trades.rs

1//! Tick-by-tick trade streaming.
2//!
3//! [`PriceStream`](super::PriceStream) coalesces activity into a last-price
4//! tick; this stream pushes every individual print, which is what
5//! execution-quality and microstructure work needs.
6
7use std::sync::Arc;
8use std::time::Duration;
9
10use serde::{Deserialize, Serialize};
11
12use super::client::StreamResult;
13use super::handle::{RECONNECT_BACKOFF, SourceStream, stream_builder, stream_handle};
14use super::polygon::{AssetClass, PolygonTradeSource};
15use super::source::ReconnectConfig;
16
17/// Channel capacity — trade prints are the highest-volume feed here.
18const CHANNEL_CAPACITY: usize = 4096;
19
20/// A single executed trade.
21#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
22#[serde(rename_all = "camelCase")]
23#[non_exhaustive]
24pub struct TradeTick {
25    /// Symbol or pair the trade executed on.
26    pub symbol: String,
27    /// Execution price.
28    pub price: f64,
29    /// Executed size (shares, contracts, or base-asset units).
30    pub size: f64,
31    /// Exchange identifier, where the venue reports one.
32    pub exchange: Option<i32>,
33    /// Trade condition codes.
34    pub conditions: Vec<i32>,
35    /// Provider trade identifier, where one is supplied.
36    pub trade_id: Option<String>,
37    /// Execution timestamp (milliseconds).
38    pub time: i64,
39}
40
41impl TradeTick {
42    /// Notional value of the print (`price * size`).
43    pub fn notional(&self) -> f64 {
44        self.price * self.size
45    }
46}
47
48stream_handle! {
49    /// A subscription to every trade print for the given symbols.
50    ///
51    /// Requires the `polygon` feature and the `POLYGON_API_KEY` environment
52    /// variable set.
53    /// This is a companion to [`PriceStream`](super::PriceStream), not a
54    /// replacement — most consumers want the coalesced tick.
55    ///
56    /// # Example
57    ///
58    /// ```no_run
59    /// use finance_query::streaming::TradeStream;
60    /// use futures::StreamExt;
61    ///
62    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
63    /// let mut trades = TradeStream::subscribe(["AAPL"]).await?;
64    ///
65    /// while let Some(trade) = trades.next().await {
66    ///     println!("{} {} @ {}", trade.symbol, trade.size, trade.price);
67    /// }
68    /// # Ok(())
69    /// # }
70    /// ```
71    TradeStream(TradeTick);
72    add: add_symbols = "Add symbols to the subscription.",
73    remove: remove_symbols = "Remove symbols from the subscription.",
74}
75
76impl TradeStream {
77    /// Subscribe to US equity trade prints for the given symbols.
78    pub async fn subscribe<S, I>(symbols: I) -> StreamResult<Self>
79    where
80        S: Into<String>,
81        I: IntoIterator<Item = S>,
82    {
83        TradeStreamBuilder::new().symbols(symbols).build().await
84    }
85}
86
87/// Builder for a [`TradeStream`].
88pub struct TradeStreamBuilder {
89    symbols: Vec<String>,
90    asset_class: AssetClass,
91    retry_delay: Duration,
92    max_reconnect_attempts: Option<u32>,
93}
94
95impl TradeStreamBuilder {
96    /// Create a builder defaulting to US equities.
97    pub fn new() -> Self {
98        Self {
99            symbols: Vec::new(),
100            asset_class: AssetClass::Stocks,
101            retry_delay: RECONNECT_BACKOFF,
102            max_reconnect_attempts: None,
103        }
104    }
105
106    /// Choose the asset class (default: [`AssetClass::Stocks`]).
107    ///
108    /// Forex and indices publish no per-trade prints; those classes are
109    /// rejected at [`build`](Self::build).
110    pub fn asset_class(mut self, class: AssetClass) -> Self {
111        self.asset_class = class;
112        self
113    }
114
115    /// Build and start the stream.
116    ///
117    /// # Errors
118    ///
119    /// Returns [`StreamError::ConnectionFailed`](super::StreamError::ConnectionFailed)
120    /// when the chosen asset class has no trade feed.
121    pub async fn build(self) -> StreamResult<TradeStream> {
122        let source = Arc::new(PolygonTradeSource::new(self.asset_class)?);
123        let reconnect =
124            ReconnectConfig::new(self.retry_delay).max_attempts(self.max_reconnect_attempts);
125        Ok(TradeStream {
126            inner: SourceStream::start(source, self.symbols, reconnect, CHANNEL_CAPACITY),
127        })
128    }
129}
130
131stream_builder!(TradeStreamBuilder, symbols = "Add symbols to subscribe to.");
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136
137    #[tokio::test]
138    async fn classes_without_trade_prints_are_rejected() {
139        for class in [AssetClass::Forex, AssetClass::Indices] {
140            assert!(
141                TradeStreamBuilder::new()
142                    .symbols(["X"])
143                    .asset_class(class)
144                    .build()
145                    .await
146                    .is_err()
147            );
148        }
149    }
150
151    #[test]
152    fn notional_multiplies_price_by_size() {
153        let tick = TradeTick {
154            price: 10.0,
155            size: 25.0,
156            ..Default::default()
157        };
158        assert!((tick.notional() - 250.0).abs() < 1e-9);
159    }
160}
finance_query/streaming/trades.rs

finance_query/streaming/
trades.rs
