finance-query 3.0.0

//! Live options-chain streaming.
//!
//! A dedicated stream type rather than a reuse of [`PriceUpdate`]: a chain is
//! many contracts per underlying, each with its own quote, open interest and
//! greeks — a shape a single-symbol price tick cannot carry.

use std::sync::Arc;
use std::time::Duration;

use serde::{Deserialize, Serialize};

use super::client::StreamResult;
use super::handle::{RECONNECT_BACKOFF, SourceStream, stream_builder, stream_handle};
use super::polygon::PolygonOptionsSource;
use super::pricing::OptionType;
use super::source::ReconnectConfig;

/// Channel capacity — a wide chain fans out many contracts per tick.
const CHANNEL_CAPACITY: usize = 2048;

/// Default interval between greeks/open-interest snapshot refreshes.
const DEFAULT_GREEKS_REFRESH: Duration = Duration::from_secs(60);

/// Option greeks for a contract.
#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct Greeks {
    /// Rate of change of price with respect to the underlying.
    pub delta: Option<f64>,
    /// Rate of change of delta with respect to the underlying.
    pub gamma: Option<f64>,
    /// Rate of change of price with respect to time.
    pub theta: Option<f64>,
    /// Rate of change of price with respect to volatility.
    pub vega: Option<f64>,
}

impl Greeks {
    /// `true` when no greek was populated.
    pub fn is_empty(&self) -> bool {
        self.delta.is_none() && self.gamma.is_none() && self.theta.is_none() && self.vega.is_none()
    }
}

/// A live update for one options contract.
///
/// Quote and trade fields arrive from the real-time WebSocket; `greeks`,
/// `implied_volatility` and `open_interest` come from the periodic chain
/// snapshot (see [`OptionsChainStreamBuilder::greeks_refresh`]) because the
/// real-time feed does not carry them.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct OptionContractUpdate {
    /// OCC contract symbol (e.g. `"O:AAPL250117C00150000"`).
    pub contract_symbol: String,
    /// Underlying symbol parsed from the contract (e.g. `"AAPL"`).
    pub underlying: String,
    /// Expiration date as a Unix timestamp (seconds).
    pub expiration: Option<i64>,
    /// Strike price.
    pub strike: Option<f64>,
    /// Call or put.
    pub option_type: Option<OptionType>,
    /// Best bid.
    pub bid: Option<f64>,
    /// Best bid size.
    pub bid_size: Option<f64>,
    /// Best ask.
    pub ask: Option<f64>,
    /// Best ask size.
    pub ask_size: Option<f64>,
    /// Last traded price.
    pub last_price: Option<f64>,
    /// Last traded size.
    pub last_size: Option<f64>,
    /// Day volume, when the snapshot supplies it.
    pub volume: Option<i64>,
    /// Open interest, from the snapshot refresh.
    pub open_interest: Option<i64>,
    /// Implied volatility, from the snapshot refresh.
    pub implied_volatility: Option<f64>,
    /// Greeks, from the snapshot refresh.
    pub greeks: Option<Greeks>,
    /// Event timestamp (milliseconds).
    pub time: i64,
}

/// Contract metadata decoded from an OCC symbol.
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct ContractParts {
    pub(crate) underlying: String,
    pub(crate) expiration: i64,
    pub(crate) option_type: OptionType,
    pub(crate) strike: f64,
}

/// Decode an OCC-style contract symbol (`O:AAPL250117C00150000`).
///
/// Returns `None` for anything that does not match the layout, so a malformed
/// upstream ticker is skipped rather than mislabeled.
pub(crate) fn parse_contract_symbol(symbol: &str) -> Option<ContractParts> {
    let body = symbol.strip_prefix("O:").unwrap_or(symbol);
    // Trailing fixed-width fields: 6 date + 1 type + 8 strike.
    if body.len() < 16 {
        return None;
    }
    let split = body.len() - 15;
    let (underlying, rest) = body.split_at(split);
    if underlying.is_empty() || !underlying.chars().all(|c| c.is_ascii_alphanumeric()) {
        return None;
    }

    let (date, rest) = rest.split_at(6);
    let (kind, strike) = rest.split_at(1);
    if !date.chars().all(|c| c.is_ascii_digit()) || !strike.chars().all(|c| c.is_ascii_digit()) {
        return None;
    }

    let option_type = match kind {
        "C" => OptionType::Call,
        "P" => OptionType::Put,
        _ => return None,
    };

    let year = 2000 + date[0..2].parse::<i32>().ok()?;
    let month = date[2..4].parse::<u32>().ok()?;
    let day = date[4..6].parse::<u32>().ok()?;
    let expiration = chrono::NaiveDate::from_ymd_opt(year, month, day)?
        .and_hms_opt(0, 0, 0)?
        .and_utc()
        .timestamp();

    Some(ContractParts {
        underlying: underlying.to_string(),
        expiration,
        option_type,
        strike: strike.parse::<f64>().ok()? / 1000.0,
    })
}

stream_handle! {
    /// A live subscription to one or more options chains.
    ///
    /// Subscribe by underlying (`"AAPL"`) to follow the whole chain, or by full
    /// OCC symbol (`"O:AAPL250117C00150000"`) to follow single contracts.
    /// Requires the `polygon` feature and the `POLYGON_API_KEY` environment
    /// variable set.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::streaming::OptionsChainStream;
    /// use futures::StreamExt;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut stream = OptionsChainStream::subscribe(["AAPL"]).await?;
    ///
    /// while let Some(contract) = stream.next().await {
    ///     println!("{} {:?}/{:?}", contract.contract_symbol, contract.bid, contract.ask);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    OptionsChainStream(OptionContractUpdate);
    add: add = "Add underlyings or contracts to the subscription.",
    remove: remove = "Remove underlyings or contracts from the subscription.",
}

impl OptionsChainStream {
    /// Subscribe to the chains of the given underlyings.
    pub async fn subscribe<S, I>(underlyings: I) -> StreamResult<Self>
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        OptionsChainStreamBuilder::new()
            .underlyings(underlyings)
            .build()
            .await
    }
}

/// Builder for an [`OptionsChainStream`].
pub struct OptionsChainStreamBuilder {
    underlyings: Vec<String>,
    retry_delay: Duration,
    max_reconnect_attempts: Option<u32>,
    greeks_refresh: Option<Duration>,
}

impl OptionsChainStreamBuilder {
    /// Create a builder with no symbols and default timings.
    pub fn new() -> Self {
        Self {
            underlyings: Vec::new(),
            retry_delay: RECONNECT_BACKOFF,
            max_reconnect_attempts: None,
            greeks_refresh: Some(DEFAULT_GREEKS_REFRESH),
        }
    }

    /// Interval between greeks/open-interest snapshot refreshes.
    ///
    /// `None` disables them, leaving only WebSocket bid/ask/last (one REST
    /// call per underlying per interval otherwise). Default: 60s.
    pub fn greeks_refresh(mut self, interval: Option<Duration>) -> Self {
        self.greeks_refresh = interval;
        self
    }

    /// Build and start the stream.
    pub async fn build(self) -> StreamResult<OptionsChainStream> {
        let source = Arc::new(PolygonOptionsSource::new(self.greeks_refresh));
        let reconnect =
            ReconnectConfig::new(self.retry_delay).max_attempts(self.max_reconnect_attempts);
        Ok(OptionsChainStream {
            inner: SourceStream::start(source, self.underlyings, reconnect, CHANNEL_CAPACITY),
        })
    }
}

stream_builder!(
    OptionsChainStreamBuilder,
    underlyings = "Add underlyings (or full OCC contract symbols) to follow."
);

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_a_call_contract_symbol() {
        let parts = parse_contract_symbol("O:AAPL250117C00150000").expect("should parse");
        assert_eq!(parts.underlying, "AAPL");
        assert_eq!(parts.option_type, OptionType::Call);
        assert!((parts.strike - 150.0).abs() < 1e-9);
        // 2025-01-17T00:00:00Z
        assert_eq!(parts.expiration, 1737072000);
    }

    #[test]
    fn parses_a_put_and_a_fractional_strike() {
        let parts = parse_contract_symbol("O:SPY261218P00512500").expect("should parse");
        assert_eq!(parts.underlying, "SPY");
        assert_eq!(parts.option_type, OptionType::Put);
        assert!((parts.strike - 512.5).abs() < 1e-9);
    }

    #[test]
    fn parses_without_the_o_prefix() {
        assert_eq!(
            parse_contract_symbol("AAPL250117C00150000")
                .unwrap()
                .underlying,
            "AAPL"
        );
    }

    #[test]
    fn rejects_malformed_symbols() {
        for bad in [
            "O:AAPL",
            "AAPL250117X00150000",
            "O:AAPL2501I7C00150000",
            "",
            "O:250117C00150000",
        ] {
            assert!(
                parse_contract_symbol(bad).is_none(),
                "expected {bad} to be rejected"
            );
        }
    }

    #[test]
    fn greeks_report_emptiness() {
        assert!(Greeks::default().is_empty());
        assert!(
            !Greeks {
                delta: Some(0.5),
                ..Default::default()
            }
            .is_empty()
        );
    }
}