rustyschwab_types/
streaming.rs

1//! Real-time streaming WebSocket types for the Schwab Streamer API.
2//!
3//! This module contains types for the WebSocket streaming interface:
4//!
5//! - Request types for subscriptions and commands
6//! - Response types for acknowledgments
7//! - Data types for real-time market data
8//! - Service enums, field tables, and typed payloads per service
9//!
10//! # Services
11//!
12//! The streaming API supports multiple services:
13//! - `LEVELONE_EQUITIES` - Real-time equity quotes
14//! - `LEVELONE_OPTIONS` - Real-time option quotes
15//! - `CHART_EQUITY` - Real-time chart data
16//! - `ACCT_ACTIVITY` - Account activity notifications
17//!
18//! Data frames arrive with numbered fields
19//! ([`DataContent::fields`](crate::streaming::DataContent::fields));
20//! [`DataItem::decode`](crate::streaming::DataItem::decode) and the
21//! per-service `from_content` constructors turn them into
22//! [`StreamPayload`](crate::streaming::StreamPayload) values with named fields.
23
24mod payload;
25
26pub use payload::*;
27
28use crate::common::{AssetMainType, AssetSubType};
29use serde::{Deserialize, Serialize};
30use serde_json::Value;
31use std::collections::HashMap;
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
35#[serde(rename_all = "camelCase")]
36pub struct StreamRequest {
37    pub requestid: String,
38    pub service: String,
39    pub command: StreamCommand,
40    #[serde(rename = "SchwabClientCustomerId")]
41    pub schwab_client_customer_id: String,
42    #[serde(rename = "SchwabClientCorrelId")]
43    pub schwab_client_correl_id: String,
44    pub parameters: StreamParameters,
45}
46
47#[derive(Debug, Clone, Serialize, Deserialize)]
48#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
49#[serde(rename_all = "camelCase")]
50pub struct StreamRequests {
51    pub requests: Vec<StreamRequest>,
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize)]
55#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
56#[serde(rename_all = "UPPERCASE")]
57#[non_exhaustive]
58pub enum StreamCommand {
59    Login,
60    Logout,
61    Subs,
62    Add,
63    Unsubs,
64    View,
65}
66
67#[derive(Debug, Clone, Serialize, Deserialize)]
68#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
69#[serde(rename_all = "camelCase")]
70pub struct StreamParameters {
71    #[serde(skip_serializing_if = "Option::is_none")]
72    pub keys: Option<String>,
73    #[serde(skip_serializing_if = "Option::is_none")]
74    pub fields: Option<String>,
75    #[serde(rename = "Authorization", skip_serializing_if = "Option::is_none")]
76    pub authorization: Option<String>,
77    #[serde(
78        rename = "SchwabClientChannel",
79        skip_serializing_if = "Option::is_none"
80    )]
81    pub schwab_client_channel: Option<String>,
82    #[serde(
83        rename = "SchwabClientFunctionId",
84        skip_serializing_if = "Option::is_none"
85    )]
86    pub schwab_client_function_id: Option<String>,
87}
88
89#[derive(Debug, Clone, Serialize, Deserialize)]
90#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
91#[serde(untagged)]
92#[non_exhaustive]
93pub enum StreamMessage {
94    #[serde(skip_deserializing)]
95    State(StreamStatus),
96    Response(StreamResponse),
97    Data(StreamData),
98    Notify(StreamNotify),
99}
100
101#[derive(Debug, Clone, Serialize, Deserialize)]
102#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
103#[non_exhaustive]
104pub struct StreamResponse {
105    pub response: Vec<ResponseItem>,
106}
107
108#[derive(Debug, Clone, Serialize, Deserialize)]
109#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
110#[serde(rename_all = "camelCase")]
111#[non_exhaustive]
112pub struct ResponseItem {
113    pub service: String,
114    pub command: String,
115    pub requestid: String,
116    #[serde(rename = "SchwabClientCorrelId")]
117    pub schwab_client_correl_id: String,
118    pub timestamp: i64,
119    pub content: ResponseContent,
120}
121
122#[derive(Debug, Clone, Serialize, Deserialize)]
123#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
124#[non_exhaustive]
125pub struct ResponseContent {
126    pub code: i32,
127    pub msg: String,
128}
129
130#[derive(Debug, Clone, Serialize, Deserialize)]
131#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
132#[non_exhaustive]
133pub struct StreamData {
134    pub data: Vec<DataItem>,
135}
136
137#[derive(Debug, Clone, Serialize, Deserialize)]
138#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
139#[serde(rename_all = "camelCase")]
140#[non_exhaustive]
141pub struct DataItem {
142    pub service: String,
143    pub timestamp: i64,
144    pub command: String,
145    pub content: Vec<DataContent>,
146}
147
148/// One item of a data frame as the streamer sends it: the subscription
149/// `key`, the service's envelope keys, and the numbered fields.
150///
151/// Decode into named, typed fields with [`StreamService::decode`] or a
152/// payload's `from_content` (for example [`LevelOneEquity::from_content`]).
153#[derive(Debug, Clone, Serialize, Deserialize)]
154#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
155#[non_exhaustive]
156pub struct DataContent {
157    pub key: String,
158    #[serde(skip_serializing_if = "Option::is_none")]
159    pub delayed: Option<bool>,
160    #[serde(rename = "assetMainType", skip_serializing_if = "Option::is_none")]
161    pub asset_main_type: Option<AssetMainType>,
162    #[serde(rename = "assetSubType", skip_serializing_if = "Option::is_none")]
163    pub asset_sub_type: Option<AssetSubType>,
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub cusip: Option<String>,
166    /// The numbered fields (`"1"`, `"2"`, …) plus any other key the service
167    /// sends, such as `seq` on chart and account-activity items.
168    #[serde(flatten)]
169    pub fields: HashMap<String, Value>,
170}
171
172#[derive(Debug, Clone, Serialize, Deserialize)]
173#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
174#[non_exhaustive]
175pub struct StreamNotify {
176    pub notify: Vec<NotifyItem>,
177}
178
179/// Schwab documents only the heartbeat notification. Anything else it sends
180/// here (an `ADMIN` notice with a `content.code`, say) is kept in `extra`
181/// rather than dropped.
182#[derive(Debug, Clone, Serialize, Deserialize)]
183#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
184#[non_exhaustive]
185pub struct NotifyItem {
186    #[serde(skip_serializing_if = "Option::is_none")]
187    pub heartbeat: Option<String>,
188    #[serde(flatten)]
189    pub extra: HashMap<String, Value>,
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
193#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
194#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
195#[non_exhaustive]
196pub enum StreamService {
197    Admin,
198    LeveloneEquities,
199    LeveloneOptions,
200    LeveloneFutures,
201    LeveloneFuturesOptions,
202    LeveloneForex,
203    NyseBook,
204    NasdaqBook,
205    OptionsBook,
206    ChartEquity,
207    ChartFutures,
208    ScreenerEquity,
209    ScreenerOption,
210    AcctActivity,
211}
212
213impl StreamService {
214    pub fn as_str(&self) -> &str {
215        match self {
216            StreamService::Admin => "ADMIN",
217            StreamService::LeveloneEquities => "LEVELONE_EQUITIES",
218            StreamService::LeveloneOptions => "LEVELONE_OPTIONS",
219            StreamService::LeveloneFutures => "LEVELONE_FUTURES",
220            StreamService::LeveloneFuturesOptions => "LEVELONE_FUTURES_OPTIONS",
221            StreamService::LeveloneForex => "LEVELONE_FOREX",
222            StreamService::NyseBook => "NYSE_BOOK",
223            StreamService::NasdaqBook => "NASDAQ_BOOK",
224            StreamService::OptionsBook => "OPTIONS_BOOK",
225            StreamService::ChartEquity => "CHART_EQUITY",
226            StreamService::ChartFutures => "CHART_FUTURES",
227            StreamService::ScreenerEquity => "SCREENER_EQUITY",
228            StreamService::ScreenerOption => "SCREENER_OPTION",
229            StreamService::AcctActivity => "ACCT_ACTIVITY",
230        }
231    }
232
233    /// Parse a service from its API wire name (e.g. `LEVELONE_EQUITIES`).
234    pub fn from_api_str(s: &str) -> Option<Self> {
235        match s {
236            "ADMIN" => Some(StreamService::Admin),
237            "LEVELONE_EQUITIES" => Some(StreamService::LeveloneEquities),
238            "LEVELONE_OPTIONS" => Some(StreamService::LeveloneOptions),
239            "LEVELONE_FUTURES" => Some(StreamService::LeveloneFutures),
240            "LEVELONE_FUTURES_OPTIONS" => Some(StreamService::LeveloneFuturesOptions),
241            "LEVELONE_FOREX" => Some(StreamService::LeveloneForex),
242            "NYSE_BOOK" => Some(StreamService::NyseBook),
243            "NASDAQ_BOOK" => Some(StreamService::NasdaqBook),
244            "OPTIONS_BOOK" => Some(StreamService::OptionsBook),
245            "CHART_EQUITY" => Some(StreamService::ChartEquity),
246            "CHART_FUTURES" => Some(StreamService::ChartFutures),
247            "SCREENER_EQUITY" => Some(StreamService::ScreenerEquity),
248            "SCREENER_OPTION" => Some(StreamService::ScreenerOption),
249            "ACCT_ACTIVITY" => Some(StreamService::AcctActivity),
250            _ => None,
251        }
252    }
253
254    /// Default subscription field IDs for this service.
255    ///
256    /// Every field of the service's table ([`fields`](Self::fields)) except
257    /// for `LEVELONE_EQUITIES`, which starts with the quote-sized subset in
258    /// [`LevelOneEquityFields::default_fields`] since watchlists are large.
259    /// Requesting a number beyond a table is rejected by the streamer with a
260    /// bad-command error.
261    pub fn default_fields(self) -> String {
262        match self {
263            Self::LeveloneEquities => LevelOneEquityFields::default_fields(),
264            service => payload::csv(service.fields().iter().map(|field| field.number)),
265        }
266    }
267}
268
269#[derive(Debug, Clone)]
270#[non_exhaustive]
271pub enum StreamResponseCode {
272    Success = 0,
273    LoginDenied = 3,
274    UnknownFailure = 9,
275    ServiceNotAvailable = 11,
276    CloseConnection = 12,
277    ReachedSymbolLimit = 19,
278    StreamConnNotFound = 20,
279    BadCommandFormat = 21,
280    FailedCommandSubs = 22,
281    FailedCommandUnsubs = 23,
282    FailedCommandAdd = 24,
283    FailedCommandView = 25,
284    SucceededCommandSubs = 26,
285    SucceededCommandUnsubs = 27,
286    SucceededCommandAdd = 28,
287    SucceededCommandView = 29,
288    StopStreaming = 30,
289}
290
291impl StreamResponseCode {
292    pub fn from_code(code: i32) -> Option<Self> {
293        match code {
294            0 => Some(Self::Success),
295            3 => Some(Self::LoginDenied),
296            9 => Some(Self::UnknownFailure),
297            11 => Some(Self::ServiceNotAvailable),
298            12 => Some(Self::CloseConnection),
299            19 => Some(Self::ReachedSymbolLimit),
300            20 => Some(Self::StreamConnNotFound),
301            21 => Some(Self::BadCommandFormat),
302            22 => Some(Self::FailedCommandSubs),
303            23 => Some(Self::FailedCommandUnsubs),
304            24 => Some(Self::FailedCommandAdd),
305            25 => Some(Self::FailedCommandView),
306            26 => Some(Self::SucceededCommandSubs),
307            27 => Some(Self::SucceededCommandUnsubs),
308            28 => Some(Self::SucceededCommandAdd),
309            29 => Some(Self::SucceededCommandView),
310            30 => Some(Self::StopStreaming),
311            _ => None,
312        }
313    }
314
315    pub fn is_success(&self) -> bool {
316        matches!(
317            self,
318            Self::Success
319                | Self::SucceededCommandSubs
320                | Self::SucceededCommandUnsubs
321                | Self::SucceededCommandAdd
322                | Self::SucceededCommandView
323        )
324    }
325
326    pub fn is_retryable(&self) -> bool {
327        matches!(
328            self,
329            Self::UnknownFailure | Self::ServiceNotAvailable | Self::StreamConnNotFound
330        )
331    }
332}
333
334/// Local lifecycle notification, never accepted from the broker wire.
335/// `lost_messages` is cumulative; any increase requires REST reconciliation
336/// for account activity. A new session does not recover missed executions.
337/// It counts exactly the frames dropped at a full queue. Frames queued
338/// before a gap are delivered ahead of the status that reports it.
339///
340/// `ended` is set when the stream stopped for good: no reconnect follows
341/// until the caller connects again. A receiver reports that status, then
342/// ends once nothing is left to deliver.
343#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
344#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
345pub struct StreamStatus {
346    pub ready: bool,
347    pub generation: u64,
348    pub lost_messages: u64,
349    #[serde(default, skip_serializing_if = "Option::is_none")]
350    pub ended: Option<StreamEnd>,
351}
352
353/// Why a stream stopped for good.
354#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
355#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
356#[serde(tag = "reason", rename_all = "snake_case")]
357#[non_exhaustive]
358pub enum StreamEnd {
359    /// The caller disconnected.
360    Disconnected,
361    /// Another session took over this login's streamer slot (response code
362    /// 12, `CLOSE_CONNECTION`); Schwab allows one per user.
363    Replaced,
364    /// Schwab stopped streaming to a session with no subscriptions
365    /// (response code 30, `STOP_STREAMING`).
366    Idle,
367    /// The server closed the socket normally without saying why.
368    ServerClosed,
369    /// Schwab refused the login, even with a freshly refreshed token.
370    LoginDenied { message: String },
371    /// The first connection never got as far as a login.
372    ConnectFailed { message: String },
373    /// The connection was lost and reconnecting is disabled or out of
374    /// attempts.
375    ReconnectExhausted,
376}
rustyschwab_types/streaming.rs

rustyschwab_types/
streaming.rs
