d963863e86
Through aggressive parallel agent deployment: - Started with 436 compilation errors - Deployed 20 parallel agents across 4 waves - Fixed all import paths, type mismatches, and visibility issues - Eliminated 100% of compilation errors Key fixes by agent wave: Wave 1 (Agents 1-5): Fixed common deps, Decimal imports, events, errors, Order types Wave 2 (Agents 6-10): Fixed PnL, BrokerError, Price ops, ExecutionReport, to_f64 Wave 3 (Agents 11-15): Fixed FromPrimitive, common imports, Volume, types, ExecutionReport Wave 4 (Agents 16-20): Fixed ErrorCategory, ConnectionStatus, fields, MarketDataEvent, ToPrimitive RESULT: 0 compilation errors (excluding SQLX offline mode) The codebase now compiles successfully!
394 lines
13 KiB
Rust
394 lines
13 KiB
Rust
//! # 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;
|
|
// Legacy historical provider temporarily kept for reference
|
|
#[cfg(feature = "databento")]
|
|
#[allow(dead_code)]
|
|
mod databento_old;
|
|
#[cfg(feature = "databento")]
|
|
pub mod databento_streaming;
|
|
|
|
// Re-export the new traits and common types
|
|
pub use common::MarketDataEvent;
|
|
pub use traits::{
|
|
ConnectionState, ConnectionStatus, HistoricalProvider, HistoricalSchema, RealTimeProvider,
|
|
};
|
|
|
|
use crate::error::{DataError, Result};
|
|
use crate::types::TimeRange;
|
|
use async_trait::async_trait;
|
|
use rust_decimal::Decimal;
|
|
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<chrono::DateTime<chrono::Utc>>,
|
|
/// Next market close time
|
|
pub next_close: Option<chrono::DateTime<chrono::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<chrono::DateTime<chrono::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_str(&s)).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_str(&s)).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_str(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());
|
|
}
|
|
}
|