foxhunt/data/src/providers/traits.rs

//! # Provider Traits
//!
//! This module defines the core traits for market data providers in the Foxhunt HFT system.
//!
//! ## Architecture
//!
//! The system uses a dual-provider architecture:
//! - **Databento**: Market microstructure data (trades, quotes, L2/L3 order books)
//! - **Benzinga Pro**: News, sentiment, analyst ratings, unusual options activity
//!
//! ## Design Principles
//!
//! - **Separation of Concerns**: Real-time streaming vs historical batch retrieval
//! - **Performance Focus**: Zero-copy parsing, minimal allocations for HFT latency
//! - **Provider Agnostic**: Common event types across different data sources
//! - **Type Safety**: Compile-time schema validation via enums

use chrono::{DateTime, Duration as ChronoDuration, Utc};
use crate::error::Result;
use crate::types::TimeRange;
use ::common::MarketDataEvent;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use futures_core::Stream;
use std::pin::Pin;
use ::common::Symbol;

/// Real-time streaming data provider trait for WebSocket/TCP feeds
///
/// This trait focuses exclusively on real-time, streaming data with minimal latency.
/// Implementers should prioritize zero-copy parsing and efficient memory management.
///
/// # Example Implementation Flow
///
/// ```no_run
/// # use async_trait::async_trait;
/// # use common::Symbol;
/// # use tokio_stream::Stream;
/// # struct MyProvider;
/// # impl MyProvider {
/// #     async fn connect(&mut self) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
/// #     async fn disconnect(&mut self) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
/// #     async fn subscribe(&mut self, symbols: Vec<Symbol>) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
/// #     async fn unsubscribe(&mut self, symbols: Vec<Symbol>) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
/// #     async fn stream(&mut self) -> Result<Box<dyn Stream<Item = String> + Unpin + Send>, Box<dyn std::error::Error>> { todo!() }
/// # }
///
/// // 1. Connect to the data feed
/// provider.connect().await?;
///
/// // 2. Subscribe to symbols of interest
/// provider.subscribe(vec![Symbol::from("SPY"), Symbol::from("QQQ")]).await?;
///
/// // 3. Process the real-time stream
/// let mut stream = provider.stream().await?;
/// while let Some(event) = stream.next().await {
///     // Process market data event with minimal latency
/// }
/// ```
#[async_trait]
pub trait RealTimeProvider: Send + Sync {
    /// Establish connection to the real-time data feed
    ///
    /// This should handle:
    /// - WebSocket/TCP connection establishment
    /// - Authentication with API keys
    /// - Initial protocol handshake
    /// - Connection pooling if supported
    ///
    /// # Errors
    ///
    /// Returns `DataError::Connection` if the connection cannot be established.
    async fn connect(&mut self) -> Result<()>;

    /// Close the connection gracefully
    ///
    /// This should:
    /// - Send proper disconnect messages
    /// - Clean up connection resources
    /// - Cancel any pending subscriptions
    /// - Close WebSocket/TCP connections
    async fn disconnect(&mut self) -> Result<()>;

    /// Subscribe to real-time data for the specified symbols
    ///
    /// # Arguments
    ///
    /// * `symbols` - List of symbols to subscribe to (e.g., "SPY", "AAPL")
    ///
    /// # Provider-Specific Behavior
    ///
    /// - **Databento**: Subscribes to MBO, trades, quotes for given symbols
    /// - **Benzinga**: Subscribes to news alerts, sentiment updates for given symbols
    ///
    /// # Errors
    ///
    /// Returns `DataError::Subscription` if subscription fails or symbols are invalid.
    async fn subscribe(&mut self, symbols: Vec<Symbol>) -> Result<()>;

    /// Unsubscribe from real-time data for the specified symbols
    ///
    /// This allows for dynamic subscription management during runtime.
    async fn unsubscribe(&mut self, symbols: Vec<Symbol>) -> Result<()>;

    /// Returns an async stream of market data events
    ///
    /// This is the core method for real-time data consumption. The stream should:
    /// - Yield events as quickly as possible (sub-millisecond for HFT)
    /// - Use zero-copy parsing where possible
    /// - Handle reconnections transparently
    /// - Emit connection status events on failures
    ///
    /// # Performance Notes
    ///
    /// Implementers should:
    /// - Use `bytes::Bytes` for zero-copy message parsing
    /// - Avoid unnecessary allocations in the hot path
    /// - Consider using `Arc<T>` for shared data structures
    /// - Implement proper backpressure handling
    ///
    /// # Returns
    ///
    /// A pinned boxed stream that yields `MarketDataEvent` items. The stream should be:
    /// - `Send` for use across task boundaries
    /// - Robust to network interruptions
    /// - Properly handle backpressure
    async fn stream(&mut self) -> Result<Pin<Box<dyn Stream<Item = MarketDataEvent> + Send>>>;

    /// Get the current connection status and health metrics
    ///
    /// This provides insight into:
    /// - Connection state (connected, disconnected, reconnecting)
    /// - Message throughput (events per second)
    /// - Latency measurements
    /// - Error counts and rates
    fn get_connection_status(&self) -> ConnectionStatus;

    /// Get the provider's name for identification and logging
    fn get_provider_name(&self) -> &'static str;
}

/// Historical data provider trait for batch data retrieval
///
/// This trait handles point-in-time historical data requests. Unlike real-time providers,
/// this focuses on bulk data retrieval with different performance characteristics.
///
/// # Example Usage
///
/// ```no_run
/// # use chrono::{DateTime, Utc};
/// # use common::Symbol;
/// # struct MyHistoricalProvider;
/// # impl MyHistoricalProvider {
/// #     async fn fetch(&self, symbol: &Symbol, schema: HistoricalSchema, range: TimeRange) -> Result<Vec<String>, Box<dyn std::error::Error>> { Ok(vec![]) }
/// # }
/// # struct TimeRange { start: DateTime<Utc>, end: DateTime<Utc> }
/// # enum HistoricalSchema { Trade }
///
/// let range = TimeRange {
///     start: Utc::now() - Duration::days(1),
///     end: Utc::now(),
/// };
///
/// let trades = provider.fetch(&Symbol::from("SPY"), HistoricalSchema::Trade, range).await?;
/// ```
#[async_trait]
pub trait HistoricalProvider: Send + Sync {
    /// Fetch historical market data for a single symbol
    ///
    /// This method retrieves historical data in batches, with the provider determining
    /// optimal chunking and rate limiting internally.
    ///
    /// # Arguments
    ///
    /// * `symbol` - The symbol to fetch data for
    /// * `schema` - The type of data to retrieve (trades, quotes, etc.)
    /// * `range` - Time range for the historical data
    ///
    /// # Provider-Specific Behavior
    ///
    /// - **Databento**: Returns tick-level data from REST API with pay-as-you-go pricing
    /// - **Benzinga**: May not support historical data for all schemas (news archives limited)
    ///
    /// # Rate Limiting
    ///
    /// Implementers should handle rate limits internally and use exponential backoff
    /// for throttling scenarios.
    ///
    /// # Errors
    ///
    /// - `DataError::Unsupported` if the schema is not supported by this provider
    /// - `DataError::RateLimit` if requests are being throttled
    /// - `DataError::InvalidRange` if the time range is invalid or too large
    async fn fetch(
        &self,
        symbol: &Symbol,
        schema: HistoricalSchema,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>>;

    /// Fetch historical data for multiple symbols in a single request
    ///
    /// This can be more efficient than individual `fetch` calls due to batch processing
    /// and reduced API overhead.
    async fn fetch_batch(
        &self,
        symbols: &[Symbol],
        schema: HistoricalSchema,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>> {
        // Default implementation uses individual fetches
        let mut all_events = Vec::new();
        for symbol in symbols {
            let mut events = self.fetch(symbol, schema, range).await?;
            all_events.append(&mut events);
        }
        // Sort by timestamp for proper ordering
        all_events.sort_by_key(|event| event.timestamp());
        Ok(all_events)
    }

    /// Check if a historical schema is supported by this provider
    ///
    /// This allows callers to check capability before making requests.
    fn supports_schema(&self, schema: HistoricalSchema) -> bool;

    /// Get the maximum time range supported in a single request
    ///
    /// Providers may have limits on how much data can be retrieved at once.
    fn max_range(&self) -> Duration;

    /// Get the provider's name for identification and logging
    fn get_provider_name(&self) -> &'static str;
}

/// Schema types for historical data requests
///
/// This enum restricts the types of historical data that can be requested,
/// providing compile-time safety and clear capability boundaries.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum HistoricalSchema {
    /// Individual trade executions
    ///
    /// Available from: Databento
    Trade,

    /// Bid/ask quote updates
    ///
    /// Available from: Databento
    Quote,

    /// Level 2 order book snapshots and updates
    ///
    /// Available from: Databento (MBO, MBP-1, MBP-10)
    OrderBookL2,

    /// Level 3 order book with individual orders
    ///
    /// Available from: Databento (MBO)
    OrderBookL3,

    /// OHLCV aggregate data (bars)
    ///
    /// Available from: Databento
    OHLCV,

    /// News articles and alerts
    ///
    /// Available from: Benzinga (limited historical archives)
    News,

    /// Sentiment analysis scores
    ///
    /// Available from: Benzinga (limited historical data)
    Sentiment,

    /// Analyst ratings and upgrades/downgrades
    ///
    /// Available from: Benzinga
    AnalystRating,

    /// Unusual options activity
    ///
    /// Available from: Benzinga
    UnusualOptions,
}

impl HistoricalSchema {
    /// Check if this schema represents market microstructure data
    pub fn is_market_data(&self) -> bool {
        matches!(
            self,
            HistoricalSchema::Trade
                | HistoricalSchema::Quote
                | HistoricalSchema::OrderBookL2
                | HistoricalSchema::OrderBookL3
                | HistoricalSchema::OHLCV
        )
    }

    /// Check if this schema represents news/sentiment data
    pub fn is_news_data(&self) -> bool {
        matches!(
            self,
            HistoricalSchema::News
                | HistoricalSchema::Sentiment
                | HistoricalSchema::AnalystRating
                | HistoricalSchema::UnusualOptions
        )
    }

    /// Get the typical provider for this schema
    pub fn typical_provider(&self) -> &'static str {
        match self {
            HistoricalSchema::Trade
            | HistoricalSchema::Quote
            | HistoricalSchema::OrderBookL2
            | HistoricalSchema::OrderBookL3
            | HistoricalSchema::OHLCV => "databento",
            HistoricalSchema::News
            | HistoricalSchema::Sentiment
            | HistoricalSchema::AnalystRating
            | HistoricalSchema::UnusualOptions => "benzinga",
        }
    }
}

/// Connection status for real-time providers
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionStatus {
    /// Current connection state
    pub state: ConnectionState,

    /// Number of active subscriptions
    pub active_subscriptions: usize,

    /// Events received per second (rolling average)
    pub events_per_second: f64,

    /// Current latency in microseconds (if measurable)
    pub latency_micros: Option<u64>,

    /// Error count in the last hour
    pub recent_error_count: u32,

    /// Last successful message timestamp
    pub last_message_time: Option<DateTime<Utc>>,

    /// Last connection attempt timestamp
    pub last_connection_attempt: Option<DateTime<Utc>>,
}

/// Connection state enumeration
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConnectionState {
    /// Not connected
    Disconnected,

    /// In the process of connecting
    Connecting,

    /// Successfully connected and receiving data
    Connected,

    /// Attempting to reconnect after a failure
    Reconnecting,

    /// Connection failed and not attempting to reconnect
    Failed,
}

impl Default for ConnectionStatus {
    fn default() -> Self {
        Self {
            state: ConnectionState::Disconnected,
            active_subscriptions: 0,
            events_per_second: 0.0,
            latency_micros: None,
            recent_error_count: 0,
            last_message_time: None,
            last_connection_attempt: None,
        }
    }
}

impl ConnectionStatus {
    /// Create a new disconnected status
    pub fn disconnected() -> Self {
        Self::default()
    }

    /// Create a connected status with basic metrics
    pub fn connected() -> Self {
        Self {
            state: ConnectionState::Connected,
            last_connection_attempt: Some(Utc::now()),
            ..Default::default()
        }
    }

    /// Check if the connection is healthy
    pub fn is_healthy(&self) -> bool {
        matches!(self.state, ConnectionState::Connected)
            && self.recent_error_count < 10
            && self
                .last_message_time
                .map(|t| Utc::now().signed_duration_since(t).num_seconds() < 30)
                .unwrap_or(false)
    }
}

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

    #[test]
    fn test_historical_schema_categorization() {
        assert!(HistoricalSchema::Trade.is_market_data());
        assert!(!HistoricalSchema::Trade.is_news_data());

        assert!(HistoricalSchema::News.is_news_data());
        assert!(!HistoricalSchema::News.is_market_data());

        assert_eq!(HistoricalSchema::Trade.typical_provider(), "databento");
        assert_eq!(HistoricalSchema::News.typical_provider(), "benzinga");
    }

    #[test]
    fn test_connection_status() {
        let status = ConnectionStatus::connected();
        assert_eq!(status.state, ConnectionState::Connected);

        let disconnected = ConnectionStatus::disconnected();
        assert_eq!(disconnected.state, ConnectionState::Disconnected);
        assert!(!disconnected.is_healthy());
    }

    #[test]
    fn test_historical_schema_serialization() {
        let schema = HistoricalSchema::OrderBookL2;
        let json = serde_json::to_string(&schema).unwrap();
        let deserialized: HistoricalSchema = serde_json::from_str(&json).unwrap();
        assert_eq!(schema, deserialized);
    }
}