foxhunt/data/src/lib.rs

//! # Foxhunt Data Module
//!
//! High-performance market data ingestion and broker integration module for HFT systems,
//! including Interactive Brokers, ICMarkets, and other data providers.
//!
//! ## Features
//!
//! - **High-Frequency Data Ingestion**: Sub-millisecond market data processing
//! - **Multiple Data Providers**: Interactive Brokers, ICMarkets
//! - **Real-time WebSocket Streams**: Level 1 and Level 2 market data
//! - **FIX Protocol Support**: Complete FIX 4.4 implementation for broker connectivity
//! - **Order Management**: Order lifecycle management with execution reports
//! - **Resilient Connectivity**: Automatic reconnection with exponential backoff
//! - **Performance Optimized**: Zero-copy parsing and lock-free data structures
//!
//! ## Architecture
//!
//! ```text
//! ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
//! │   Market Data   │    │   Broker Conn   │    │   Data Store    │
//! │   Providers     │───▶│   Management    │───▶│   & Cache       │
//! └─────────────────┘    └─────────────────┘    └─────────────────┘
//!          │                       │                       │
//!          ▼                       ▼                       ▼
//! ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
//! │   WebSocket     │    │   FIX Protocol  │    │   Event Stream  │
//! │   Streams       │    │   Engine        │    │   Publisher     │
//! └─────────────────┘    └─────────────────┘    └─────────────────┘
//! ```
//!
//! ## Core Components
//!
//! ### Market Data Providers
//! - **Databento**: Real-time and historical market data
//! - **ICMarkets**: FIX protocol integration for forex and CFDs
//! - **Interactive Brokers**: TWS API integration for equities and derivatives
//!
//! ### FIX Protocol Engine
//! - Complete FIX 4.4 implementation
//! - Session management with heartbeat
//! - Order lifecycle management
//! - Execution report processing
//! - Message parsing with zero-copy optimization
//!
//! ### Performance Features
//! - Lock-free data structures for high-throughput scenarios
//! - Memory pools for allocation efficiency
//! - SIMD-optimized message parsing
//! - CPU affinity for deterministic latency
//!
//! ## Usage
//!
//! ```rust
//! // NOTE: Broker clients moved to core module in monolithic architecture
//! use foxhunt_core::brokers::{
//!     ICMarketsClient, ICMarketsConfig, FixOrder
//! };
//! use foxhunt_core::prelude::{OrderSide, OrderType, ExecutionReport};
//! // REMOVED: Polygon client - replaced with Databento
//! use data::types::{MarketDataEvent, Subscription};
//!
//! #[tokio::main]
//! async fn main() -> anyhow::Result<()> {
//!     // Initialize ICMarkets FIX client
//!     let config = ICMarketsConfig {
//!         host: "fix-demo.icmarkets.com".to_string(),
//!         port: 9880,
//!         sender_comp_id: "CLIENT".to_string(),
//!         target_comp_id: "ICMARKETS".to_string(),
//!         username: "your_username".to_string(),
//!         password: "your_password".to_string(),
//!         heartbeat_interval: 30,
//!         ..Default::default()
//!     };
//!     
//!     let mut client = ICMarketsClient::new(config);
//!     client.connect().await?;
//!     
//!     // Submit an order
//!     let order = FixOrder {
//!         cl_ord_id: "ORDER001".to_string(),
//!         symbol: "EURUSD".to_string(),
//!         side: OrderSide::Buy,
//!         ord_type: OrderType::Market,
//!         order_qty: 10000.0,
//!         price: None,
//!         stop_px: None,
//!         time_in_force: 1, // GTC
//!         account: None,
//!     };
//!     
//!     let order_id = client.submit_order(order).await?;
//!     println!("Order submitted: {}", order_id);
//!     
//!     Ok(())
//! }
//! ```
//!
//! ## Configuration
//!
//! The module supports environment-based configuration:
//!
//! ```toml
//! [data]
//! // REMOVED: polygon_api_key configuration
//! ib_host = "${IB_TWS_HOST:-127.0.0.1}"
//! ib_port = "${IB_TWS_PORT:-7497}"
//!
//! # ICMarkets configuration
//! icmarkets = {
//!     host = "${ICMARKETS_HOST:-fix-demo.icmarkets.com}",
//!     port = "${ICMARKETS_PORT:-9880}",
//!     username = "${ICMARKETS_USERNAME}",
//!     password = "${ICMARKETS_PASSWORD}"
//! }
//! ```

#![warn(
    missing_docs,
    rust_2018_idioms,
    unused_qualifications,
    clippy::cognitive_complexity,
    clippy::large_enum_variant,
    clippy::type_complexity
)]
#![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

pub mod brokers;
// pub mod config; // Temporarily disabled - complex fixes needed
pub mod error;
pub mod parquet_persistence; // Parquet market data persistence for replay
pub mod features; // Feature engineering for ML models
pub mod providers; // Data providers (Databento, Benzinga)
pub mod storage;
pub mod training_pipeline; // Training data pipeline for ML models
pub mod types;
pub mod unified_feature_extractor; // Unified feature extraction across systems
pub mod utils;
pub mod validation; // Data validation and quality control

#[cfg(test)]
mod storage_test;
mod storage_standalone_test;

// #[cfg(test)]
// REMOVED: polygon test module

// Tracing macros
use tracing::{error, info, warn};

// Re-export commonly used types - broker clients moved to core module
// pub use brokers::{...}; // Broker clients now in core module
// Databento and Benzinga providers
pub use crate::providers::databento::{DatabentoHistoricalProvider, DatabentoConfig};
pub use crate::providers::benzinga::{BenzingaHistoricalProvider, BenzingaConfig, NewsEvent};
pub use crate::unified_feature_extractor::{UnifiedFeatureExtractor, UnifiedFeatureExtractorConfig};
pub use crate::types::{MarketDataEvent, Subscription, TradeEvent};
pub use error::{DataError, Result};
pub use foxhunt_core::prelude::OrderSide;
pub use foxhunt_core::types::events::OrderEvent;
pub use foxhunt_core::types::OrderType;
use tokio::sync::broadcast;

/// Data module configuration - Simplified version with disabled modules
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct DataConfig {
    // REMOVED: Polygon configuration - replaced with Databento

    /// Interactive Brokers configuration
    pub interactive_brokers: Option<brokers::IBConfig>,

    // ICMarkets configuration moved to core module
    /// General data settings
    pub settings: DataSettings,
}

/// General data module settings
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct DataSettings {
    /// Maximum reconnection attempts for any provider
    pub max_reconnect_attempts: u32,

    /// Base reconnection delay in milliseconds
    pub base_reconnect_delay: u64,

    /// Maximum reconnection delay in milliseconds
    pub max_reconnect_delay: u64,

    /// Buffer size for market data events
    pub market_data_buffer_size: usize,

    /// Buffer size for order update events
    pub order_event_buffer_size: usize,

    /// Enable performance monitoring
    pub enable_metrics: bool,
}

impl Default for DataConfig {
    fn default() -> Self {
        Self {
            // REMOVED: polygon: None,
            interactive_brokers: None,
            // icmarkets configuration moved to core module
            settings: DataSettings {
                max_reconnect_attempts: 10,
                base_reconnect_delay: 1000,
                max_reconnect_delay: 60000,
                market_data_buffer_size: 10000,
                order_event_buffer_size: 10000,
                enable_metrics: true,
            },
        }
    }
}

impl DataConfig {
    /// Load configuration from environment variables
    pub fn from_env() -> Result<Self> {
        let mut config = Self::default();

        // REMOVED: Polygon configuration loading

        // Load Interactive Brokers configuration
        if std::env::var("IB_TWS_HOST").is_ok() {
            config.interactive_brokers = Some(brokers::IBConfig::default());
        }

        // ICMarkets configuration moved to core module

        Ok(config)
    }
}

/// Initialize the data module with configuration
pub async fn initialize(config: DataConfig) -> Result<DataManager> {
    DataManager::new(config).await
}

/// Main data manager for coordinating all data providers and brokers
pub struct DataManager {
    config: DataConfig,
    // REMOVED: polygon_client: Option<crate::polygon::PolygonClient>,
    ib_client: Option<brokers::InteractiveBrokersAdapter>,
    // icmarkets_client moved to core module
    market_data_broadcast_tx: broadcast::Sender<MarketDataEvent>,
    order_update_broadcast_tx: broadcast::Sender<OrderEvent>,
}

impl DataManager {
    /// Create a new data manager
    pub async fn new(config: DataConfig) -> Result<Self> {
        // Initialize broadcast channels with buffer sizes from config
        let (market_data_broadcast_tx, _market_data_broadcast_rx) =
            broadcast::channel(config.settings.market_data_buffer_size);
        let (order_update_broadcast_tx, _order_update_broadcast_rx) =
            broadcast::channel::<OrderEvent>(
                config.settings.order_event_buffer_size,
            );

        // REMOVED: Polygon client initialization

        let ib_client = if let Some(ib_config) = &config.interactive_brokers {
            Some(brokers::InteractiveBrokersAdapter::new(
                ib_config.clone(),
            ))
        } else {
            None
        };

        // icmarkets_client initialization moved to core module

        Ok(Self {
            config,
            // REMOVED: polygon_client,
            ib_client,
            // icmarkets_client field removed
            market_data_broadcast_tx,
            order_update_broadcast_tx,
        })
    }

    // ICMarkets client methods moved to core module

    /// Start all configured data providers and brokers
    pub async fn start(&mut self) -> Result<()> {
        // ICMarkets connection handling moved to core module

        // REMOVED: Polygon client startup code

        // Start Interactive Brokers client if configured
        if let Some(client) = &mut self.ib_client {
            info!("Starting Interactive Brokers connection");

            match client.connect().await {
                Ok(()) => {
                    info!("Interactive Brokers connection established successfully");

                    // Note: IB execution subscription would be implemented here when the
                    // subscribe_executions method is available in InteractiveBrokersAdapter
                    info!("Interactive Brokers connection ready - execution subscription not yet implemented");
                }
                Err(e) => {
                    error!("Failed to connect to Interactive Brokers: {}", e);
                    warn!("Continuing startup without Interactive Brokers connection");
                }
            }
        }

        Ok(())
    }

    /// Stop all data providers and brokers
    pub async fn stop(&mut self) -> Result<()> {
        // ICMarkets disconnect handling moved to core module

        // Stop other clients as needed...

        Ok(())
    }

    /// Subscribe to market data events
    pub fn subscribe_market_data_events(&self) -> broadcast::Receiver<MarketDataEvent> {
        self.market_data_broadcast_tx.subscribe()
    }

    /// Subscribe to order update events
    pub fn subscribe_order_update_events(
        &self,
    ) -> broadcast::Receiver<OrderEvent> {
        self.order_update_broadcast_tx.subscribe()
    }

    /// Subscribe to market data for specific symbols and data types
    pub async fn subscribe_market_data(&self, subscription: Subscription) -> Result<()> {
        // REMOVED: Polygon subscription code

        // Subscribe with other clients as needed...
        // ICMarkets doesn't typically handle market data subscriptions in the same way

        info!(
            "Market data subscription request received for: {:?}",
            subscription.symbols
        );
        Ok(())
    }
}

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

    #[test]
    fn test_config_default() {
        let config = DataConfig::default();
        assert!(config.interactive_brokers.is_none()); // Default has no IB config
        assert_eq!(config.settings.max_reconnect_attempts, 10);
    }

    #[tokio::test]
    async fn test_data_manager_creation() {
        let config = DataConfig::default();
        let data_manager = DataManager::new(config).await;
        assert!(data_manager.is_ok());
    }
}