foxhunt/data/src/providers/mod.rs

//! # Market Data Providers Module
//!
//! This module contains implementations for various market data providers in the
//! Foxhunt HFT trading system with a focus on dual-provider architecture.
//!
//! ## Architecture
//!
//! The system uses a dual-provider approach:
//! - **Databento**: Market microstructure data (trades, quotes, L2/L3 order books)
//! - **Benzinga Pro**: News, sentiment, analyst ratings, unusual options activity
//! - **Polygon.io**: Legacy provider (being phased out)
//!
//! ## Provider Traits
//!
//! - `RealTimeProvider`: Streaming WebSocket data with sub-millisecond latency
//! - `HistoricalProvider`: Batch historical data retrieval with rate limiting
//! - `MarketDataProvider`: Legacy unified interface (backwards compatibility)
//!
//! ## Features
//!
//! - Zero-copy message parsing for maximum HFT performance
//! - Unified event types across all providers via `MarketDataEvent`
//! - Automatic reconnection with exponential backoff
//! - Provider-specific error handling and rate limiting
//! - Real-time connection health monitoring

// Core trait definitions and common types
pub mod common;
pub mod traits;

// Provider implementations
pub mod benzinga;

// Databento provider - only available when feature is enabled
#[cfg(feature = "databento")]
pub mod databento;
#[cfg(feature = "databento")]
pub mod databento_streaming;

// Re-export core traits for external use
pub use traits::{
    ConnectionState, ConnectionStatus as TraitConnectionStatus, HistoricalProvider,
    HistoricalSchema, RealTimeProvider,
};

use crate::error::{DataError, Result};
use crate::types::TimeRange;
use ::common::MarketDataEvent;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
// use common::Symbol;

/// Configuration for market data providers
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProviderConfig {
    /// Provider name (polygon, databento, benzinga)
    pub name: String,
    /// API endpoint URL
    pub endpoint: String,
    /// API key or credentials
    pub api_key: String,
    /// Enable real-time data streaming
    pub enable_realtime: bool,
    /// Maximum concurrent connections
    pub max_connections: usize,
    /// Rate limit (requests per second)
    pub rate_limit: u32,
    /// Connection timeout in milliseconds
    pub timeout_ms: u64,
    /// Enable Level 2 data
    pub enable_level2: bool,
    /// Subscription symbols
    pub symbols: Vec<String>,
}

/// Legacy market data provider trait for backwards compatibility
///
/// This trait provides a unified interface for providers that implement both
/// real-time and historical capabilities. New providers should implement
/// `RealTimeProvider` and/or `HistoricalProvider` directly for better
/// separation of concerns.
#[async_trait]
pub trait MarketDataProvider: Send + Sync {
    /// Connect to the data provider
    async fn connect(&mut self) -> Result<()>;

    /// Disconnect from the data provider
    async fn disconnect(&mut self) -> Result<()>;

    /// Subscribe to real-time market data for symbols
    async fn subscribe(&mut self, symbols: Vec<String>) -> Result<()>;

    /// Unsubscribe from symbols
    async fn unsubscribe(&mut self, symbols: Vec<String>) -> Result<()>;

    /// Get historical market data
    async fn get_historical_data(
        &self,
        symbol: &str,
        timeframe: &str,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>>;

    /// Get current market status
    async fn get_market_status(&self) -> Result<MarketStatus>;

    /// Get provider health status
    fn get_health_status(&self) -> ProviderHealthStatus;

    /// Get provider name
    fn get_name(&self) -> &str;
}

/// Market status information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarketStatus {
    /// Market is currently open
    pub is_open: bool,
    /// Next market open time
    pub next_open: Option<DateTime<Utc>>,
    /// Next market close time
    pub next_close: Option<DateTime<Utc>>,
    /// Market timezone
    pub timezone: String,
    /// Extended hours trading available
    pub extended_hours: bool,
}

/// Provider health status
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProviderHealthStatus {
    /// Provider is connected
    pub connected: bool,
    /// Last successful connection time
    pub last_connected: Option<DateTime<Utc>>,
    /// Number of active subscriptions
    pub active_subscriptions: usize,
    /// Messages received per second
    pub messages_per_second: f64,
    /// Connection latency in microseconds
    pub latency_micros: Option<u64>,
    /// Error count in last hour
    pub error_count: u32,
}

/// Provider factory for creating different provider instances
pub struct ProviderFactory;

impl ProviderFactory {
    /// Create a new provider instance based on configuration
    pub fn create_provider(
        config: ProviderConfig,
        _event_tx: mpsc::UnboundedSender<MarketDataEvent>,
    ) -> Result<Box<dyn MarketDataProvider>> {
        match config.name.as_str() {
            "databento" => {
                // Databento streaming provider for real-time data
                Err(DataError::Configuration {
                    field: "provider.name".to_string(),
                    message: "Use DatabentoStreamingProvider for real-time data or DatabentoHistoricalProvider for historical data.".to_string(),
                })
            },
            "benzinga" => {
                // Benzinga news and sentiment provider
                Err(DataError::Configuration {
                    field: "provider.name".to_string(),
                    message: "Use BenzingaProvider for news and sentiment data.".to_string(),
                })
            },
            _ => Err(DataError::Configuration {
                field: "provider.name".to_string(),
                message: format!(
                    "Unknown provider: {}. Available providers: databento, benzinga",
                    config.name
                ),
            }),
        }
    }
}

/// Provider manager for coordinating multiple providers
pub struct ProviderManager {
    providers: Vec<Box<dyn MarketDataProvider>>,
    event_tx: mpsc::UnboundedSender<MarketDataEvent>,
    health_monitor: HealthMonitor,
}

impl ProviderManager {
    /// Create a new provider manager
    pub fn new(event_tx: mpsc::UnboundedSender<MarketDataEvent>) -> Self {
        Self {
            providers: Vec::new(),
            event_tx,
            health_monitor: HealthMonitor::new(),
        }
    }

    /// Add a provider to the manager
    pub fn add_provider(&mut self, provider: Box<dyn MarketDataProvider>) {
        self.providers.push(provider);
    }

    /// Connect all providers
    pub async fn connect_all(&mut self) -> Result<()> {
        for provider in &mut self.providers {
            if let Err(e) = provider.connect().await {
                tracing::error!("Failed to connect provider {}: {}", provider.get_name(), e);
                continue;
            }
            tracing::info!("Connected to provider: {}", provider.get_name());
        }
        Ok(())
    }

    /// Subscribe to symbols across all providers
    pub async fn subscribe_all(&mut self, symbols: Vec<String>) -> Result<()> {
        for provider in &mut self.providers {
            if let Err(e) = provider.subscribe(symbols.clone()).await {
                tracing::error!(
                    "Failed to subscribe on provider {}: {}",
                    provider.get_name(),
                    e
                );
                continue;
            }
        }
        Ok(())
    }

    /// Get health status for all providers
    pub fn get_all_health_status(&self) -> Vec<(String, ProviderHealthStatus)> {
        self.providers
            .iter()
            .map(|p| (p.get_name().to_string(), p.get_health_status()))
            .collect()
    }

    /// Start health monitoring
    pub async fn start_health_monitoring(&mut self) {
        self.health_monitor.start(&self.providers).await;
    }
}

/// Health monitor for tracking provider status
struct HealthMonitor {
    monitoring: bool,
}

impl HealthMonitor {
    fn new() -> Self {
        Self { monitoring: false }
    }

    async fn start(&mut self, _providers: &[Box<dyn MarketDataProvider>]) {
        if self.monitoring {
            return;
        }

        self.monitoring = true;
        tracing::info!("Started provider health monitoring");

        // Health monitoring implementation would go here
        // This would periodically check provider status and emit alerts
    }
}

// Blanket implementation to provide backwards compatibility
// Any type that implements both RealTimeProvider and HistoricalProvider
// automatically implements the legacy MarketDataProvider trait
#[async_trait]
impl<T> MarketDataProvider for T
where
    T: RealTimeProvider + HistoricalProvider,
{
    async fn connect(&mut self) -> Result<()> {
        RealTimeProvider::connect(self).await
    }

    async fn disconnect(&mut self) -> Result<()> {
        RealTimeProvider::disconnect(self).await
    }

    async fn subscribe(&mut self, symbols: Vec<String>) -> Result<()> {
        let symbol_structs: Vec<::common::Symbol> = symbols
            .into_iter()
            .map(|s| ::common::Symbol::from(s.as_str()))
            .collect();
        RealTimeProvider::subscribe(self, symbol_structs).await
    }

    async fn unsubscribe(&mut self, symbols: Vec<String>) -> Result<()> {
        let symbol_structs: Vec<::common::Symbol> = symbols
            .into_iter()
            .map(|s| ::common::Symbol::from(s.as_str()))
            .collect();
        RealTimeProvider::unsubscribe(self, symbol_structs).await
    }

    async fn get_historical_data(
        &self,
        symbol: &str,
        timeframe: &str,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>> {
        // Convert timeframe string to HistoricalSchema
        let schema = match timeframe.to_lowercase().as_str() {
            "trades" | "trade" => HistoricalSchema::Trade,
            "quotes" | "quote" => HistoricalSchema::Quote,
            "orderbook" | "l2" => HistoricalSchema::OrderBookL2,
            "mbo" | "l3" => HistoricalSchema::OrderBookL3,
            "bars" | "ohlcv" | "candles" => HistoricalSchema::OHLCV,
            "news" => HistoricalSchema::News,
            "sentiment" => HistoricalSchema::Sentiment,
            _ => HistoricalSchema::Trade, // Default fallback
        };

        // Convert string to Symbol
        let symbol_struct = ::common::Symbol::from(symbol);
        // Fetch data from the historical provider - already returns common::MarketDataEvent
        let results = HistoricalProvider::fetch(self, &symbol_struct, schema, range).await?;
        // No conversion needed - HistoricalProvider::fetch returns common::MarketDataEvent
        Ok(results)
    }

    async fn get_market_status(&self) -> Result<MarketStatus> {
        // Default implementation - providers can override
        Ok(MarketStatus {
            is_open: true,
            next_open: None,
            next_close: None,
            timezone: "US/Eastern".to_string(),
            extended_hours: false,
        })
    }

    fn get_health_status(&self) -> ProviderHealthStatus {
        let connection_status = RealTimeProvider::get_connection_status(self);
        ProviderHealthStatus {
            connected: matches!(connection_status.state, ConnectionState::Connected),
            last_connected: connection_status.last_connection_attempt,
            active_subscriptions: connection_status.active_subscriptions,
            messages_per_second: connection_status.events_per_second,
            latency_micros: connection_status.latency_micros,
            error_count: connection_status.recent_error_count,
        }
    }

    fn get_name(&self) -> &str {
        RealTimeProvider::get_provider_name(self)
    }
}

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

    #[tokio::test]
    async fn test_provider_manager_creation() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let manager = ProviderManager::new(tx);
        assert_eq!(manager.providers.len(), 0);
    }

    #[test]
    fn test_provider_config_serialization() {
        let config = ProviderConfig {
            name: "databento".to_string(),
            endpoint: "wss://api.databento.com/ws".to_string(),
            api_key: std::env::var("DATABENTO_API_KEY")
                .unwrap_or_else(|_| "DATABENTO_API_KEY_REQUIRED".to_string()),
            enable_realtime: true,
            max_connections: 5,
            rate_limit: 100,
            timeout_ms: 5000,
            enable_level2: true,
            symbols: vec!["SPY".to_string(), "QQQ".to_string()],
        };

        let json = serde_json::to_string(&config).unwrap();
        let deserialized: ProviderConfig = serde_json::from_str(&json).unwrap();
        assert_eq!(config.name, deserialized.name);
    }

    #[test]
    fn test_historical_schema_conversion() {
        use traits::HistoricalSchema;

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