finance_query/streaming/
options.rs

1//! Live options-chain streaming.
2//!
3//! A dedicated stream type rather than a reuse of [`PriceUpdate`]: a chain is
4//! many contracts per underlying, each with its own quote, open interest and
5//! greeks — a shape a single-symbol price tick cannot carry.
6
7use std::sync::Arc;
8use std::time::Duration;
9
10use serde::{Deserialize, Serialize};
11
12use super::client::StreamResult;
13use super::handle::{RECONNECT_BACKOFF, SourceStream, stream_builder, stream_handle};
14use super::polygon::PolygonOptionsSource;
15use super::pricing::OptionType;
16use super::source::ReconnectConfig;
17
18/// Channel capacity — a wide chain fans out many contracts per tick.
19const CHANNEL_CAPACITY: usize = 2048;
20
21/// Default interval between greeks/open-interest snapshot refreshes.
22const DEFAULT_GREEKS_REFRESH: Duration = Duration::from_secs(60);
23
24/// Option greeks for a contract.
25#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
26#[serde(rename_all = "camelCase")]
27#[non_exhaustive]
28pub struct Greeks {
29    /// Rate of change of price with respect to the underlying.
30    pub delta: Option<f64>,
31    /// Rate of change of delta with respect to the underlying.
32    pub gamma: Option<f64>,
33    /// Rate of change of price with respect to time.
34    pub theta: Option<f64>,
35    /// Rate of change of price with respect to volatility.
36    pub vega: Option<f64>,
37}
38
39impl Greeks {
40    /// `true` when no greek was populated.
41    pub fn is_empty(&self) -> bool {
42        self.delta.is_none() && self.gamma.is_none() && self.theta.is_none() && self.vega.is_none()
43    }
44}
45
46/// A live update for one options contract.
47///
48/// Quote and trade fields arrive from the real-time WebSocket; `greeks`,
49/// `implied_volatility` and `open_interest` come from the periodic chain
50/// snapshot (see [`OptionsChainStreamBuilder::greeks_refresh`]) because the
51/// real-time feed does not carry them.
52#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
53#[serde(rename_all = "camelCase")]
54#[non_exhaustive]
55pub struct OptionContractUpdate {
56    /// OCC contract symbol (e.g. `"O:AAPL250117C00150000"`).
57    pub contract_symbol: String,
58    /// Underlying symbol parsed from the contract (e.g. `"AAPL"`).
59    pub underlying: String,
60    /// Expiration date as a Unix timestamp (seconds).
61    pub expiration: Option<i64>,
62    /// Strike price.
63    pub strike: Option<f64>,
64    /// Call or put.
65    pub option_type: Option<OptionType>,
66    /// Best bid.
67    pub bid: Option<f64>,
68    /// Best bid size.
69    pub bid_size: Option<f64>,
70    /// Best ask.
71    pub ask: Option<f64>,
72    /// Best ask size.
73    pub ask_size: Option<f64>,
74    /// Last traded price.
75    pub last_price: Option<f64>,
76    /// Last traded size.
77    pub last_size: Option<f64>,
78    /// Day volume, when the snapshot supplies it.
79    pub volume: Option<i64>,
80    /// Open interest, from the snapshot refresh.
81    pub open_interest: Option<i64>,
82    /// Implied volatility, from the snapshot refresh.
83    pub implied_volatility: Option<f64>,
84    /// Greeks, from the snapshot refresh.
85    pub greeks: Option<Greeks>,
86    /// Event timestamp (milliseconds).
87    pub time: i64,
88}
89
90/// Contract metadata decoded from an OCC symbol.
91#[derive(Clone, Debug, PartialEq)]
92pub(crate) struct ContractParts {
93    pub(crate) underlying: String,
94    pub(crate) expiration: i64,
95    pub(crate) option_type: OptionType,
96    pub(crate) strike: f64,
97}
98
99/// Decode an OCC-style contract symbol (`O:AAPL250117C00150000`).
100///
101/// Returns `None` for anything that does not match the layout, so a malformed
102/// upstream ticker is skipped rather than mislabeled.
103pub(crate) fn parse_contract_symbol(symbol: &str) -> Option<ContractParts> {
104    let body = symbol.strip_prefix("O:").unwrap_or(symbol);
105    // Trailing fixed-width fields: 6 date + 1 type + 8 strike.
106    if body.len() < 16 {
107        return None;
108    }
109    let split = body.len() - 15;
110    let (underlying, rest) = body.split_at(split);
111    if underlying.is_empty() || !underlying.chars().all(|c| c.is_ascii_alphanumeric()) {
112        return None;
113    }
114
115    let (date, rest) = rest.split_at(6);
116    let (kind, strike) = rest.split_at(1);
117    if !date.chars().all(|c| c.is_ascii_digit()) || !strike.chars().all(|c| c.is_ascii_digit()) {
118        return None;
119    }
120
121    let option_type = match kind {
122        "C" => OptionType::Call,
123        "P" => OptionType::Put,
124        _ => return None,
125    };
126
127    let year = 2000 + date[0..2].parse::<i32>().ok()?;
128    let month = date[2..4].parse::<u32>().ok()?;
129    let day = date[4..6].parse::<u32>().ok()?;
130    let expiration = chrono::NaiveDate::from_ymd_opt(year, month, day)?
131        .and_hms_opt(0, 0, 0)?
132        .and_utc()
133        .timestamp();
134
135    Some(ContractParts {
136        underlying: underlying.to_string(),
137        expiration,
138        option_type,
139        strike: strike.parse::<f64>().ok()? / 1000.0,
140    })
141}
142
143stream_handle! {
144    /// A live subscription to one or more options chains.
145    ///
146    /// Subscribe by underlying (`"AAPL"`) to follow the whole chain, or by full
147    /// OCC symbol (`"O:AAPL250117C00150000"`) to follow single contracts.
148    /// Requires the `polygon` feature and the `POLYGON_API_KEY` environment
149    /// variable set.
150    ///
151    /// # Example
152    ///
153    /// ```no_run
154    /// use finance_query::streaming::OptionsChainStream;
155    /// use futures::StreamExt;
156    ///
157    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
158    /// let mut stream = OptionsChainStream::subscribe(["AAPL"]).await?;
159    ///
160    /// while let Some(contract) = stream.next().await {
161    ///     println!("{} {:?}/{:?}", contract.contract_symbol, contract.bid, contract.ask);
162    /// }
163    /// # Ok(())
164    /// # }
165    /// ```
166    OptionsChainStream(OptionContractUpdate);
167    add: add = "Add underlyings or contracts to the subscription.",
168    remove: remove = "Remove underlyings or contracts from the subscription.",
169}
170
171impl OptionsChainStream {
172    /// Subscribe to the chains of the given underlyings.
173    pub async fn subscribe<S, I>(underlyings: I) -> StreamResult<Self>
174    where
175        S: Into<String>,
176        I: IntoIterator<Item = S>,
177    {
178        OptionsChainStreamBuilder::new()
179            .underlyings(underlyings)
180            .build()
181            .await
182    }
183}
184
185/// Builder for an [`OptionsChainStream`].
186pub struct OptionsChainStreamBuilder {
187    underlyings: Vec<String>,
188    retry_delay: Duration,
189    max_reconnect_attempts: Option<u32>,
190    greeks_refresh: Option<Duration>,
191}
192
193impl OptionsChainStreamBuilder {
194    /// Create a builder with no symbols and default timings.
195    pub fn new() -> Self {
196        Self {
197            underlyings: Vec::new(),
198            retry_delay: RECONNECT_BACKOFF,
199            max_reconnect_attempts: None,
200            greeks_refresh: Some(DEFAULT_GREEKS_REFRESH),
201        }
202    }
203
204    /// Interval between greeks/open-interest snapshot refreshes.
205    ///
206    /// `None` disables them, leaving only WebSocket bid/ask/last (one REST
207    /// call per underlying per interval otherwise). Default: 60s.
208    pub fn greeks_refresh(mut self, interval: Option<Duration>) -> Self {
209        self.greeks_refresh = interval;
210        self
211    }
212
213    /// Build and start the stream.
214    pub async fn build(self) -> StreamResult<OptionsChainStream> {
215        let source = Arc::new(PolygonOptionsSource::new(self.greeks_refresh));
216        let reconnect =
217            ReconnectConfig::new(self.retry_delay).max_attempts(self.max_reconnect_attempts);
218        Ok(OptionsChainStream {
219            inner: SourceStream::start(source, self.underlyings, reconnect, CHANNEL_CAPACITY),
220        })
221    }
222}
223
224stream_builder!(
225    OptionsChainStreamBuilder,
226    underlyings = "Add underlyings (or full OCC contract symbols) to follow."
227);
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232
233    #[test]
234    fn parses_a_call_contract_symbol() {
235        let parts = parse_contract_symbol("O:AAPL250117C00150000").expect("should parse");
236        assert_eq!(parts.underlying, "AAPL");
237        assert_eq!(parts.option_type, OptionType::Call);
238        assert!((parts.strike - 150.0).abs() < 1e-9);
239        // 2025-01-17T00:00:00Z
240        assert_eq!(parts.expiration, 1737072000);
241    }
242
243    #[test]
244    fn parses_a_put_and_a_fractional_strike() {
245        let parts = parse_contract_symbol("O:SPY261218P00512500").expect("should parse");
246        assert_eq!(parts.underlying, "SPY");
247        assert_eq!(parts.option_type, OptionType::Put);
248        assert!((parts.strike - 512.5).abs() < 1e-9);
249    }
250
251    #[test]
252    fn parses_without_the_o_prefix() {
253        assert_eq!(
254            parse_contract_symbol("AAPL250117C00150000")
255                .unwrap()
256                .underlying,
257            "AAPL"
258        );
259    }
260
261    #[test]
262    fn rejects_malformed_symbols() {
263        for bad in [
264            "O:AAPL",
265            "AAPL250117X00150000",
266            "O:AAPL2501I7C00150000",
267            "",
268            "O:250117C00150000",
269        ] {
270            assert!(
271                parse_contract_symbol(bad).is_none(),
272                "expected {bad} to be rejected"
273            );
274        }
275    }
276
277    #[test]
278    fn greeks_report_emptiness() {
279        assert!(Greeks::default().is_empty());
280        assert!(
281            !Greeks {
282                delta: Some(0.5),
283                ..Default::default()
284            }
285            .is_empty()
286        );
287    }
288}
finance_query/streaming/options.rs

finance_query/streaming/
options.rs
