rustrade_data/exchange/alpaca/
quote.rs

1use super::channel::AlpacaChannel;
2use crate::{
3    Identifier,
4    books::Level,
5    error::DataError,
6    event::MarketEvent,
7    exchange::ExchangeSub,
8    subscription::{
9        Map, SubscriptionKind,
10        book::OrderBookL1,
11        quote::{Quote, Quotes},
12    },
13    transformer::ExchangeTransformer,
14};
15use chrono::{DateTime, Utc};
16use rust_decimal::Decimal;
17use rustrade_instrument::exchange::ExchangeId;
18use rustrade_integration::{
19    Transformer, protocol::websocket::WsMessage, subscription::SubscriptionId,
20};
21use serde::{Deserialize, Deserializer};
22use smol_str::SmolStr;
23use std::marker::PhantomData;
24use tokio::sync::mpsc;
25
26/// Deserialize "S" (symbol) field as the associated [`SubscriptionId`] for quotes.
27fn de_quote_subscription_id<'de, D>(deserializer: D) -> Result<SubscriptionId, D::Error>
28where
29    D: serde::de::Deserializer<'de>,
30{
31    <&str as Deserialize>::deserialize(deserializer)
32        .map(|symbol| ExchangeSub::from((AlpacaChannel::Quotes, symbol)).id())
33}
34
35/// Alpaca quote message.
36///
37/// Unified struct for both equities and crypto quotes with optional fields.
38///
39/// ### Equities Example (IEX/SIP)
40/// ```json
41/// {"T":"q","S":"AAPL","ax":"V","ap":150.25,"as":100,"bx":"Q","bp":150.20,"bs":200,"c":["R"],"z":"C","t":"2026-05-02T14:00:00Z"}
42/// ```
43///
44/// ### Crypto Example
45/// ```json
46/// {"T":"q","S":"BTC/USD","ap":60000.50,"as":1.0,"bp":60000.00,"bs":2.0,"t":"2026-05-02T14:00:00Z"}
47/// ```
48#[derive(Debug, Deserialize)]
49pub struct AlpacaQuote {
50    /// Subscription ID constructed from symbol during deserialization.
51    /// Avoids per-event `format!` allocation in hot path.
52    #[serde(rename = "S", deserialize_with = "de_quote_subscription_id")]
53    pub subscription_id: SubscriptionId,
54    #[serde(rename = "bp")]
55    pub bid_price: Decimal,
56    #[serde(rename = "bs")]
57    pub bid_size: Decimal,
58    #[serde(rename = "ap")]
59    pub ask_price: Decimal,
60    #[serde(rename = "as")]
61    pub ask_size: Decimal,
62    #[serde(rename = "t")]
63    pub timestamp: DateTime<Utc>,
64    #[serde(rename = "bx", default)]
65    pub bid_exchange: Option<SmolStr>,
66    #[serde(rename = "ax", default)]
67    pub ask_exchange: Option<SmolStr>,
68    #[serde(rename = "z", default)]
69    pub tape: Option<SmolStr>,
70}
71
72/// Alpaca WebSocket message wrapper for quotes.
73///
74/// Alpaca sends all messages as JSON arrays: `[{"T":"q",...},{"T":"q",...}]`.
75/// This wrapper deserializes the array and extracts quote messages.
76#[derive(Debug)]
77pub struct AlpacaQuoteMessage(Vec<AlpacaQuote>);
78
79/// Internal enum for single-pass deserialization of Alpaca array messages.
80/// Uses `#[serde(tag = "T")]` to dispatch on message type in one parse pass,
81/// avoiding the intermediate `Vec<serde_json::Value>` allocation.
82#[derive(Deserialize)]
83#[serde(tag = "T")]
84enum AlpacaArrayQuoteMsg {
85    #[serde(rename = "q")]
86    Quote(AlpacaQuote),
87    #[serde(other)]
88    Other,
89}
90
91impl<'de> Deserialize<'de> for AlpacaQuoteMessage {
92    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
93    where
94        D: Deserializer<'de>,
95    {
96        let messages = Vec::<AlpacaArrayQuoteMsg>::deserialize(deserializer)?;
97        let mut quotes = Vec::with_capacity(messages.len());
98        for msg in messages {
99            if let AlpacaArrayQuoteMsg::Quote(quote) = msg {
100                quotes.push(quote);
101            }
102        }
103        Ok(AlpacaQuoteMessage(quotes))
104    }
105}
106
107impl From<AlpacaQuote> for Quote {
108    fn from(quote: AlpacaQuote) -> Self {
109        Self {
110            bid_price: quote.bid_price,
111            bid_amount: quote.bid_size,
112            ask_price: quote.ask_price,
113            ask_amount: quote.ask_size,
114        }
115    }
116}
117
118/// The quote as a top-of-book, stamped with the quote's own time.
119///
120/// The book is the feed's, not the market's: IEX quotes are IEX's own top of book, SIP quotes are
121/// the NBBO, and crypto quotes are Alpaca's crypto venue.
122///
123/// # A zero price is absent
124/// Every Alpaca quote carries a price for both sides, so a side with nothing quoted on it can only
125/// arrive as a zero price. A zero-priced level would assert someone will trade at zero,
126/// and a mid-price averaged against it is wrong in a way nothing downstream can detect. So a zero
127/// price maps to `None`, the rule the Binance and London Strategic Edge L1 decoders follow. A zero
128/// *size* at a real price is kept: the price is still a quote.
129impl From<AlpacaQuote> for OrderBookL1 {
130    fn from(quote: AlpacaQuote) -> Self {
131        Self {
132            // Must equal the event's `time_exchange`, which the transformer takes from the same
133            // field: downstream state orders L1 updates on this one.
134            last_update_time: quote.timestamp,
135            best_bid: quoted(quote.bid_price, quote.bid_size),
136            best_ask: quoted(quote.ask_price, quote.ask_size),
137        }
138    }
139}
140
141/// One side of the book, or `None` where Alpaca published no quote for it.
142fn quoted(price: Decimal, size: Decimal) -> Option<Level> {
143    (!price.is_zero()).then(|| Level::new(price, size))
144}
145
146/// Transformer for Alpaca quote messages, producing `Kind`'s event from each quote.
147///
148/// Handles array-wrapped messages and processes each quote individually, looking up the correct
149/// instrument for each symbol. Serves [`Quotes`] and
150/// [`OrderBooksL1`](crate::subscription::book::OrderBooksL1), both read from Alpaca's `quotes`
151/// channel.
152#[derive(Debug)]
153pub struct AlpacaQuoteTransformer<Exchange, InstrumentKey, Kind = Quotes> {
154    instrument_map: Map<InstrumentKey>,
155    exchange_id: ExchangeId,
156    phantom: PhantomData<(Exchange, Kind)>,
157}
158
159impl<Exchange, InstrumentKey, Kind> ExchangeTransformer<Exchange, InstrumentKey, Kind>
160    for AlpacaQuoteTransformer<Exchange, InstrumentKey, Kind>
161where
162    Exchange: crate::exchange::Connector + Send,
163    InstrumentKey: Clone + Send + Sync,
164    Kind: SubscriptionKind + Send,
165    Kind::Event: From<AlpacaQuote>,
166{
167    async fn init(
168        instrument_map: Map<InstrumentKey>,
169        _: &[MarketEvent<InstrumentKey, Kind::Event>],
170        _: mpsc::UnboundedSender<WsMessage>,
171    ) -> Result<Self, DataError> {
172        Ok(Self {
173            instrument_map,
174            exchange_id: Exchange::ID,
175            phantom: PhantomData,
176        })
177    }
178}
179
180impl<Exchange, InstrumentKey, Kind> Transformer
181    for AlpacaQuoteTransformer<Exchange, InstrumentKey, Kind>
182where
183    Exchange: crate::exchange::Connector,
184    InstrumentKey: Clone,
185    Kind: SubscriptionKind,
186    Kind::Event: From<AlpacaQuote>,
187{
188    type Error = DataError;
189    type Input = AlpacaQuoteMessage;
190    type Output = MarketEvent<InstrumentKey, Kind::Event>;
191    type OutputIter = Vec<Result<Self::Output, Self::Error>>;
192
193    fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
194        let mut results = Vec::with_capacity(input.0.len());
195        let time_received = Utc::now();
196
197        for quote in input.0 {
198            match self.instrument_map.find(&quote.subscription_id) {
199                Ok(instrument) => {
200                    results.push(Ok(MarketEvent {
201                        time_exchange: quote.timestamp,
202                        time_received,
203                        exchange: self.exchange_id,
204                        instrument: instrument.clone(),
205                        kind: Kind::Event::from(quote),
206                    }));
207                }
208                Err(unidentified) => {
209                    results.push(Err(DataError::Socket(unidentified.to_string())));
210                }
211            }
212        }
213
214        results
215    }
216}
217
218#[cfg(test)]
219#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
220mod tests {
221    use super::*;
222    use rust_decimal_macros::dec;
223
224    #[test]
225    fn test_de_equities_quote() {
226        let input = r#"{"T":"q","S":"AAPL","ax":"V","ap":150.25,"as":100,"bx":"Q","bp":150.20,"bs":200,"c":["R"],"z":"C","t":"2026-05-02T14:00:00Z"}"#;
227        let quote: AlpacaQuote = serde_json::from_str(input).unwrap();
228
229        assert_eq!(quote.subscription_id.as_ref(), "quotes|AAPL");
230        assert_eq!(quote.bid_price, dec!(150.20));
231        assert_eq!(quote.bid_size, dec!(200));
232        assert_eq!(quote.ask_price, dec!(150.25));
233        assert_eq!(quote.ask_size, dec!(100));
234        assert_eq!(quote.bid_exchange, Some(SmolStr::new("Q")));
235        assert_eq!(quote.ask_exchange, Some(SmolStr::new("V")));
236        assert_eq!(quote.tape, Some(SmolStr::new("C")));
237    }
238
239    #[test]
240    fn test_de_crypto_quote() {
241        let input = r#"{"T":"q","S":"BTC/USD","ap":60000.50,"as":1.0,"bp":60000.00,"bs":2.0,"t":"2026-05-02T14:00:00Z"}"#;
242        let quote: AlpacaQuote = serde_json::from_str(input).unwrap();
243
244        assert_eq!(quote.subscription_id.as_ref(), "quotes|BTC/USD");
245        assert_eq!(quote.bid_price, dec!(60000.00));
246        assert_eq!(quote.bid_size, dec!(2.0));
247        assert_eq!(quote.ask_price, dec!(60000.50));
248        assert_eq!(quote.ask_size, dec!(1.0));
249        assert!(quote.bid_exchange.is_none());
250        assert!(quote.ask_exchange.is_none());
251        assert!(quote.tape.is_none());
252    }
253
254    #[test]
255    fn test_subscription_id() {
256        let input = r#"{"T":"q","S":"SPY","bp":450.0,"bs":100,"ap":450.05,"as":50,"t":"2026-05-02T14:00:00Z"}"#;
257        let quote: AlpacaQuote = serde_json::from_str(input).unwrap();
258        assert_eq!(quote.subscription_id.as_ref(), "quotes|SPY");
259    }
260
261    #[test]
262    fn a_quote_becomes_a_book_stamped_with_its_own_time() {
263        let input = r#"{"T":"q","S":"BTC/USD","ap":60000.50,"as":1.0,"bp":60000.00,"bs":2.0,"t":"2026-05-02T14:00:00Z"}"#;
264        let quote: AlpacaQuote = serde_json::from_str(input).unwrap();
265        let time = quote.timestamp;
266
267        let book = OrderBookL1::from(quote);
268
269        assert_eq!(book.last_update_time, time);
270        assert_eq!(book.best_bid, Some(Level::new(dec!(60000.00), dec!(2.0))));
271        assert_eq!(book.best_ask, Some(Level::new(dec!(60000.50), dec!(1.0))));
272    }
273
274    #[test]
275    fn a_zero_priced_side_is_absent_but_a_zero_sized_one_is_kept() {
276        let input =
277            r#"{"T":"q","S":"AAPL","bp":0,"bs":0,"ap":150.25,"as":0,"t":"2026-05-02T14:00:00Z"}"#;
278        let quote: AlpacaQuote = serde_json::from_str(input).unwrap();
279
280        let book = OrderBookL1::from(quote);
281
282        assert_eq!(book.best_bid, None);
283        assert_eq!(book.best_ask, Some(Level::new(dec!(150.25), dec!(0))));
284    }
285
286    #[test]
287    fn each_quote_in_a_frame_becomes_one_book_under_its_own_instrument() {
288        use crate::{exchange::alpaca::AlpacaCrypto, subscription::book::OrderBooksL1};
289        use rustrade_integration::subscription::SubscriptionId;
290
291        let map = Map([
292            (SubscriptionId::from("quotes|BTC/USD"), 1_u8),
293            (SubscriptionId::from("quotes|ETH/USD"), 2_u8),
294        ]
295        .into_iter()
296        .collect());
297        let mut transformer = AlpacaQuoteTransformer::<AlpacaCrypto, u8, OrderBooksL1> {
298            instrument_map: map,
299            exchange_id: ExchangeId::AlpacaCrypto,
300            phantom: PhantomData,
301        };
302        let frame: AlpacaQuoteMessage = serde_json::from_str(
303            r#"[{"T":"q","S":"BTC/USD","ap":2,"as":1,"bp":1,"bs":1,"t":"2026-05-02T14:00:00Z"},
304                {"T":"q","S":"ETH/USD","ap":4,"as":1,"bp":3,"bs":1,"t":"2026-05-02T14:00:01Z"}]"#,
305        )
306        .unwrap();
307
308        let events = transformer
309            .transform(frame)
310            .into_iter()
311            .map(Result::unwrap)
312            .collect::<Vec<_>>();
313
314        assert_eq!(events.len(), 2);
315        for (event, (instrument, bid)) in events.iter().zip([(1, dec!(1)), (2, dec!(3))]) {
316            assert_eq!(event.exchange, ExchangeId::AlpacaCrypto);
317            assert_eq!(event.instrument, instrument);
318            assert_eq!(event.kind.last_update_time, event.time_exchange);
319            assert_eq!(event.kind.best_bid.unwrap().price, bid);
320        }
321    }
322}
rustrade_data/exchange/alpaca/quote.rs

rustrade_data/exchange/alpaca/
quote.rs
