cc4e0c5a2d
Re-export HardwareTimestamp through data crate instead of ml/risk depending directly on trading_engine. Reduces coupling between the ML pipeline and the trading engine. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
378 lines
17 KiB
Rust
378 lines
17 KiB
Rust
#![allow(unused_extern_crates)]
|
|
#![allow(unused_crate_dependencies)]
|
|
#![allow(dead_code)] // Data providers and processors defined but not yet integrated from services
|
|
#![allow(unsafe_code)] // Intentional unsafe for high-performance data processing
|
|
#![allow(missing_docs)] // Internal implementation details don't require documentation
|
|
#![allow(missing_debug_implementations)] // Not all types need Debug
|
|
#![allow(clippy::float_arithmetic)]
|
|
// Data processing requires float arithmetic
|
|
// Performance-critical HFT code - pedantic lints that would require major refactoring
|
|
#![allow(clippy::str_to_string)] // High-performance string conversions
|
|
#![allow(clippy::as_conversions)] // Low-level data parsing requires type conversions
|
|
#![allow(clippy::default_numeric_fallback)] // Protocol parsing with numeric literals
|
|
#![allow(clippy::arithmetic_side_effects)] // Market data calculations are performance-critical
|
|
#![allow(clippy::indexing_slicing)] // Protocol buffer parsing requires indexing
|
|
#![allow(clippy::doc_markdown)] // API names in docs (ICMarkets, InteractiveBrokers, etc.)
|
|
#![allow(clippy::module_name_repetitions)] // Broker/provider naming conventions
|
|
#![allow(clippy::map_err_ignore)] // Error context not always needed in data pipeline
|
|
#![allow(clippy::result_large_err)] // Error types sized for comprehensive error reporting
|
|
#![allow(clippy::missing_const_for_fn)] // Runtime dynamic behavior in many functions
|
|
#![deny(clippy::unwrap_used, clippy::expect_used)]
|
|
#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]
|
|
#![allow(clippy::clone_on_ref_ptr)] // Arc/Rc clones required for concurrent access
|
|
#![allow(clippy::integer_division)] // Price calculations require division
|
|
#![allow(clippy::wildcard_enum_match_arm)] // Future-proof protocol extensions
|
|
#![allow(clippy::similar_names)] // Domain-specific names (side/size, etc.)
|
|
#![allow(clippy::else_if_without_else)] // State machine logic doesn't always need else
|
|
#![allow(clippy::single_char_lifetime_names)] // Standard Rust lifetime conventions
|
|
#![allow(clippy::useless_conversion)] // Type system conversions for API compatibility
|
|
#![allow(clippy::used_underscore_binding)] // Protocol field parsing with underscore prefix
|
|
#![allow(clippy::if_then_some_else_none)] // Conditional logic readability
|
|
#![allow(clippy::cognitive_complexity)] // Complex protocol state machines
|
|
#![allow(clippy::shadow_reuse)] // Variable shadowing in nested scopes
|
|
#![allow(clippy::let_underscore_must_use)] // Intentionally discarding results in data pipeline
|
|
#![allow(clippy::unnecessary_wraps)] // Consistent Result types across trait implementations
|
|
#![allow(clippy::wildcard_imports)] // Internal module imports
|
|
#![allow(clippy::shadow_unrelated)] // Shadowing for readability
|
|
#![allow(clippy::non_ascii_literal)] // Currency symbols and market data
|
|
#![allow(clippy::unwrap_or_default)] // Performance optimizations
|
|
#![allow(clippy::unwrap_in_result)] // Verified safe unwraps in error paths
|
|
#![allow(clippy::string_slice)] // Protocol parsing requires string slicing
|
|
#![allow(clippy::redundant_closure)] // Explicit closures for clarity
|
|
#![allow(clippy::new_without_default)] // Builder pattern implementations
|
|
#![allow(clippy::upper_case_acronyms)] // API/FIX protocol acronyms
|
|
#![allow(clippy::type_complexity)] // Complex generic types in broker traits
|
|
#![allow(clippy::same_name_method)] // Trait and inherent method coexistence
|
|
#![allow(clippy::string_to_string)] // String conversions in protocol parsing
|
|
#![allow(clippy::rc_buffer)] // Rc<Vec> for shared buffer access
|
|
#![allow(clippy::infinite_loop)] // Event loop implementations
|
|
#![allow(clippy::unnecessary_cast)] // Explicit casts for protocol compatibility
|
|
#![allow(clippy::too_many_lines)] // Complex protocol handlers
|
|
#![allow(clippy::ptr_arg)] // Vec arguments for builder patterns
|
|
#![allow(clippy::needless_range_loop)] // Manual indexing for performance
|
|
#![allow(clippy::clone_on_copy)] // Explicit clone calls for clarity
|
|
#![allow(clippy::unnecessary_sort_by)] // Custom comparison logic
|
|
#![allow(clippy::unnecessary_map_or)] // Explicit mapping for readability
|
|
#![allow(clippy::manual_range_contains)] // Explicit range checks
|
|
#![allow(clippy::undocumented_unsafe_blocks)] // Unsafe blocks documented inline
|
|
#![allow(clippy::unchecked_duration_subtraction)] // Duration arithmetic validated by logic
|
|
#![allow(clippy::single_match_else)] // Explicit match for readability
|
|
#![allow(clippy::shadow_same)] // Shadowing for progressive refinement
|
|
#![allow(clippy::reserve_after_initialization)] // Dynamic capacity management
|
|
#![allow(clippy::redundant_clone)] // Clone required for ownership transfer
|
|
#![allow(clippy::modulo_arithmetic)] // Modulo for circular buffer indexing
|
|
#![allow(clippy::manual_ok_err)] // Explicit Ok/Err construction
|
|
#![allow(clippy::manual_clamp)] // Custom clamping logic
|
|
#![allow(clippy::len_zero)] // Explicit len() == 0 checks
|
|
#![allow(clippy::into_iter_on_ref)] // Explicit iterator creation
|
|
#![allow(clippy::impl_trait_in_params)] // Trait object parameters
|
|
#![allow(clippy::explicit_counter_loop)] // Manual loop counters for control
|
|
#![allow(clippy::doc_lazy_continuation)] // Documentation formatting
|
|
#![allow(clippy::await_holding_lock)] // Lock scope carefully managed
|
|
#![allow(clippy::assign_op_pattern)] // Explicit assignment patterns
|
|
#![allow(clippy::manual_let_else)] // if-let pattern preferred in streaming error paths
|
|
|
|
//! # 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(rust_2018_idioms, unused_qualifications, clippy::large_enum_variant)]
|
|
// Note: cognitive_complexity and type_complexity are allowed at crate-level for HFT protocol code
|
|
// dead_code lint intentionally enabled — compiler should flag unused items
|
|
// Note: Deprecated fields are kept for backwards compatibility but usage updated
|
|
#![allow(unexpected_cfgs)] // Allow unexpected cfg attributes
|
|
#![allow(private_bounds)] // Allow private type bounds
|
|
#![allow(unreachable_pub)] // Allow unreachable public items
|
|
// Note: unwrap/expect/panic lints are handled at crate-level above for HFT performance code
|
|
|
|
pub mod brokers;
|
|
// pub mod config; // Temporarily disabled - complex fixes needed
|
|
pub mod dbn_uploader; // DBN file uploader to MinIO
|
|
pub mod error;
|
|
pub mod features; // Feature engineering for ML models
|
|
pub mod parquet_persistence; // Parquet market data persistence for replay
|
|
pub mod providers; // Data providers (Databento, Benzinga)
|
|
pub mod replay; // Real market data replay infrastructure for tests/benchmarks
|
|
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
|
|
|
|
// Temporarily disabled due to compilation errors
|
|
// #[cfg(test)]
|
|
// mod storage_test;
|
|
|
|
// #[cfg(test)]
|
|
// REMOVED: polygon test module
|
|
|
|
// Tracing macros
|
|
use tracing::{error, info, warn};
|
|
|
|
// === External Re-exports ===
|
|
// Re-export HardwareTimestamp so downstream crates (ml) don't need a direct trading_engine dep
|
|
pub use trading_engine::timing::HardwareTimestamp;
|
|
// Commonly used external types
|
|
use tokio::sync::broadcast;
|
|
// Import configuration and event types that are actually used
|
|
use config::data_config::DataModuleConfig;
|
|
// OrderEvent type is not currently used in data module - removed import
|
|
use crate::brokers::{IBConfig, InteractiveBrokersAdapter};
|
|
use crate::error::Result;
|
|
use ::common::{MarketDataEvent, Subscription};
|
|
|
|
// Using direct imports from common crate - NO backward compatibility aliases
|
|
|
|
/// Initialize the data module with configuration
|
|
pub async fn initialize(config: DataModuleConfig) -> Result<DataManager> {
|
|
DataManager::new(config).await
|
|
}
|
|
|
|
/// Main data manager for coordinating all data providers and brokers
|
|
pub struct DataManager {
|
|
config: DataModuleConfig,
|
|
// REMOVED: polygon_client: Option<crate::polygon::PolygonClient>,
|
|
ib_client: Option<InteractiveBrokersAdapter>,
|
|
// icmarkets_client moved to core module
|
|
market_data_broadcast_tx: broadcast::Sender<MarketDataEvent>,
|
|
order_update_broadcast_tx: broadcast::Sender<MarketDataEvent>,
|
|
}
|
|
|
|
impl DataManager {
|
|
/// Create a new data manager
|
|
pub async fn new(config: DataModuleConfig) -> 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::<MarketDataEvent>(config.settings.order_event_buffer_size);
|
|
|
|
// REMOVED: Polygon client initialization
|
|
|
|
let ib_client = if let Some(ib_config) = &config.interactive_brokers {
|
|
let broker_config = 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(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<MarketDataEvent> {
|
|
self.market_data_broadcast_tx.subscribe()
|
|
}
|
|
|
|
/// Subscribe to order update events
|
|
pub fn subscribe_order_update_events(&self) -> broadcast::Receiver<MarketDataEvent> {
|
|
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(())
|
|
}
|
|
}
|
|
|
|
// Temporarily disabled due to compilation errors
|
|
// #[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());
|
|
// }
|
|
// }
|