//! # 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 core::brokers::{ //! ICMarketsClient, ICMarketsConfig, FixOrder //! }; //! use 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 features; // Feature engineering for ML models // pub mod parquet_persistence; // Parquet market data persistence for replay - TEMPORARILY DISABLED due to arrow compatibility issue 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; // #[cfg(test)] // REMOVED: polygon test module // Tracing macros use tracing::{error, info, warn}; // Re-export commonly used types for clean API // === Core Error Handling === pub use error::{DataError, Result}; // === Broker Integration === // Note: Broker clients moved to trading_engine module in monolithic architecture pub use brokers::{ BrokerClient, BrokerConfig, BrokerResult, BrokerAdapter, BrokerType, BrokerFactory, InteractiveBrokersAdapter, IBConfig }; // === Data Providers === // Databento provider - only available when feature is enabled #[cfg(feature = "databento")] pub use crate::providers::databento::{ DatabentoConfig, DatabentoHistoricalProvider }; // Benzinga provider pub use crate::providers::benzinga::{ BenzingaConfig, BenzingaHistoricalProvider, NewsEvent }; // Provider traits and common types pub use crate::providers::{ RealTimeProvider, HistoricalProvider, MarketDataProvider, ProviderConfig, ProviderFactory, ProviderManager, MarketStatus, ProviderHealthStatus }; // Common event types from providers pub use crate::providers::common::{ MarketDataEvent as CommonMarketDataEvent, TradeEvent as CommonTradeEvent, QuoteEvent as CommonQuoteEvent, NewsEvent as CommonNewsEvent, OrderBookSnapshot, OrderBookUpdate, ConnectionState }; // === Data Types === pub use crate::types::{ MarketDataEvent, Subscription, TradeEvent, QuoteEvent, TimeRange, MarketDataType, Aggregate, Level2Update, Position, Account, ConnectionEvent, ErrorEvent }; // === Feature Engineering === pub use crate::unified_feature_extractor::{ UnifiedFeatureExtractor, UnifiedFeatureExtractorConfig }; // === Storage and Persistence === pub use crate::storage::{ StorageManager, EnhancedDatasetMetadata, StorageStats, ExportFormat }; // === Validation === pub use crate::validation::{ DataValidator, ValidationResult, ValidationError }; // === Training Pipeline === pub use crate::training_pipeline::{ TrainingDataPipeline, FeatureProcessor, TechnicalIndicatorsCalculator, MicrostructureAnalyzer, TLOBProcessor, RegimeDetector, DatasetMetadata, DatasetSchema, FeatureColumn, TargetColumn }; // === Utilities === pub use crate::utils::{ format_timestamp, calculate_percentage_change, normalize_symbol }; // === External Re-exports === // Commonly used external types use tokio::sync::broadcast; // Import canonical types from trading_engine prelude per TYPE_GOVERNANCE.md use common::prelude::*; use common::types::OrderEvent; // Add missing OrderEvent import // Import shared configuration from foxhunt-config-crate use config::{DataModuleConfig, DataModuleSettings}; // Using direct imports from foxhunt-config-crate - NO backward compatibility aliases // Data module configuration moved to foxhunt-config-crate shared library // Use: config::DataModuleConfig and config::DataModuleSettings // DataModuleConfig implementation moved to foxhunt-config-crate shared library /// Initialize the data module with configuration pub async fn initialize(config: DataModuleConfig) -> Result { DataManager::new(config).await } /// Main data manager for coordinating all data providers and brokers pub struct DataManager { config: DataModuleConfig, // REMOVED: polygon_client: Option, ib_client: Option, // icmarkets_client moved to core module market_data_broadcast_tx: broadcast::Sender, order_update_broadcast_tx: broadcast::Sender, } impl DataManager { /// Create a new data manager pub async fn new(config: DataModuleConfig) -> Result { // 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::(config.settings.order_event_buffer_size); // REMOVED: Polygon client initialization let ib_client = if let Some(ib_config) = &config.interactive_brokers { let broker_config = brokers::IBConfig { host: ib_config.host.clone(), port: ib_config.port, client_id: ib_config.client_id as i32, account_id: std::env::var("IB_ACCOUNT_ID").unwrap_or_else(|_| "DU123456".to_string()), connection_timeout: ib_config.timeout_seconds, heartbeat_interval: 30, max_reconnect_attempts: 5, request_timeout: 30, }; Some(brokers::InteractiveBrokersAdapter::new(broker_config)) } 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 { self.market_data_broadcast_tx.subscribe() } /// Subscribe to order update events pub fn subscribe_order_update_events(&self) -> broadcast::Receiver { 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 = DataModuleConfig::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 = DataModuleConfig::default(); let data_manager = DataManager::new(config).await; assert!(data_manager.is_ok()); } }