rustrade_data/exchange/alpaca/
subscription.rs

1use super::channel::AlpacaChannel;
2use rustrade_integration::{Validator, error::SocketError};
3use serde::{Deserialize, Deserializer, Serialize};
4use smol_str::SmolStr;
5
6/// Alpaca WebSocket subscription response wrapper.
7///
8/// Alpaca sends all messages as JSON arrays: `[{"T":"subscription",...}]`.
9/// This wrapper deserializes the array and validates each element.
10#[derive(Clone, PartialEq, Debug, Serialize)]
11pub struct AlpacaSubResponse(pub Vec<AlpacaSubResponseInner>);
12
13impl<'de> Deserialize<'de> for AlpacaSubResponse {
14    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
15    where
16        D: Deserializer<'de>,
17    {
18        Vec::<AlpacaSubResponseInner>::deserialize(deserializer).map(AlpacaSubResponse)
19    }
20}
21
22impl Validator for AlpacaSubResponse {
23    type Error = SocketError;
24
25    fn validate(self) -> Result<Self, SocketError> {
26        for inner in &self.0 {
27            if let AlpacaSubResponseInner::Error { code, msg } = inner {
28                return Err(SocketError::Subscribe(format!(
29                    "Alpaca subscription error (code {:?}): {}",
30                    code, msg
31                )));
32            }
33        }
34        Ok(self)
35    }
36}
37
38/// Individual Alpaca WebSocket message.
39///
40/// ### Raw Payload Examples
41///
42/// #### Subscription Success
43/// ```json
44/// [{"T":"subscription","trades":["AAPL","SPY"],"quotes":["AAPL"]}]
45/// ```
46///
47/// #### Error
48/// ```json
49/// [{"T":"error","code":400,"msg":"invalid syntax"}]
50/// ```
51#[derive(Clone, PartialEq, Debug, Deserialize, Serialize)]
52#[serde(tag = "T", rename_all = "lowercase")]
53pub enum AlpacaSubResponseInner {
54    /// Successful subscription confirmation.
55    Subscription {
56        #[serde(default)]
57        trades: Vec<SmolStr>,
58        #[serde(default)]
59        quotes: Vec<SmolStr>,
60        #[serde(default)]
61        bars: Vec<SmolStr>,
62    },
63    /// Error response.
64    Error {
65        #[serde(default)]
66        code: Option<i32>,
67        msg: SmolStr,
68    },
69    /// Success message (auth confirmation, handled separately).
70    Success { msg: SmolStr },
71}
72
73impl AlpacaSubResponseInner {
74    /// Every `(channel, symbol)` pair this message reports the connection holding.
75    ///
76    /// Only `Subscription` names symbols; `Success` and `Error` cover nothing. The report is the
77    /// connection's whole state, not only what the last request added.
78    pub(super) fn covered(&self) -> impl Iterator<Item = (AlpacaChannel, &SmolStr)> {
79        let (trades, quotes): (&[SmolStr], &[SmolStr]) = match self {
80            // `bars` is deliberately ignored: AlpacaChannel models only trades and quotes, so no
81            // subscription is ever keyed against a bars channel and nothing could match it.
82            Self::Subscription {
83                trades,
84                quotes,
85                bars: _,
86            } => (trades, quotes),
87            Self::Error { .. } | Self::Success { .. } => (&[], &[]),
88        };
89
90        let trades = trades.iter().map(|market| (AlpacaChannel::Trades, market));
91        let quotes = quotes.iter().map(|market| (AlpacaChannel::Quotes, market));
92
93        trades.chain(quotes)
94    }
95}
96
97#[cfg(test)]
98#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
99mod tests {
100    use super::*;
101
102    #[test]
103    fn test_de_subscription_success() {
104        let input = r#"[{"T":"subscription","trades":["AAPL","SPY"],"quotes":["AAPL"]}]"#;
105        let response: AlpacaSubResponse = serde_json::from_str(input).unwrap();
106
107        match &response.0[0] {
108            AlpacaSubResponseInner::Subscription { trades, quotes, .. } => {
109                assert_eq!(trades, &vec!["AAPL", "SPY"]);
110                assert_eq!(quotes, &vec!["AAPL"]);
111            }
112            _ => panic!("expected Subscription variant"),
113        }
114    }
115
116    #[test]
117    fn test_de_subscription_trades_only() {
118        let input = r#"[{"T":"subscription","trades":["BTC/USD"]}]"#;
119        let response: AlpacaSubResponse = serde_json::from_str(input).unwrap();
120
121        match &response.0[0] {
122            AlpacaSubResponseInner::Subscription { trades, quotes, .. } => {
123                assert_eq!(trades, &vec!["BTC/USD"]);
124                assert!(quotes.is_empty());
125            }
126            _ => panic!("expected Subscription variant"),
127        }
128    }
129
130    #[test]
131    fn test_de_error() {
132        let input = r#"[{"T":"error","code":400,"msg":"invalid syntax"}]"#;
133        let response: AlpacaSubResponse = serde_json::from_str(input).unwrap();
134
135        match &response.0[0] {
136            AlpacaSubResponseInner::Error { code, msg } => {
137                assert_eq!(*code, Some(400));
138                assert_eq!(msg, "invalid syntax");
139            }
140            _ => panic!("expected Error variant"),
141        }
142    }
143
144    #[test]
145    fn test_de_success() {
146        let input = r#"[{"T":"success","msg":"authenticated"}]"#;
147        let response: AlpacaSubResponse = serde_json::from_str(input).unwrap();
148
149        match &response.0[0] {
150            AlpacaSubResponseInner::Success { msg } => {
151                assert_eq!(msg, "authenticated");
152            }
153            _ => panic!("expected Success variant"),
154        }
155    }
156
157    #[test]
158    fn test_validate_subscription() {
159        let response = AlpacaSubResponse(vec![AlpacaSubResponseInner::Subscription {
160            trades: vec!["AAPL".into()],
161            quotes: vec![],
162            bars: vec![],
163        }]);
164        assert!(response.validate().is_ok());
165    }
166
167    #[test]
168    fn test_validate_error() {
169        let response = AlpacaSubResponse(vec![AlpacaSubResponseInner::Error {
170            code: Some(400),
171            msg: "bad request".into(),
172        }]);
173        assert!(response.validate().is_err());
174    }
175
176    #[test]
177    fn test_validate_success_passes() {
178        // Success messages are treated as benign during subscription validation
179        // (they may arrive if auth confirmation is re-sent)
180        let response = AlpacaSubResponse(vec![AlpacaSubResponseInner::Success {
181            msg: "authenticated".into(),
182        }]);
183        assert!(response.validate().is_ok());
184    }
185}
rustrade_data/exchange/alpaca/subscription.rs

rustrade_data/exchange/alpaca/
subscription.rs
