rustrade_data/exchange/alpaca/
trade.rs

1use super::channel::AlpacaChannel;
2use crate::{
3    Identifier,
4    error::DataError,
5    event::MarketEvent,
6    exchange::ExchangeSub,
7    subscription::{Map, trade::PublicTrade},
8    transformer::ExchangeTransformer,
9};
10use chrono::{DateTime, Utc};
11use rust_decimal::Decimal;
12use rustrade_instrument::{Side, exchange::ExchangeId};
13use rustrade_integration::{
14    Transformer, protocol::websocket::WsMessage, subscription::SubscriptionId,
15};
16use serde::{Deserialize, Deserializer};
17use smol_str::{SmolStr, format_smolstr};
18use std::marker::PhantomData;
19use tokio::sync::mpsc;
20
21/// Deserialize "S" (symbol) field as the associated [`SubscriptionId`] for trades.
22fn de_trade_subscription_id<'de, D>(deserializer: D) -> Result<SubscriptionId, D::Error>
23where
24    D: serde::de::Deserializer<'de>,
25{
26    <&str as Deserialize>::deserialize(deserializer)
27        .map(|symbol| ExchangeSub::from((AlpacaChannel::Trades, symbol)).id())
28}
29
30/// Alpaca trade message.
31///
32/// Unified struct for both equities and crypto trades with optional fields.
33///
34/// ### Equities Example (IEX/SIP)
35/// ```json
36/// {"T":"t","S":"AAPL","i":123,"x":"V","p":150.25,"s":100,"c":["@"],"z":"C","t":"2026-05-02T14:00:00Z"}
37/// ```
38///
39/// ### Crypto Example
40/// ```json
41/// {"T":"t","S":"BTC/USD","i":456,"p":60000.50,"s":0.5,"tks":"B","t":"2026-05-02T14:00:00Z"}
42/// ```
43#[derive(Debug, Deserialize)]
44pub struct AlpacaTrade {
45    /// Subscription ID constructed from symbol during deserialization.
46    /// Avoids per-event `format!` allocation in hot path.
47    #[serde(rename = "S", deserialize_with = "de_trade_subscription_id")]
48    pub subscription_id: SubscriptionId,
49    #[serde(rename = "i")]
50    pub id: u64,
51    #[serde(rename = "p")]
52    pub price: Decimal,
53    #[serde(rename = "s")]
54    pub size: Decimal,
55    #[serde(rename = "t")]
56    pub timestamp: DateTime<Utc>,
57    #[serde(rename = "x", default)]
58    pub exchange: Option<SmolStr>,
59    #[serde(rename = "z", default)]
60    pub tape: Option<SmolStr>,
61    #[serde(rename = "tks", default)]
62    pub taker_side: Option<SmolStr>,
63}
64
65impl AlpacaTrade {
66    /// Returns the taker side for crypto trades, or `None` for equities.
67    ///
68    /// Alpaca equities (IEX/SIP) do not provide taker side information — the `tks` field
69    /// is only present on crypto trades.
70    fn side(&self) -> Option<Side> {
71        match self.taker_side.as_deref() {
72            Some("B") => Some(Side::Buy),
73            Some("S") => Some(Side::Sell),
74            None => None,
75            Some(other) => {
76                tracing::warn!(tks = other, "AlpacaTrade: unexpected taker_side value");
77                None
78            }
79        }
80    }
81}
82
83/// Alpaca WebSocket message wrapper.
84///
85/// Alpaca sends all messages as JSON arrays: `[{"T":"t",...},{"T":"t",...}]`.
86/// This wrapper deserializes the array and extracts trade messages.
87#[derive(Debug)]
88pub struct AlpacaTradeMessage(Vec<AlpacaTrade>);
89
90/// Internal enum for single-pass deserialization of Alpaca array messages.
91/// Uses `#[serde(tag = "T")]` to dispatch on message type in one parse pass,
92/// avoiding the intermediate `Vec<serde_json::Value>` allocation.
93#[derive(Deserialize)]
94#[serde(tag = "T")]
95enum AlpacaArrayTradeMsg {
96    #[serde(rename = "t")]
97    Trade(AlpacaTrade),
98    #[serde(other)]
99    Other,
100}
101
102impl<'de> Deserialize<'de> for AlpacaTradeMessage {
103    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
104    where
105        D: Deserializer<'de>,
106    {
107        let messages = Vec::<AlpacaArrayTradeMsg>::deserialize(deserializer)?;
108        let mut trades = Vec::with_capacity(messages.len());
109        for msg in messages {
110            if let AlpacaArrayTradeMsg::Trade(trade) = msg {
111                trades.push(trade);
112            }
113        }
114        Ok(AlpacaTradeMessage(trades))
115    }
116}
117
118/// Custom transformer for Alpaca trade messages.
119///
120/// Handles array-wrapped messages and processes each trade individually,
121/// looking up the correct instrument for each symbol.
122#[derive(Debug)]
123pub struct AlpacaTradeTransformer<Exchange, InstrumentKey> {
124    instrument_map: Map<InstrumentKey>,
125    exchange_id: ExchangeId,
126    phantom: PhantomData<Exchange>,
127}
128
129impl<Exchange, InstrumentKey>
130    ExchangeTransformer<Exchange, InstrumentKey, crate::subscription::trade::PublicTrades>
131    for AlpacaTradeTransformer<Exchange, InstrumentKey>
132where
133    Exchange: crate::exchange::Connector + Send,
134    InstrumentKey: Clone + Send + Sync,
135{
136    async fn init(
137        instrument_map: Map<InstrumentKey>,
138        _: &[MarketEvent<InstrumentKey, PublicTrade>],
139        _: mpsc::UnboundedSender<WsMessage>,
140    ) -> Result<Self, DataError> {
141        Ok(Self {
142            instrument_map,
143            exchange_id: Exchange::ID,
144            phantom: PhantomData,
145        })
146    }
147}
148
149impl<Exchange, InstrumentKey> Transformer for AlpacaTradeTransformer<Exchange, InstrumentKey>
150where
151    Exchange: crate::exchange::Connector,
152    InstrumentKey: Clone,
153{
154    type Error = DataError;
155    type Input = AlpacaTradeMessage;
156    type Output = MarketEvent<InstrumentKey, PublicTrade>;
157    type OutputIter = Vec<Result<Self::Output, Self::Error>>;
158
159    fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
160        let mut results = Vec::with_capacity(input.0.len());
161        let time_received = Utc::now();
162
163        for trade in input.0 {
164            match self.instrument_map.find(&trade.subscription_id) {
165                Ok(instrument) => {
166                    results.push(Ok(MarketEvent {
167                        time_exchange: trade.timestamp,
168                        time_received,
169                        exchange: self.exchange_id,
170                        instrument: instrument.clone(),
171                        kind: PublicTrade {
172                            id: format_smolstr!("{}", trade.id),
173                            price: trade.price,
174                            amount: trade.size,
175                            side: trade.side(),
176                        },
177                    }));
178                }
179                Err(unidentified) => {
180                    results.push(Err(DataError::Socket(unidentified.to_string())));
181                }
182            }
183        }
184
185        results
186    }
187}
188
189#[cfg(test)]
190#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
191mod tests {
192    use super::*;
193    use rust_decimal_macros::dec;
194
195    #[test]
196    fn test_de_equities_trade() {
197        let input = r#"{"T":"t","S":"AAPL","i":123,"x":"V","p":150.25,"s":100,"c":["@"],"z":"C","t":"2026-05-02T14:00:00Z"}"#;
198        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
199
200        assert_eq!(trade.subscription_id.as_ref(), "trades|AAPL");
201        assert_eq!(trade.id, 123);
202        assert_eq!(trade.price, dec!(150.25));
203        assert_eq!(trade.size, dec!(100));
204        assert_eq!(trade.exchange, Some(SmolStr::new("V")));
205        assert_eq!(trade.tape, Some(SmolStr::new("C")));
206        assert!(trade.taker_side.is_none());
207    }
208
209    #[test]
210    fn test_de_crypto_trade() {
211        let input = r#"{"T":"t","S":"BTC/USD","i":456,"p":60000.50,"s":0.5,"tks":"B","t":"2026-05-02T14:00:00Z"}"#;
212        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
213
214        assert_eq!(trade.subscription_id.as_ref(), "trades|BTC/USD");
215        assert_eq!(trade.id, 456);
216        assert_eq!(trade.price, dec!(60000.50));
217        assert_eq!(trade.size, dec!(0.5));
218        assert!(trade.exchange.is_none());
219        assert!(trade.tape.is_none());
220        assert_eq!(trade.taker_side, Some(SmolStr::new("B")));
221        assert_eq!(trade.side(), Some(Side::Buy));
222    }
223
224    #[test]
225    fn test_crypto_side_sell() {
226        let input = r#"{"T":"t","S":"ETH/USD","i":789,"p":3000.0,"s":1.0,"tks":"S","t":"2026-05-02T14:00:00Z"}"#;
227        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
228        assert_eq!(trade.side(), Some(Side::Sell));
229    }
230
231    #[test]
232    fn test_equities_side_none() {
233        let input = r#"{"T":"t","S":"AAPL","i":123,"x":"V","p":150.25,"s":100,"c":["@"],"z":"C","t":"2026-05-02T14:00:00Z"}"#;
234        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
235        assert!(trade.side().is_none());
236    }
237
238    #[test]
239    fn test_subscription_id() {
240        let input = r#"{"T":"t","S":"AAPL","i":123,"p":150.25,"s":100,"t":"2026-05-02T14:00:00Z"}"#;
241        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
242        assert_eq!(trade.subscription_id.as_ref(), "trades|AAPL");
243    }
244
245    #[test]
246    fn test_unexpected_taker_side_returns_none() {
247        let input = r#"{"T":"t","S":"BTC/USD","i":456,"p":60000.50,"s":0.5,"tks":"X","t":"2026-05-02T14:00:00Z"}"#;
248        let trade: AlpacaTrade = serde_json::from_str(input).unwrap();
249        assert_eq!(trade.taker_side, Some(SmolStr::new("X")));
250        assert!(trade.side().is_none());
251    }
252}
rustrade_data/exchange/alpaca/trade.rs

rustrade_data/exchange/alpaca/
trade.rs
