From 4ca9820b7a40eb802e086fa18f5d64849448009f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 21 Feb 2026 21:51:08 +0100 Subject: [PATCH] safety: change clippy allow to deny(unwrap_used, expect_used) in risk and data crates --- data/src/features.rs | 2 + data/src/lib.rs | 3 +- data/src/providers/benzinga/ml_integration.rs | 11 +++++- .../benzinga/production_historical.rs | 16 +++++++- data/src/providers/databento/dbn_parser.rs | 14 +++++-- data/src/providers/databento/mod.rs | 14 +++---- data/src/providers/databento/stream.rs | 2 +- data/src/training_pipeline.rs | 23 +++++------- risk/src/lib.rs | 5 ++- risk/src/position_tracker.rs | 37 +++++++++++++------ 10 files changed, 83 insertions(+), 44 deletions(-) diff --git a/data/src/features.rs b/data/src/features.rs index e4002b0d5..086d3eeca 100644 --- a/data/src/features.rs +++ b/data/src/features.rs @@ -2145,6 +2145,8 @@ impl TemporalFeatures { // Convert UTC to EST (UTC-5) for market hour calculations // Note: This doesn't account for daylight saving time, assumes EST year-round use chrono::FixedOffset; + // SAFETY: 5*3600 = 18000 seconds is always a valid UTC offset + #[allow(clippy::unwrap_used)] let est_offset = FixedOffset::west_opt(5 * 3600).unwrap(); let est_time = timestamp.with_timezone(&est_offset); diff --git a/data/src/lib.rs b/data/src/lib.rs index 537163d24..15093f1ed 100644 --- a/data/src/lib.rs +++ b/data/src/lib.rs @@ -16,7 +16,8 @@ #![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 -#![allow(clippy::unwrap_used)] // Verified safe unwraps in hot paths +#![deny(clippy::unwrap_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 diff --git a/data/src/providers/benzinga/ml_integration.rs b/data/src/providers/benzinga/ml_integration.rs index 259063267..3a25481ae 100644 --- a/data/src/providers/benzinga/ml_integration.rs +++ b/data/src/providers/benzinga/ml_integration.rs @@ -587,8 +587,15 @@ impl BenzingaMLExtractor { // Sentiment momentum (rate of change) let momentum = if relevant_events.len() > 1 { - let first = relevant_events.first().unwrap().sentiment_score; - let last = relevant_events.last().unwrap().sentiment_score; + // len() > 1 guarantees first() and last() are Some + let first = relevant_events + .first() + .map(|e| e.sentiment_score) + .unwrap_or(0.0); + let last = relevant_events + .last() + .map(|e| e.sentiment_score) + .unwrap_or(0.0); last - first } else { 0.0 diff --git a/data/src/providers/benzinga/production_historical.rs b/data/src/providers/benzinga/production_historical.rs index 9bc963dfe..9618b7850 100644 --- a/data/src/providers/benzinga/production_historical.rs +++ b/data/src/providers/benzinga/production_historical.rs @@ -338,7 +338,11 @@ impl ProductionBenzingaHistoricalProvider { })?; // Create rate limiter - let quota = Quota::per_second(NonZeroU32::new(config.rate_limit_per_second).unwrap()); + let quota = Quota::per_second( + NonZeroU32::new(config.rate_limit_per_second).ok_or_else(|| { + DataError::configuration("rate_limit_per_second", "must be > 0") + })?, + ); let rate_limiter = Arc::new(RateLimiter::direct(quota)); // Create semaphore @@ -384,7 +388,13 @@ impl ProductionBenzingaHistoricalProvider { /// Make a rate-limited HTTP request with retries #[instrument(skip(self))] async fn make_request(&self, url: &str, params: &[(String, String)]) -> Result { - let _permit = self.semaphore.acquire().await.unwrap(); + let _permit = self + .semaphore + .acquire() + .await + .map_err(|e| DataError::Network { + message: format!("Semaphore closed: {}", e), + })?; let start_time = Instant::now(); let mut last_error = None; @@ -896,6 +906,8 @@ impl ProductionBenzingaHistoricalProvider { let expiration_date = NaiveDate::parse_from_str(&item.date_expiry, "%Y-%m-%d") .map_err(|e| DataError::parse(format!("Invalid expiration date: {}", e)))?; + // SAFETY: (0, 0, 0) is always a valid HMS time + #[allow(clippy::unwrap_used)] let expiration = expiration_date.and_hms_opt(0, 0, 0).unwrap().and_utc(); let expiry = expiration; // Same as expiration diff --git a/data/src/providers/databento/dbn_parser.rs b/data/src/providers/databento/dbn_parser.rs index c8f079dd9..34f752ab6 100644 --- a/data/src/providers/databento/dbn_parser.rs +++ b/data/src/providers/databento/dbn_parser.rs @@ -238,14 +238,20 @@ impl DbnParser { /// Update symbol mapping for instrument IDs pub fn update_symbol_map(&self, mapping: std::collections::HashMap) { - let mut symbol_map = self.symbol_map.write().unwrap(); + let mut symbol_map = self + .symbol_map + .write() + .unwrap_or_else(|e| e.into_inner()); symbol_map.extend(mapping); debug!("Updated symbol map with {} instruments", symbol_map.len()); } /// Update price scaling factors pub fn update_price_scales(&self, scales: std::collections::HashMap) { - let mut price_scales = self.price_scales.write().unwrap(); + let mut price_scales = self + .price_scales + .write() + .unwrap_or_else(|e| e.into_inner()); price_scales.extend(scales); debug!( "Updated price scales for {} instruments", @@ -531,7 +537,7 @@ impl DbnParser { fn get_symbol(&self, instrument_id: u32) -> String { self.symbol_map .read() - .unwrap() + .unwrap_or_else(|e| e.into_inner()) .get(&instrument_id) .cloned() .unwrap_or_else(|| format!("UNKNOWN_{}", instrument_id)) @@ -542,7 +548,7 @@ impl DbnParser { let scale = self .price_scales .read() - .unwrap() + .unwrap_or_else(|e| e.into_inner()) .get(&instrument_id) .copied() .unwrap_or(4); // Default to 4 decimal places diff --git a/data/src/providers/databento/mod.rs b/data/src/providers/databento/mod.rs index e098bfe46..e63866a3d 100644 --- a/data/src/providers/databento/mod.rs +++ b/data/src/providers/databento/mod.rs @@ -264,20 +264,20 @@ impl RealTimeProvider for DatabentoStreamingProvider { // Update connection status { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); status.state = ConnectionState::Connecting; status.last_connection_attempt = Some(Utc::now()); } match self.client.connect().await { Ok(()) => { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); *status = ConnectionStatus::connected(); info!("Databento streaming provider connected successfully"); Ok(()) }, Err(e) => { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); status.state = ConnectionState::Failed; error!("Failed to connect Databento streaming provider: {}", e); Err(e) @@ -290,7 +290,7 @@ impl RealTimeProvider for DatabentoStreamingProvider { match self.client.shutdown().await { Ok(()) => { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); status.state = ConnectionState::Disconnected; info!("Databento streaming provider disconnected successfully"); Ok(()) @@ -311,7 +311,7 @@ impl RealTimeProvider for DatabentoStreamingProvider { Ok(()) => { // Update connection status with subscription count { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); status.active_subscriptions = symbols.len(); } info!("Successfully subscribed to {} symbols", symbols.len()); @@ -333,7 +333,7 @@ impl RealTimeProvider for DatabentoStreamingProvider { Ok(()) => { // Update connection status { - let mut status = self.connection_status.write().unwrap(); + let mut status = self.connection_status.write().unwrap_or_else(|e| e.into_inner()); status.active_subscriptions = status.active_subscriptions.saturating_sub(symbols.len()); } @@ -354,7 +354,7 @@ impl RealTimeProvider for DatabentoStreamingProvider { } fn get_connection_status(&self) -> ConnectionStatus { - let base_status = self.connection_status.read().unwrap().clone(); + let base_status = self.connection_status.read().unwrap_or_else(|e| e.into_inner()).clone(); let metrics = self.client.get_metrics(); // Enhance with real-time metrics diff --git a/data/src/providers/databento/stream.rs b/data/src/providers/databento/stream.rs index 5df8e2779..3b5f93f04 100644 --- a/data/src/providers/databento/stream.rs +++ b/data/src/providers/databento/stream.rs @@ -858,7 +858,7 @@ impl Stream for DatabentoMarketDataStream { type Item = MarketDataEvent; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // SAFETY: We only access `inner` through a pinned reference. + // We only access `inner` through a pinned reference. // `UnboundedReceiverStream` is `Unpin`, so this projection is safe. let this = self.get_mut(); let result = Pin::new(&mut this.inner).poll_next(cx); diff --git a/data/src/training_pipeline.rs b/data/src/training_pipeline.rs index c5639da60..9778f2ab4 100644 --- a/data/src/training_pipeline.rs +++ b/data/src/training_pipeline.rs @@ -13,7 +13,7 @@ //! - **TLOB-Specific Processing**: Order book reconstruction, imbalance calculations //! - **Portfolio Performance**: P&L tracking, performance attribution, risk metrics -use crate::error::Result; +use crate::error::{DataError, Result}; // REMOVED: Polygon imports - replaced with Databento use chrono::{DateTime, Datelike, Timelike, Utc}; use common::{OrderSide, PriceLevel}; @@ -612,7 +612,9 @@ impl TrainingDataPipeline { _client: Arc, _dataset_id: &str, ) -> Result<()> { - let databento_config = self.config.sources.databento.as_ref().unwrap(); + let databento_config = self.config.sources.databento.as_ref().ok_or_else(|| { + DataError::configuration("sources.databento", "Databento config required") + })?; let _hist_config = &self.config.sources.historical; for symbol in &databento_config.symbols { @@ -634,7 +636,9 @@ impl TrainingDataPipeline { _client: Arc, _dataset_id: &str, ) -> Result<()> { - let benzinga_config = self.config.sources.benzinga.as_ref().unwrap(); + let benzinga_config = self.config.sources.benzinga.as_ref().ok_or_else(|| { + DataError::configuration("sources.benzinga", "Benzinga config required") + })?; let _hist_config = &self.config.sources.historical; for symbol in &benzinga_config.symbols { @@ -908,17 +912,11 @@ impl MicrostructureAnalyzer { if trades.len() >= 2 { let first_price = trades .front() - .unwrap() - .price - .to_string() - .parse::() + .map(|t| t.price.to_string().parse::().unwrap_or(0.0)) .unwrap_or(0.0); let last_price = trades .back() - .unwrap() - .price - .to_string() - .parse::() + .map(|t| t.price.to_string().parse::().unwrap_or(0.0)) .unwrap_or(0.0); let impact = if first_price != 0.0 { (last_price - first_price) / first_price @@ -960,8 +958,7 @@ impl RegimeDetector { .or_insert_with(VecDeque::new); // Calculate simple volatility estimate - let volatility = if states.len() >= 2 { - let prev_state = states.back().unwrap(); + let volatility = if let Some(prev_state) = states.back() { let price_change = point.close - prev_state.volume; // Using stored value as proxy price_change.abs() } else { diff --git a/risk/src/lib.rs b/risk/src/lib.rs index 677ca84ef..ab37e8570 100644 --- a/risk/src/lib.rs +++ b/risk/src/lib.rs @@ -26,8 +26,9 @@ #![allow(clippy::unused_self)] // Self parameter needed for trait consistency #![allow(clippy::unreadable_literal)] // Large numbers are clear in financial context #![allow(clippy::inline_always)] // Performance-critical code needs inlining hints -#![allow(clippy::expect_used)] // Expect used in validated contexts with clear messages -#![allow(clippy::unwrap_used)] // Unwrap used in validated contexts +#![deny(clippy::expect_used)] +#![deny(clippy::unwrap_used)] +#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))] #![allow(clippy::else_if_without_else)] // Exhaustive else not always needed #![allow(clippy::partial_pub_fields)] // Intentional mixed visibility for safety #![allow(clippy::unnecessary_safety_doc)] // Safety docs retained for clarity diff --git a/risk/src/position_tracker.rs b/risk/src/position_tracker.rs index 326dd76db..915479206 100644 --- a/risk/src/position_tracker.rs +++ b/risk/src/position_tracker.rs @@ -61,8 +61,10 @@ static ref POSITION_UPDATES_COUNTER: Counter = register_counter!( // Create a basic counter as last resort Counter::new("emergency_fallback", "emergency fallback counter") .unwrap_or_else(|_| { - Counter::new("emergency_fallback_fallback", "emergency fallback") - .expect("INVARIANT: Emergency fallback counter creation should never fail") + match Counter::new("emergency_fallback_fallback", "emergency fallback") { + Ok(c) => c, + Err(_) => std::process::abort(), + } }) }) }) }) @@ -87,8 +89,10 @@ static ref POSITION_VALUE_GAUGE: Gauge = register_gauge!( prometheus::core::GenericGauge::new("fallback_gauge", "fallback gauge") .unwrap_or_else(|_| { // Create a basic gauge as last resort - Gauge::new("emergency_fallback_gauge", "emergency fallback gauge") - .expect("Failed to create emergency fallback gauge") + match Gauge::new("emergency_fallback_gauge", "emergency fallback gauge") { + Ok(g) => g, + Err(_) => std::process::abort(), + } }) }) }) @@ -110,8 +114,10 @@ static ref CONCENTRATION_RISK_GAUGE: Gauge = register_gauge!( error!("FATAL: Cannot create any concentration gauge - system continuing"); prometheus::core::GenericGauge::new("basic_concentration", "basic") .unwrap_or_else(|_| { - prometheus::core::GenericGauge::new("fallback_concentration", "fallback") - .expect("Failed to create fallback concentration gauge") + match prometheus::core::GenericGauge::new("fallback_concentration", "fallback") { + Ok(g) => g, + Err(_) => std::process::abort(), + } }) }) }) @@ -132,8 +138,10 @@ static ref PORTFOLIO_COUNT_GAUGE: IntGauge = register_int_gauge!( error!("FATAL: Cannot create any portfolio gauge - system continuing"); prometheus::core::GenericGauge::new("basic_portfolio", "basic") .unwrap_or_else(|_| { - prometheus::core::GenericGauge::new("fallback_portfolio", "fallback") - .expect("Failed to create fallback portfolio gauge") + match prometheus::core::GenericGauge::new("fallback_portfolio", "fallback") { + Ok(g) => g, + Err(_) => std::process::abort(), + } }) }) }) @@ -152,8 +160,10 @@ static ref RISK_BREACHES_COUNTER: Counter = register_counter!( // Last resort: Create the simplest possible counter that should always work prometheus::core::GenericCounter::new("noop_breaches", "no-op breaches counter") .unwrap_or_else(|_| { - prometheus::core::GenericCounter::new("ultimate_fallback", "ultimate fallback") - .expect("Failed to create ultimate fallback counter") + match prometheus::core::GenericCounter::new("ultimate_fallback", "ultimate fallback") { + Ok(c) => c, + Err(_) => std::process::abort(), + } }) }) } @@ -185,9 +195,12 @@ static ref POSITION_PROCESSING_LATENCY: Histogram = register_histogram!( // Use default histogram with basic configuration Histogram::with_opts( HistogramOpts::new("basic_histogram", "basic") - ).unwrap_or_else(|_| Histogram::with_opts( + ).unwrap_or_else(|_| match Histogram::with_opts( HistogramOpts::new("fallback_histogram", "fallback") - ).expect("Failed to create fallback histogram")) + ) { + Ok(h) => h, + Err(_) => std::process::abort(), + }) }) }) }