//! Database connection utilities and configurations //! //! This module provides shared database connection management utilities //! that can be used across all Foxhunt services. use serde::{Deserialize, Serialize}; use sqlx::{Pool, Postgres}; use std::time::Duration; use thiserror::Error; // Import centralized database configuration pub use config::database::DatabaseConfig; use config::structures::BacktestingDatabaseConfig; /// Database-specific errors #[derive(Debug, Error)] #[allow(clippy::module_name_repetitions)] pub enum DatabaseError { /// Connection failed - wrapper around SQLx connection errors #[error("Connection failed: {0}")] Connection(#[from] sqlx::Error), /// Query exceeded maximum allowed execution time #[error("Query timeout: operation took {actual_ms}ms, max allowed {max_ms}ms")] QueryTimeout { /// Actual execution time in milliseconds actual_ms: u64, /// Maximum allowed execution time in milliseconds max_ms: u64, }, /// Connection pool has no available connections #[error("Pool exhausted: no connections available")] PoolExhausted, /// Database configuration is invalid or missing required parameters #[error("Configuration error: {0}")] Configuration(String), /// Performance constraint violation detected #[error("Performance violation: {0}")] Performance(String), } /// Database connection configuration (local extended version) #[derive(Debug, Clone, Deserialize, Serialize)] pub struct LocalDatabaseConfig { /// Database connection URL pub url: String, /// Pool configuration pub pool: PoolConfig, /// Performance settings pub performance: PerformanceConfig, } /// Connection pool configuration #[derive(Debug, Clone, Deserialize, Serialize)] pub struct PoolConfig { /// Maximum number of connections in the pool pub max_connections: u32, /// Minimum number of connections to maintain pub min_connections: u32, /// Connection timeout in milliseconds pub connect_timeout_ms: u64, /// Connection acquire timeout in milliseconds pub acquire_timeout_ms: u64, /// Maximum connection lifetime in seconds pub max_lifetime_seconds: u64, /// Idle timeout in seconds pub idle_timeout_seconds: u64, } /// Performance configuration for HFT operations #[derive(Debug, Clone, Deserialize, Serialize)] pub struct PerformanceConfig { /// Query timeout in microseconds for HFT operations pub query_timeout_micros: u64, /// Enable connection prewarming pub enable_prewarming: bool, /// Enable statement preparation pub enable_prepared_statements: bool, /// Enable query logging for slow queries pub enable_slow_query_logging: bool, /// Slow query threshold in microseconds pub slow_query_threshold_micros: u64, } impl Default for LocalDatabaseConfig { fn default() -> Self { Self { url: "postgresql://foxhunt:password@localhost:5432/foxhunt".to_owned(), pool: PoolConfig::default(), performance: PerformanceConfig::default(), } } } impl Default for PoolConfig { fn default() -> Self { Self { max_connections: 50, min_connections: 10, connect_timeout_ms: 5000, acquire_timeout_ms: 5000, max_lifetime_seconds: 3600, idle_timeout_seconds: 300, } } } impl Default for PerformanceConfig { fn default() -> Self { Self { query_timeout_micros: 800, // <1ms for HFT operations enable_prewarming: true, enable_prepared_statements: true, enable_slow_query_logging: true, slow_query_threshold_micros: 1000, // Log queries >1ms } } } /// Convert from centralized config to common crate config with HFT optimizations #[allow(clippy::integer_division)] impl From for LocalDatabaseConfig { fn from(config: DatabaseConfig) -> Self { Self { url: config.url, pool: PoolConfig { max_connections: config.max_connections, // 20% of max, min 2. Uses integer division intentionally for simplicity. min_connections: (config.max_connections / 5).max(2), connect_timeout_ms: u64::try_from(config.connect_timeout.as_millis()) .unwrap_or(5000), // Pool connect timeout (not query timeout) acquire_timeout_ms: 5000, // Pool acquire timeout (not query timeout) max_lifetime_seconds: 3600, // 1 hour default idle_timeout_seconds: 300, // 5 minutes default }, performance: PerformanceConfig { query_timeout_micros: u64::try_from(config.query_timeout.as_micros().min(800)) .unwrap_or(800), // Convert to microseconds, cap at 800μs for HFT enable_prewarming: true, enable_prepared_statements: true, enable_slow_query_logging: config.enable_query_logging, slow_query_threshold_micros: 1000, // 1ms threshold }, } } } /// Convert from backtesting config to common crate config with backtesting optimizations #[allow(clippy::integer_division)] impl From for LocalDatabaseConfig { fn from(config: BacktestingDatabaseConfig) -> Self { let max_conn = config.max_connections.unwrap_or(10); Self { url: config.database_url, pool: PoolConfig { max_connections: max_conn, // 25% of max, min 2. Uses integer division intentionally for simplicity. min_connections: (max_conn / 4).max(2), connect_timeout_ms: config.acquire_timeout_ms.unwrap_or(1000), // Use acquire timeout as connection timeout acquire_timeout_ms: 5000, // Pool acquire timeout (not query timeout) max_lifetime_seconds: 3600, // 1 hour default idle_timeout_seconds: 600, // 10 minutes for backtesting }, performance: PerformanceConfig { query_timeout_micros: 10000, // 10ms default for backtesting queries enable_prewarming: true, enable_prepared_statements: true, enable_slow_query_logging: config.enable_logging.unwrap_or(false), slow_query_threshold_micros: 5000, // 5ms threshold for backtesting }, } } } /// Database connection pool wrapper #[derive(Debug, Clone)] #[allow(clippy::module_name_repetitions)] pub struct DatabasePool { pool: Pool, config: LocalDatabaseConfig, } impl DatabasePool { /// Create a new database connection pool /// /// # Errors /// /// Returns `DatabaseError` if: /// - Connection URL is invalid /// - Database connection fails pub async fn new(config: LocalDatabaseConfig) -> Result { use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; // Parse connection options let mut connect_options: PgConnectOptions = config .url .parse() .map_err(|e| DatabaseError::Configuration(format!("Invalid URL: {}", e)))?; // Configure connection-level optimizations connect_options = connect_options .application_name("foxhunt-service") .statement_cache_capacity(1000); // Create connection pool with optimized settings let pool = PgPoolOptions::new() .max_connections(config.pool.max_connections) .min_connections(config.pool.min_connections) .acquire_timeout(Duration::from_millis(config.pool.acquire_timeout_ms)) .max_lifetime(Duration::from_secs(config.pool.max_lifetime_seconds)) .idle_timeout(Duration::from_secs(config.pool.idle_timeout_seconds)) .test_before_acquire(true) .connect_with(connect_options) .await .map_err(DatabaseError::Connection)?; // Pre-warm connections if enabled if config.performance.enable_prewarming { for _ in 0..config.pool.min_connections { let _conn = pool.acquire().await.map_err(DatabaseError::Connection)?; sqlx::query("SELECT 1") .fetch_one(&pool) .await .map_err(DatabaseError::Connection)?; } } Ok(Self { pool, config }) } /// Get the underlying connection pool pub const fn pool(&self) -> &Pool { &self.pool } /// Get current configuration pub const fn config(&self) -> &LocalDatabaseConfig { &self.config } /// Health check for the database connection /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database connection fails /// - Query times out pub async fn health_check(&self) -> Result<(), DatabaseError> { let result = tokio::time::timeout( Duration::from_millis(100), sqlx::query("SELECT 1").fetch_one(&self.pool), ) .await; match result { Ok(Ok(_)) => Ok(()), Ok(Err(e)) => Err(DatabaseError::Connection(e)), Err(_) => Err(DatabaseError::QueryTimeout { actual_ms: 100, max_ms: 100, }), } } /// Get connection pool statistics #[allow(clippy::integer_division)] pub fn pool_stats(&self) -> PoolStats { PoolStats { size: self.pool.size(), idle: u32::try_from(self.pool.num_idle()).unwrap_or(0), active: self .pool .size() .saturating_sub(u32::try_from(self.pool.num_idle()).unwrap_or(0)), max_size: self.config.pool.max_connections, } } } /// Connection pool statistics #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PoolStats { /// Current pool size pub size: u32, /// Number of idle connections pub idle: u32, /// Number of active connections pub active: u32, /// Maximum pool size pub max_size: u32, } impl PoolStats { /// Calculate pool utilization percentage #[allow(clippy::float_arithmetic)] pub fn utilization_percentage(&self) -> f64 { (f64::from(self.active) / f64::from(self.max_size)) * 100.0 } /// Check if pool is healthy (not over-utilized) pub fn is_healthy(&self) -> bool { self.utilization_percentage() < 80.0 } } // ============================================================================ // Wave D: Regime Tracking Database Helpers // ============================================================================ /// Regime state record from database #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RegimeState { pub symbol: String, pub regime: String, pub confidence: f64, pub event_timestamp: chrono::DateTime, pub cusum_s_plus: Option, pub cusum_s_minus: Option, pub adx: Option, pub stability: Option, } /// Regime transition record from database #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RegimeTransition { pub symbol: String, pub from_regime: String, pub to_regime: String, pub event_timestamp: chrono::DateTime, pub duration_bars: Option, pub transition_probability: Option, } /// Adaptive strategy metrics record from database #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AdaptiveStrategyMetrics { pub symbol: String, pub regime: String, pub event_timestamp: chrono::DateTime, pub position_multiplier: f64, pub stop_loss_multiplier: f64, pub regime_sharpe: Option, pub risk_budget_utilization: Option, pub total_trades: i32, pub winning_trades: i32, pub total_pnl: i64, } impl DatabasePool { /// Get the latest regime state for a symbol /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database query fails /// - No regime state found for symbol pub async fn get_latest_regime(&self, symbol: &str) -> Result { let record = sqlx::query!( r#" SELECT regime, confidence, event_timestamp, cusum_s_plus, cusum_s_minus, adx, stability FROM get_latest_regime($1) "#, symbol ) .fetch_one(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(RegimeState { symbol: symbol.to_string(), regime: record.regime.unwrap_or_else(|| "Normal".to_string()), confidence: record.confidence.unwrap_or(0.0), event_timestamp: record.event_timestamp.unwrap_or_else(chrono::Utc::now), cusum_s_plus: record.cusum_s_plus, cusum_s_minus: record.cusum_s_minus, adx: record.adx, stability: record.stability, }) } /// Insert a new regime state /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database insert fails /// - Constraint violation (duplicate timestamp) #[allow(clippy::too_many_arguments)] pub async fn insert_regime_state( &self, symbol: &str, regime: &str, confidence: f64, event_timestamp: chrono::DateTime, cusum_s_plus: Option, cusum_s_minus: Option, adx: Option, stability: Option, ) -> Result<(), DatabaseError> { sqlx::query!( r#" INSERT INTO regime_states ( symbol, regime, confidence, event_timestamp, cusum_s_plus, cusum_s_minus, adx, stability ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (symbol, event_timestamp) DO UPDATE SET regime = EXCLUDED.regime, confidence = EXCLUDED.confidence, cusum_s_plus = EXCLUDED.cusum_s_plus, cusum_s_minus = EXCLUDED.cusum_s_minus, adx = EXCLUDED.adx, stability = EXCLUDED.stability "#, symbol, regime, confidence, event_timestamp, cusum_s_plus, cusum_s_minus, adx, stability ) .execute(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(()) } /// Insert a regime transition /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database insert fails /// - Invalid transition (from_regime == to_regime) #[allow(clippy::too_many_arguments)] pub async fn insert_regime_transition( &self, symbol: &str, from_regime: &str, to_regime: &str, event_timestamp: chrono::DateTime, duration_bars: Option, transition_probability: Option, adx_at_transition: Option, cusum_alert_triggered: bool, ) -> Result<(), DatabaseError> { sqlx::query!( r#" INSERT INTO regime_transitions ( symbol, from_regime, to_regime, event_timestamp, duration_bars, transition_probability, adx_at_transition, cusum_alert_triggered ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) "#, symbol, from_regime, to_regime, event_timestamp, duration_bars, transition_probability, adx_at_transition, cusum_alert_triggered ) .execute(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(()) } /// Get regime transitions for a symbol /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database query fails pub async fn get_regime_transitions( &self, symbol: &str, limit: i32, ) -> Result, DatabaseError> { let records = sqlx::query_as!( RegimeTransition, r#" SELECT symbol, from_regime, to_regime, event_timestamp, duration_bars, transition_probability FROM regime_transitions WHERE symbol = $1 ORDER BY event_timestamp DESC LIMIT $2 "#, symbol, limit as i64 ) .fetch_all(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(records) } /// Insert or update adaptive strategy metrics /// /// # Errors /// /// Returns `DatabaseError` if: /// - Database insert/update fails #[allow(clippy::too_many_arguments)] pub async fn upsert_adaptive_strategy_metrics( &self, symbol: &str, regime: &str, event_timestamp: chrono::DateTime, position_multiplier: f64, stop_loss_multiplier: f64, regime_sharpe: Option, risk_budget_utilization: Option, total_trades: i32, winning_trades: i32, total_pnl: i64, ) -> Result<(), DatabaseError> { sqlx::query!( r#" INSERT INTO adaptive_strategy_metrics ( symbol, regime, event_timestamp, position_multiplier, stop_loss_multiplier, regime_sharpe, risk_budget_utilization, total_trades, winning_trades, total_pnl ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (symbol, event_timestamp, regime) DO UPDATE SET position_multiplier = EXCLUDED.position_multiplier, stop_loss_multiplier = EXCLUDED.stop_loss_multiplier, regime_sharpe = EXCLUDED.regime_sharpe, risk_budget_utilization = EXCLUDED.risk_budget_utilization, total_trades = adaptive_strategy_metrics.total_trades + EXCLUDED.total_trades, winning_trades = adaptive_strategy_metrics.winning_trades + EXCLUDED.winning_trades, total_pnl = adaptive_strategy_metrics.total_pnl + EXCLUDED.total_pnl "#, symbol, regime, event_timestamp, position_multiplier, stop_loss_multiplier, regime_sharpe, risk_budget_utilization, total_trades, winning_trades, total_pnl ) .execute(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(()) } /// Get regime performance metrics for a symbol /// /// # Errors /// /// Returns `DatabaseError` if database query fails pub async fn get_regime_performance( &self, symbol: Option<&str>, window_hours: i32, ) -> Result, DatabaseError> { let records = sqlx::query_as!( RegimePerformance, r#" SELECT regime, total_trades, win_rate, avg_sharpe, avg_position_multiplier, avg_stop_loss_multiplier, total_pnl as "total_pnl: rust_decimal::Decimal", avg_risk_utilization FROM get_regime_performance($1, $2) "#, symbol, window_hours ) .fetch_all(&self.pool) .await .map_err(DatabaseError::Connection)?; Ok(records) } } /// Regime performance metrics from database #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RegimePerformance { pub regime: Option, pub total_trades: Option, pub win_rate: Option, pub avg_sharpe: Option, pub avg_position_multiplier: Option, pub avg_stop_loss_multiplier: Option, pub total_pnl: Option, pub avg_risk_utilization: Option, }