fmp_rust_sdk/websocket/
streams.rs

1use std::{collections::BTreeSet, fmt, str::FromStr, time::Duration};
2
3use serde::Deserialize;
4
5use super::{WebSocketConfig, WebSocketError};
6
7const STREAM_NAMES: [&str; 14] = [
8    "nasdaq-basic-w-nls-plus",
9    "iex-tops",
10    "tsx-level-1",
11    "tsxv-level-1",
12    "lse-level-1",
13    "cboe-index-main",
14    "fmp-crypto-stream",
15    "fmp-index-stream",
16    "fmp-commodity-stream",
17    "fmp-currency-stream",
18    "fmp-us-equities-stream",
19    "fmp-us-otc-stream",
20    "fmp-uk-equities-stream",
21    "fmp-ca-equities-stream",
22];
23
24/// A documented standard WebSocket stream, including its delayed variant.
25#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
26pub struct MarketStream(String);
27
28impl MarketStream {
29    /// Returns the exact stream identifier sent to FMP.
30    pub fn as_str(&self) -> &str {
31        &self.0
32    }
33}
34
35impl FromStr for MarketStream {
36    type Err = WebSocketError;
37
38    fn from_str(input: &str) -> Result<Self, Self::Err> {
39        let name = input.strip_suffix("-delayed").unwrap_or(input);
40        if !STREAM_NAMES.contains(&name) {
41            return Err(WebSocketError::Configuration {
42                reason: format!(
43                    "unknown FMP WebSocket stream {input:?}; use a stream listed in FMP's standard WebSocket documentation, optionally ending in -delayed"
44                ),
45            });
46        }
47        Ok(Self(input.to_owned()))
48    }
49}
50
51impl fmt::Display for MarketStream {
52    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53        formatter.write_str(self.as_str())
54    }
55}
56
57/// Explicit stream grants confirmed in the account dashboard or with FMP.
58///
59/// A billing plan does not imply access to an exchange or delayed feed. The
60/// server remains authoritative about each configured grant.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct StreamEntitlements {
63    streams: BTreeSet<MarketStream>,
64}
65
66impl StreamEntitlements {
67    /// Creates an immutable allowlist containing at least one granted stream.
68    pub fn new(streams: Vec<MarketStream>) -> Result<Self, WebSocketError> {
69        if streams.is_empty() {
70            return Err(WebSocketError::Configuration {
71                reason: "configure at least one explicitly granted FMP WebSocket stream".to_owned(),
72            });
73        }
74        Ok(Self {
75            streams: streams.into_iter().collect(),
76        })
77    }
78
79    /// Returns whether this exact real-time or delayed stream has a grant.
80    pub fn allows(&self, stream: &MarketStream) -> bool {
81        self.streams.contains(stream)
82    }
83}
84
85/// Explicit timeouts and retry settings for the standard stream protocol.
86#[derive(Debug, Clone)]
87pub struct StreamConfig {
88    pub(super) config: WebSocketConfig,
89}
90
91impl StreamConfig {
92    /// Creates a configuration using FMP's documented nested login response.
93    ///
94    /// Timeouts and attempt count must be positive. Transient connection
95    /// failures are retried before login; active connections require an explicit
96    /// reconnect after failure.
97    pub fn new(
98        handshake_timeout: Duration,
99        receive_timeout: Duration,
100        connect_attempts: u32,
101        retry_delay: Duration,
102    ) -> Result<Self, WebSocketError> {
103        Ok(Self {
104            config: WebSocketConfig::new(
105                handshake_timeout,
106                receive_timeout,
107                connect_attempts,
108                retry_delay,
109                super::protocol::confirm_standard_login,
110            )?,
111        })
112    }
113}
114
115/// Required fields in the standard protocol's nested login response.
116#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
117pub struct AuthenticationStatus {
118    /// The provider's authentication status code; success is 200.
119    pub status: u16,
120    /// The provider's authentication message.
121    pub message: String,
122}
123
124/// A standard stream heartbeat with the provider's required timestamp.
125#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
126pub struct Heartbeat {
127    /// The provider timestamp, preserved without assuming a time unit.
128    pub timestamp: u64,
129}
130
131/// A standard stream quote with required symbol, price, and timestamp.
132///
133/// Additional quote metadata depends on the asset and stream. These fields
134/// accept null or absence and are validated when supplied. Unrelated provider
135/// fields are ignored.
136#[derive(Debug, Clone, PartialEq, Deserialize)]
137#[serde(rename_all = "camelCase")]
138pub struct StreamQuote {
139    /// The asset's required nonempty symbol.
140    pub symbol: String,
141    /// The asset's name, when supplied.
142    pub name: Option<String>,
143    /// The current price.
144    pub price: f64,
145    /// The price change as a percentage.
146    pub changes_percentage: Option<f64>,
147    /// The signed price change.
148    pub change: Option<f64>,
149    /// The current day's lowest price.
150    pub day_low: Option<f64>,
151    /// The current day's highest price.
152    pub day_high: Option<f64>,
153    /// The past year's lowest price.
154    pub year_low: Option<f64>,
155    /// The past year's highest price.
156    pub year_high: Option<f64>,
157    /// The market capitalization.
158    pub market_cap: Option<f64>,
159    /// The 50-day average price.
160    pub price_avg50: Option<f64>,
161    /// The 200-day average price.
162    pub price_avg200: Option<f64>,
163    /// The exchange identifier, when supplied.
164    pub exchange: Option<String>,
165    /// The traded volume; fractional quantities are preserved.
166    pub volume: Option<f64>,
167    /// The average traded volume.
168    pub avg_volume: Option<f64>,
169    /// The opening price.
170    pub open: Option<f64>,
171    /// The previous closing price.
172    pub previous_close: Option<f64>,
173    /// Earnings per share, when available.
174    pub eps: Option<f64>,
175    /// The price-to-earnings ratio, when available.
176    pub pe: Option<f64>,
177    /// The provider's earnings announcement timestamp, when available.
178    pub earnings_announcement: Option<String>,
179    /// Outstanding shares, when available.
180    pub shares_outstanding: Option<f64>,
181    /// The provider timestamp, preserved without assuming a time unit.
182    pub timestamp: u64,
183}
184
185pub(super) fn validate_stream_quote(quote: &StreamQuote) -> Result<(), &'static str> {
186    if quote.symbol.is_empty()
187        || quote
188            .symbol
189            .chars()
190            .any(|character| character.is_whitespace() || character.is_control())
191    {
192        return Err(
193            "quote symbol must be nonempty and contain no whitespace or control characters",
194        );
195    }
196    for value in [quote.name.as_ref(), quote.exchange.as_ref()]
197        .into_iter()
198        .flatten()
199    {
200        if value.trim().is_empty() || value.chars().any(char::is_control) {
201            return Err(
202                "quote name and exchange must be nonempty and contain no control characters when supplied",
203            );
204        }
205    }
206    for value in [
207        Some(quote.price),
208        quote.day_low,
209        quote.day_high,
210        quote.year_low,
211        quote.year_high,
212        quote.price_avg50,
213        quote.price_avg200,
214        quote.open,
215        quote.previous_close,
216    ]
217    .into_iter()
218    .flatten()
219    {
220        if !value.is_finite() {
221            return Err("quote prices must be finite; negative commodity prices are permitted");
222        }
223    }
224    for value in [
225        quote.market_cap,
226        quote.volume,
227        quote.avg_volume,
228        quote.shares_outstanding,
229    ]
230    .into_iter()
231    .flatten()
232    {
233        if !value.is_finite() || value < 0.0 {
234            return Err(
235                "quote market capitalization, volumes, and outstanding shares must be finite and nonnegative",
236            );
237        }
238    }
239    for value in [quote.change, quote.changes_percentage, quote.eps, quote.pe]
240        .into_iter()
241        .flatten()
242    {
243        if !value.is_finite() {
244            return Err(
245                "quote changes, earnings per share, and price-to-earnings ratio must be finite",
246            );
247        }
248    }
249    if quote
250        .earnings_announcement
251        .as_ref()
252        .is_some_and(|value| value.trim().is_empty() || value.chars().any(char::is_control))
253    {
254        return Err(
255            "quote earnings announcement must be nonempty and contain no control characters when supplied",
256        );
257    }
258    Ok(())
259}
fmp_rust_sdk/websocket/streams.rs

fmp_rust_sdk/websocket/
streams.rs
