rustrade-data 0.20.0

//! Alpaca Market Data Integration Tests
//!
//! These tests verify connectivity and data reception from Alpaca market data streams.
//!
//! # Status
//!
//! **Tested locally, CI planned (paper trading allowed by Alpaca):**
//! - Crypto streaming: trades and quotes (24/7, free)
//! - IEX equity streaming: trades and quotes (market hours, free)
//!
//! **NOT tested (requires Algo Trader Plus subscription):**
//! - SIP equity streaming (`test_sip_*`) — implemented but unverified
//!
//! Tests are marked `#[ignore]` to avoid CI failures without credentials.
//!
//! # Prerequisites
//!
//! 1. Alpaca paper trading account (https://app.alpaca.markets)
//! 2. Environment variables set (see .env.template):
//!    - ALPACA_API_KEY: API key
//!    - ALPACA_SECRET_KEY: Secret key
//!
//! # Running
//!
//! ```bash
//! # Load env vars from .env
//! source .env
//!
//! # Run all Alpaca data integration tests
//! cargo test --test alpaca_data --features alpaca -- --ignored
//!
//! # Run specific test
//! cargo test --test alpaca_data --features alpaca test_crypto_trade_stream -- --ignored
//! ```
//!
//! # Market Hours Note
//!
//! US equity market hours are 9:30 AM - 4:00 PM ET (9:30 PM - 4:00 AM SGT next day).
//! The IEX feed only provides trade data during market hours. Tests that wait for
//! trade events (`test_iex_*_receives_data`) may timeout outside market hours.
//! Connection and subscription tests work anytime.
//!
//! Crypto streams (BTC/USD, etc.) are available 24/7 and are the primary validation.
//!
//! - WebSocket tests are marked `#[serial]` to prevent connection conflicts. Alpaca allows one
//!   data-stream connection per feed and rejects another with "auth failed: connection limit
//!   exceeded". Clones of one `AlpacaSubscriber` share that connection, but each test builds its
//!   own subscriber, so running them in parallel fails every connection but the one that wins the
//!   race.

#![cfg(feature = "alpaca")]
// Test code: unwrap/expect panics are the correct failure mode for test assertions
#![allow(clippy::unwrap_used, clippy::expect_used)]

use futures_util::StreamExt;
use rust_decimal::Decimal;
use rustrade_data::{
    exchange::alpaca::{AlpacaCrypto, AlpacaIex, AlpacaSip, AlpacaSubscriber},
    streams::{
        Streams,
        reconnect::{Event, stream::ReconnectingStream},
    },
    subscription::{quote::Quotes, trade::PublicTrades},
};
use rustrade_instrument::instrument::market_data::kind::MarketDataInstrumentKind;
use serial_test::serial;
use std::{collections::BTreeSet, time::Duration};
use tracing_subscriber::{EnvFilter, fmt};

fn init_logging() {
    let _ = fmt()
        .with_env_filter(
            EnvFilter::builder()
                .with_default_directive(tracing::Level::DEBUG.into())
                .from_env_lossy(),
        )
        .try_init();
}

// ============================================================================
// Crypto Stream Tests (24/7 availability - primary validation)
// ============================================================================

#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_trade_stream_connection() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to crypto trade stream: {:?}",
        streams.err()
    );
    tracing::info!("Crypto trade stream connected and subscribed");
}

/// Every trade the crypto stream delivers inside a bounded window is well formed and routed to the
/// subscribed instrument. Whether a trade arrives at all is not asserted.
///
/// Alpaca's crypto feed carries its own venue's fills, not a consolidated tape, and the delay
/// before the first BTC/USD trade is large and variable: within one half-hour window (2026-09-19)
/// it measured under 60s, ~136s, ~175s, and none at all in ~189s. No deadline separates "not
/// subscribed" from "not yet traded", so requiring a trade made the test fail on market activity.
/// The subscription itself is asserted deterministically by `init()`: the subscriber holds it open
/// until the venue names every requested symbol, as `test_crypto_trade_stream_connection` also
/// relies on.
#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_trade_stream_validates_delivered_trades() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await
        .expect("Failed to init crypto stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    // Polled in short slices so the closing summary can tell a quiet stream from an eventful one:
    // a slice with nothing in it counts as a quiet window, a reconnect or other non-item event as a
    // non-trade event. A stream that ends is the only outcome that fails without a trade.
    const WINDOW: Duration = Duration::from_secs(60);
    const SLICE: Duration = Duration::from_secs(10);
    let deadline = tokio::time::Instant::now() + WINDOW;
    let mut trades = 0usize;
    let mut non_trade_events = 0usize;
    let mut quiet_windows = 0usize;

    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        match tokio::time::timeout(remaining.min(SLICE), stream.next()).await {
            Err(_) => quiet_windows += 1,
            Ok(None) => panic!(
                "Stream ended inside the window (trades={trades}, non-trade events={non_trade_events}, \
                 quiet windows={quiet_windows})"
            ),
            Ok(Some(Event::Item(trade))) => {
                trades += 1;
                tracing::info!(?trade, "Received crypto trade");
                assert_eq!(
                    trade.instrument.base.as_ref(),
                    "btc",
                    "Trade routed to an instrument that was never subscribed"
                );
                assert!(trade.kind.price > Decimal::ZERO, "Invalid trade price");
                assert!(trade.kind.amount > Decimal::ZERO, "Invalid trade amount");
            }
            Ok(Some(_)) => non_trade_events += 1,
        }
    }

    // Deliberately not asserted: whether BTC/USD traded is market activity. Logged so a run that
    // saw nothing is still legible, and so the first-trade delay can be re-measured.
    tracing::info!(
        trades,
        non_trade_events,
        quiet_windows,
        window = ?WINDOW,
        "crypto trade window closed"
    );
}

#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_quote_stream_connection() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to crypto quote stream: {:?}",
        streams.err()
    );
    tracing::info!("Crypto quote stream connected and subscribed");
}

#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_quote_stream_receives_data() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await
        .expect("Failed to init crypto quote stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    // Same shape as the trade test above: a quiet window must not end the run early. Quotes tick
    // continuously and independently of trade flow, so 60 seconds is ample here.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
    let mut non_quote_events = 0usize;

    while tokio::time::Instant::now() < deadline {
        match tokio::time::timeout(Duration::from_secs(30), stream.next()).await {
            Err(_) => continue,
            Ok(None) => panic!("Stream ended without data (non-quote events={non_quote_events})"),
            Ok(Some(Event::Item(quote))) => {
                tracing::info!(?quote, "Received crypto quote");
                assert!(quote.kind.bid_price > Decimal::ZERO, "Invalid bid price");
                assert!(quote.kind.ask_price > Decimal::ZERO, "Invalid ask price");
                assert!(
                    quote.kind.ask_price >= quote.kind.bid_price,
                    "Ask < Bid (crossed market)"
                );
                return;
            }
            Ok(Some(_)) => non_quote_events += 1,
        }
    }

    panic!("No crypto quotes received within 60s (non-quote events={non_quote_events})");
}

/// Verifies that two instruments subscribe over a single connection, and that any event which
/// does arrive routes to one of them.
///
/// ### Why this no longer waits for both symbols to tick
///
/// It used to: it required a quote for BTC/USD *and* ETH/USD within 60s, and failed about a
/// quarter of CI runs. The subscription was never at fault. On every failing run Alpaca was sent
/// `{"action":"subscribe","quotes":["BTC/USD","ETH/USD"]}` and confirmed
/// `quotes: ["BTC/USD", "ETH/USD"]`, and no event was dropped, misrouted or failed to
/// deserialise -- the second symbol simply had not ticked yet.
///
/// Alpaca's crypto feed publishes a quote when top-of-book changes, and the delay before a given
/// symbol *first* ticks is both large and highly variable. A single 300s window over four symbols
/// confirmed on one connection measured first-quote delays of 1s (BTC), 15s (LTC), 96s (SOL) and
/// 132s (ETH); ETH then delivered 76 quotes in the remaining 168s. No deadline distinguishes "not
/// subscribed" from "not yet ticked", so waiting on both symbols asserted market activity rather
/// than library behaviour.
///
/// What carries the subscription assertion instead is
/// [`AlpacaSubscriber`]'s subscribe:
/// `init()` now succeeds only once Alpaca has confirmed every requested symbol *by name*, so the
/// `streams.is_ok()` assertion below is the multi-instrument check this test previously
/// approximated by waiting for data -- and it is deterministic.
///
/// The window that follows is a bonus: it cannot require data, but any event it does see must
/// belong to a subscribed instrument, which is what catches misrouting.
#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_multiple_symbols() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [
                (
                    AlpacaCrypto::default(),
                    "btc",
                    "usd",
                    MarketDataInstrumentKind::Spot,
                    Quotes,
                ),
                (
                    AlpacaCrypto::default(),
                    "eth",
                    "usd",
                    MarketDataInstrumentKind::Spot,
                    Quotes,
                ),
            ],
        )
        .init()
        .await;

    // Alpaca confirms the symbols it holds, and the subscriber holds `init()` open until every
    // requested one is named. Reaching here therefore means both were subscribed -- a partial
    // subscription fails above, reporting which symbol was missing.
    assert!(
        streams.is_ok(),
        "Failed to subscribe to multiple crypto symbols: {:?}",
        streams.err()
    );

    let mut stream = streams
        .unwrap()
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    let subscribed = BTreeSet::from(["btc".to_string(), "eth".to_string()]);
    let mut seen = BTreeSet::<String>::new();
    let mut events = 0usize;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);

    while seen.len() < subscribed.len() && tokio::time::Instant::now() < deadline {
        let remaining = deadline - tokio::time::Instant::now();
        match tokio::time::timeout(remaining, stream.next()).await {
            Ok(Some(Event::Item(event))) => {
                events += 1;
                let base = event.instrument.base.as_ref().to_string();
                // A quote can only reach here by matching a SubscriptionId this test registered,
                // so an unknown base means the instrument map routed an event to the wrong
                // instrument -- the one failure a quiet market cannot explain away.
                assert!(
                    subscribed.contains(&base),
                    "Quote routed to an instrument that was never subscribed: \
                     base={base}, subscribed={subscribed:?}"
                );
                seen.insert(base);
            }
            Ok(Some(_)) => {}
            Ok(None) => break,
            Err(_) => break,
        }
    }

    // Deliberately not asserted: which symbols ticked is market activity. Logged so a run that
    // saw nothing is still legible, and so the first-tick delays above can be re-measured.
    tracing::info!(events, ?seen, "crypto multi-symbol quote window closed");
}

/// Trades and quotes from one feed at once, through clones of one subscriber.
///
/// Alpaca allows one connection per feed, and each kind is its own `subscribe`, so this needs the
/// two streams to share a socket: opened on separate sockets, the second `init()` fails with
/// `connection limit exceeded`. Both `init()`s succeeding is therefore the assertion.
///
/// The window that follows cannot require data -- see `test_crypto_multiple_symbols` for how long
/// a crypto symbol can stay quiet -- but any event it does see must be well formed. The shared
/// connection splits frames that mix trades and quotes between the two streams, and a message
/// routed to the wrong one would surface here as an error.
#[tokio::test]
#[ignore]
#[serial]
async fn test_crypto_trades_and_quotes_share_one_connection() {
    init_logging();

    let subscriber = AlpacaSubscriber::from_env().unwrap();

    let trades = Streams::<PublicTrades>::builder()
        .subscribe(
            subscriber.clone(),
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await
        .expect("trades on the crypto feed");

    let quotes = Streams::<Quotes>::builder()
        .subscribe(
            subscriber,
            [(
                AlpacaCrypto::default(),
                "btc",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await
        .expect("quotes on the crypto feed, on the connection the trades already hold");

    let mut trades = trades.select_all();
    let mut quotes = quotes.select_all();

    let (mut trade_events, mut quote_events) = (0usize, 0usize);
    let deadline = tokio::time::sleep(Duration::from_secs(20));
    tokio::pin!(deadline);

    loop {
        tokio::select! {
            () = &mut deadline => break,
            Some(event) = trades.next() => match event {
                Event::Item(Ok(_)) => trade_events += 1,
                Event::Item(Err(error)) => panic!("trade stream error: {error:?}"),
                Event::Reconnecting(exchange) => panic!("trade stream reconnecting: {exchange}"),
            },
            Some(event) = quotes.next() => match event {
                Event::Item(Ok(_)) => quote_events += 1,
                Event::Item(Err(error)) => panic!("quote stream error: {error:?}"),
                Event::Reconnecting(exchange) => panic!("quote stream reconnecting: {exchange}"),
            },
        }
    }

    tracing::info!(
        trade_events,
        quote_events,
        "crypto trades and quotes shared one connection"
    );
}

// ============================================================================
// IEX Equity Stream Tests (market hours only for data reception)
// ============================================================================

#[tokio::test]
#[ignore]
async fn test_iex_trade_stream_connection() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaIex::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to IEX trade stream: {:?}",
        streams.err()
    );
    tracing::info!("IEX trade stream connected and subscribed");
}

#[tokio::test]
#[ignore]
async fn test_iex_quote_stream_connection() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaIex::default(),
                "aapl",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to IEX quote stream: {:?}",
        streams.err()
    );
    tracing::info!("IEX quote stream connected and subscribed");
}

#[tokio::test]
#[ignore]
async fn test_iex_multiple_symbols_subscription() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [
                (
                    AlpacaIex::default(),
                    "spy",
                    "usd",
                    MarketDataInstrumentKind::Spot,
                    PublicTrades,
                ),
                (
                    AlpacaIex::default(),
                    "aapl",
                    "usd",
                    MarketDataInstrumentKind::Spot,
                    PublicTrades,
                ),
            ],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to subscribe to multiple IEX symbols: {:?}",
        streams.err()
    );
    tracing::info!("IEX multi-symbol subscription confirmed");
}

/// Test receiving trade data from IEX.
///
/// NOTE: This test may timeout outside US market hours (9:30 AM - 4:00 PM ET).
/// The IEX feed only provides trade data during market hours.
#[tokio::test]
#[ignore]
async fn test_iex_trade_stream_receives_data() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaIex::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await
        .expect("Failed to init IEX stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    tracing::info!("Waiting for IEX trade data (may timeout outside market hours)...");

    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);

    while tokio::time::Instant::now() < deadline {
        let timeout = tokio::time::timeout(Duration::from_secs(30), stream.next()).await;

        match timeout {
            Ok(Some(Event::Item(trade))) => {
                tracing::info!(?trade, "Received IEX trade");
                assert!(trade.kind.price > Decimal::ZERO, "Invalid trade price");
                assert!(trade.kind.amount > Decimal::ZERO, "Invalid trade amount");
                return;
            }
            Ok(Some(_)) => continue,
            Ok(None) => panic!("Stream ended unexpectedly"),
            Err(_) => {
                tracing::warn!("Timeout waiting for IEX data - may be outside market hours");
                continue;
            }
        }
    }

    panic!(
        "No IEX trade events received within timeout. \
         If outside US market hours (9:30 AM - 4:00 PM ET), this is expected."
    );
}

/// Test receiving quote data from IEX.
///
/// NOTE: This test may timeout outside US market hours.
#[tokio::test]
#[ignore]
async fn test_iex_quote_stream_receives_data() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaIex::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await
        .expect("Failed to init IEX quote stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    tracing::info!("Waiting for IEX quote data (may timeout outside market hours)...");

    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);

    while tokio::time::Instant::now() < deadline {
        let timeout = tokio::time::timeout(Duration::from_secs(30), stream.next()).await;

        match timeout {
            Ok(Some(Event::Item(quote))) => {
                tracing::info!(?quote, "Received IEX quote");
                assert!(quote.kind.bid_price > Decimal::ZERO, "Invalid bid price");
                assert!(quote.kind.ask_price > Decimal::ZERO, "Invalid ask price");
                return;
            }
            Ok(Some(_)) => continue,
            Ok(None) => panic!("Stream ended unexpectedly"),
            Err(_) => {
                tracing::warn!("Timeout waiting for IEX quote - may be outside market hours");
                continue;
            }
        }
    }

    panic!(
        "No IEX quote events received within timeout. \
         If outside US market hours (9:30 AM - 4:00 PM ET), this is expected."
    );
}

// ============================================================================
// SIP Equity Stream Tests (requires SIP subscription)
// ============================================================================

/// Test SIP trade stream connection.
///
/// NOTE: Requires SIP subscription. Will fail with auth error without subscription.
#[tokio::test]
#[ignore]
async fn test_sip_trade_stream_connection() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaSip::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to SIP trade stream: {:?}",
        streams.err()
    );
    tracing::info!("SIP trade stream connected and subscribed");
}

/// Test SIP quote stream connection.
///
/// NOTE: Requires SIP subscription.
#[tokio::test]
#[ignore]
async fn test_sip_quote_stream_connection() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaSip::default(),
                "aapl",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await;

    assert!(
        streams.is_ok(),
        "Failed to connect to SIP quote stream: {:?}",
        streams.err()
    );
    tracing::info!("SIP quote stream connected and subscribed");
}

/// Test receiving trade data from SIP (consolidated tape).
///
/// NOTE: Requires SIP subscription. May timeout outside US market hours.
#[tokio::test]
#[ignore]
async fn test_sip_trade_stream_receives_data() {
    init_logging();

    let streams = Streams::<PublicTrades>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaSip::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                PublicTrades,
            )],
        )
        .init()
        .await
        .expect("Failed to init SIP stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    tracing::info!("Waiting for SIP trade data (consolidated tape)...");

    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);

    while tokio::time::Instant::now() < deadline {
        let timeout = tokio::time::timeout(Duration::from_secs(30), stream.next()).await;

        match timeout {
            Ok(Some(Event::Item(trade))) => {
                tracing::info!(?trade, "Received SIP trade");
                assert!(trade.kind.price > Decimal::ZERO, "Invalid trade price");
                assert!(trade.kind.amount > Decimal::ZERO, "Invalid trade amount");
                return;
            }
            Ok(Some(_)) => continue,
            Ok(None) => panic!("Stream ended unexpectedly"),
            Err(_) => {
                tracing::warn!("Timeout waiting for SIP data - may be outside market hours");
                continue;
            }
        }
    }

    panic!(
        "No SIP trade events received within timeout. \
         If outside US market hours (9:30 AM - 4:00 PM ET), this is expected."
    );
}

/// Test receiving quote data from SIP (consolidated NBBO).
///
/// NOTE: Requires SIP subscription. May timeout outside US market hours.
#[tokio::test]
#[ignore]
async fn test_sip_quote_stream_receives_data() {
    init_logging();

    let streams = Streams::<Quotes>::builder()
        .subscribe(
            AlpacaSubscriber::from_env().unwrap(),
            [(
                AlpacaSip::default(),
                "spy",
                "usd",
                MarketDataInstrumentKind::Spot,
                Quotes,
            )],
        )
        .init()
        .await
        .expect("Failed to init SIP quote stream");

    let mut stream = streams
        .select_all()
        .with_error_handler(|e| tracing::warn!(?e, "Stream error"));

    tracing::info!("Waiting for SIP quote data (consolidated NBBO)...");

    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);

    while tokio::time::Instant::now() < deadline {
        let timeout = tokio::time::timeout(Duration::from_secs(30), stream.next()).await;

        match timeout {
            Ok(Some(Event::Item(quote))) => {
                tracing::info!(?quote, "Received SIP quote (NBBO)");
                assert!(quote.kind.bid_price > Decimal::ZERO, "Invalid bid price");
                assert!(quote.kind.ask_price > Decimal::ZERO, "Invalid ask price");
                return;
            }
            Ok(Some(_)) => continue,
            Ok(None) => panic!("Stream ended unexpectedly"),
            Err(_) => {
                tracing::warn!("Timeout waiting for SIP quote - may be outside market hours");
                continue;
            }
        }
    }

    panic!(
        "No SIP quote events received within timeout. \
         If outside US market hours (9:30 AM - 4:00 PM ET), this is expected."
    );
}

// ============================================================================
// Options Market Data Tests (REST API - available 24/7)
// ============================================================================

use chrono::Utc;
use rustrade_data::exchange::alpaca::options::{
    AlpacaOptionContractQuery, AlpacaOptionFeed, AlpacaOptionsClient,
};

/// Test fetching AAPL option contracts.
///
/// Verifies that we can discover option contracts with various filters.
#[tokio::test]
#[ignore]
async fn test_fetch_aapl_option_contracts() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    // Query AAPL options expiring in the next 60 days
    let today = Utc::now().date_naive();
    let sixty_days = today + chrono::Duration::days(60);

    let query = AlpacaOptionContractQuery::new(vec!["AAPL".into()])
        .expiration_gte(today)
        .expiration_lte(sixty_days)
        .limit(100);

    let contracts = client
        .fetch_contracts(&query)
        .await
        .expect("Failed to fetch contracts");

    tracing::info!(count = contracts.len(), "Fetched AAPL option contracts");

    assert!(!contracts.is_empty(), "No AAPL options found");

    // Verify contract data quality
    for contract in contracts.iter().take(5) {
        tracing::info!(
            symbol = %contract.symbol,
            expiration = %contract.expiration_date,
            strike = %contract.strike_price,
            option_type = %contract.option_type,
            style = %contract.style,
            "Contract"
        );

        assert!(
            contract.symbol.starts_with("AAPL"),
            "Symbol should start with AAPL"
        );
        assert!(
            contract.expiration_date >= today,
            "Expiration should be >= today"
        );
        assert!(
            contract.expiration_date <= sixty_days,
            "Expiration should be <= 60 days out"
        );
        assert!(
            contract.strike_price > Decimal::ZERO,
            "Strike should be positive"
        );
        assert!(
            contract.option_type == "call" || contract.option_type == "put",
            "Option type should be call or put"
        );
        assert!(
            contract.style == "american" || contract.style == "european",
            "Style should be american or european"
        );
    }
}

/// Test fetching option contracts with call-only filter.
#[tokio::test]
#[ignore]
async fn test_fetch_call_options_only() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    let today = Utc::now().date_naive();
    let thirty_days = today + chrono::Duration::days(30);

    let query = AlpacaOptionContractQuery::new(vec!["AAPL".into()])
        .expiration_gte(today)
        .expiration_lte(thirty_days)
        .call_only()
        .limit(50);

    let contracts = client
        .fetch_contracts(&query)
        .await
        .expect("Failed to fetch call options");

    tracing::info!(count = contracts.len(), "Fetched AAPL call options");

    assert!(!contracts.is_empty(), "No AAPL call options found");

    // Verify all are calls
    for contract in &contracts {
        assert_eq!(
            contract.option_type, "call",
            "Expected call option, got {}",
            contract.option_type
        );
    }
}

/// Test fetching option chain snapshots with Greeks.
///
/// Uses the indicative (delayed) feed which is free.
#[tokio::test]
#[ignore]
async fn test_fetch_aapl_chain_snapshot_with_greeks() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    // First get some contracts
    let today = Utc::now().date_naive();
    let thirty_days = today + chrono::Duration::days(30);

    // `limit` is the PAGE SIZE, not a cap on results: `fetch_contracts` pages until the
    // chain is exhausted. A page size of 10 walks the whole 30-day AAPL chain ten contracts
    // at a time and trips the client's 100-page runaway guard with
    // `PaginationLimitExceeded { pages: 101, limit: 100 }`. Use the same page size as
    // `test_fetch_aapl_option_contracts`, which covers a wider window without tripping it.
    let query = AlpacaOptionContractQuery::new(vec!["AAPL".into()])
        .expiration_gte(today)
        .expiration_lte(thirty_days)
        .limit(100);

    let contracts = client
        .fetch_contracts(&query)
        .await
        .expect("Failed to fetch contracts");

    assert!(!contracts.is_empty(), "No contracts to fetch snapshots for");

    // Snapshot only a small slice. The snapshot endpoint takes an explicit symbol list and
    // the full 30-day AAPL chain runs to thousands of contracts, which is both needlessly
    // slow and far more than this test needs to verify Greeks come back populated.
    let symbols: Vec<String> = contracts
        .iter()
        .take(10)
        .map(|c| c.symbol.clone())
        .collect();

    let snapshots = client
        .fetch_snapshots(&symbols, AlpacaOptionFeed::Indicative)
        .await
        .expect("Failed to fetch snapshots");

    tracing::info!(
        requested = symbols.len(),
        received = snapshots.len(),
        "Fetched option snapshots"
    );

    // We might not get snapshots for all symbols (some may not have recent activity)
    // but we should get at least some
    assert!(!snapshots.is_empty(), "No snapshots received");

    // Log snapshot details
    let mut greeks_count = 0;
    for snapshot in &snapshots {
        if snapshot.has_greeks() {
            greeks_count += 1;
            let greeks = snapshot.greeks();
            tracing::info!(
                symbol = %snapshot.symbol,
                delta = ?greeks.delta,
                gamma = ?greeks.gamma,
                theta = ?greeks.theta,
                vega = ?greeks.vega,
                iv = ?greeks.implied_volatility,
                "Snapshot with Greeks"
            );
        }

        if let Some(ref quote) = snapshot.latest_quote {
            tracing::info!(
                symbol = %snapshot.symbol,
                bid = %quote.bid_price,
                ask = %quote.ask_price,
                "Snapshot quote"
            );
            assert!(
                quote.bid_price >= Decimal::ZERO,
                "Bid should be non-negative"
            );
            assert!(
                quote.ask_price >= Decimal::ZERO,
                "Ask should be non-negative"
            );
        }
    }

    tracing::info!(
        total = snapshots.len(),
        with_greeks = greeks_count,
        "Snapshot summary"
    );
}

/// Test the convenience method for fetching entire chain snapshots.
#[tokio::test]
#[ignore]
async fn test_fetch_chain_snapshots_convenience() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    // SPY has a large active option chain (50k+ contracts) — a good stress test
    // for pagination and the 100-symbol batch limit.
    let snapshots = client
        .fetch_chain_snapshots("SPY", AlpacaOptionFeed::Indicative)
        .await
        .expect("Failed to fetch chain snapshots");

    tracing::info!(count = snapshots.len(), "Fetched SPY chain snapshots");

    // SPY has many options, we should get a substantial number
    assert!(!snapshots.is_empty(), "No SPY chain snapshots received");
}

// ============================================================================
// OPRA Options Tests (requires OPRA subscription)
// ============================================================================

/// Test fetching option snapshots with real-time OPRA feed.
///
/// NOTE: Requires OPRA subscription. Will return empty or error without subscription.
#[tokio::test]
#[ignore]
async fn test_fetch_opra_snapshots() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    // Get some AAPL contracts
    let today = Utc::now().date_naive();
    let thirty_days = today + chrono::Duration::days(30);

    let query = AlpacaOptionContractQuery::new(vec!["AAPL".into()])
        .expiration_gte(today)
        .expiration_lte(thirty_days)
        .limit(10);

    let contracts = client
        .fetch_contracts(&query)
        .await
        .expect("Failed to fetch contracts");

    assert!(!contracts.is_empty(), "No contracts to fetch snapshots for");

    let symbols: Vec<String> = contracts.iter().map(|c| c.symbol.clone()).collect();

    // Fetch with OPRA (real-time) feed
    let snapshots = client
        .fetch_snapshots(&symbols, AlpacaOptionFeed::Opra)
        .await
        .expect("Failed to fetch OPRA snapshots");

    tracing::info!(
        requested = symbols.len(),
        received = snapshots.len(),
        "Fetched OPRA option snapshots"
    );

    assert!(!snapshots.is_empty(), "No OPRA snapshots received");

    // Verify we got real-time data with Greeks
    let mut greeks_count = 0;
    for snapshot in &snapshots {
        if snapshot.has_greeks() {
            greeks_count += 1;
            let greeks = snapshot.greeks();
            tracing::info!(
                symbol = %snapshot.symbol,
                delta = ?greeks.delta,
                gamma = ?greeks.gamma,
                theta = ?greeks.theta,
                vega = ?greeks.vega,
                iv = ?greeks.implied_volatility,
                "OPRA snapshot with Greeks"
            );
        }
    }

    tracing::info!(
        total = snapshots.len(),
        with_greeks = greeks_count,
        "OPRA snapshot summary"
    );
}

/// Test fetching entire chain with OPRA feed.
///
/// NOTE: Requires OPRA subscription.
#[tokio::test]
#[ignore]
async fn test_fetch_opra_chain_snapshots() {
    init_logging();

    let client = AlpacaOptionsClient::from_env().expect("Failed to create options client");

    let snapshots = client
        .fetch_chain_snapshots("AAPL", AlpacaOptionFeed::Opra)
        .await
        .expect("Failed to fetch OPRA chain snapshots");

    tracing::info!(count = snapshots.len(), "Fetched AAPL OPRA chain snapshots");

    assert!(
        !snapshots.is_empty(),
        "No AAPL OPRA chain snapshots received"
    );
}