//! Wave D: Comprehensive Profiling and Bottleneck Analysis //! //! **Agent D38: Profiling and Bottleneck Analysis** //! //! This test performs comprehensive profiling of the complete 54-feature extraction pipeline //! using real Databento data. It identifies CPU hotspots, memory allocation patterns, and //! cache performance to guide optimization efforts. //! //! ## Test Strategy //! 1. Load 5000 bars of real ES.FUT data from Databento //! 2. Process through complete 54-feature pipeline: //! - Wave C features (201 features, indices 0-200) //! - Wave D CUSUM features (10 features, indices 201-210) //! - Wave D ADX features (5 features, indices 211-215) //! - Wave D Transition features (5 features, indices 216-220) //! - Wave D Adaptive features (4 features, indices 221-224) //! 3. Measure per-stage latencies and identify bottlenecks //! 4. Track memory allocations and cache performance //! 5. Generate optimization recommendations //! //! ## Profiling Commands //! ```bash //! # CPU profiling with flamegraph //! cargo flamegraph --test wave_d_profiling_test -p ml --release -- --nocapture //! //! # Cache profiling with perf //! perf stat -e cache-references,cache-misses,L1-dcache-load-misses \ //! cargo test -p ml --test wave_d_profiling_test --release -- --nocapture //! ``` //! //! ## Performance Targets //! - Total pipeline latency P99: <100μs/bar //! - Wave C features: <40μs/bar //! - Wave D features: <25μs/bar //! - Cache miss rate: <5% //! - No hotspots >20% CPU time use anyhow::{Context, Result}; use chrono::{TimeZone, Utc}; use dbn::decode::{dbn::Decoder, DecodeRecord}; use dbn::OhlcvMsg; use std::fs::File; use std::io::BufReader; use std::path::Path; use std::time::Instant; use ml::ensemble::MarketRegime; use ml::features::extraction::OHLCVBar; use ml::features::pipeline::FeatureExtractionPipeline; use ml::features::regime_adaptive::RegimeAdaptiveFeatures; use ml::features::regime_adx::{OHLCVBar as ADXBar, RegimeADXFeatures}; use ml::features::regime_cusum::RegimeCUSUMFeatures; use ml::features::regime_transition::RegimeTransitionFeatures; /// Latency histogram for profiling analysis #[derive(Debug, Clone)] struct LatencyHistogram { samples: Vec, } impl LatencyHistogram { fn new() -> Self { Self { samples: Vec::with_capacity(5000), } } fn record(&mut self, latency_us: u64) { self.samples.push(latency_us); } fn p50(&self) -> u64 { self.percentile(0.50) } fn p90(&self) -> u64 { self.percentile(0.90) } fn p99(&self) -> u64 { self.percentile(0.99) } fn percentile(&self, p: f64) -> u64 { if self.samples.is_empty() { return 0; } let mut sorted = self.samples.clone(); sorted.sort_unstable(); let index = ((sorted.len() as f64 - 1.0) * p) as usize; sorted[index.min(sorted.len() - 1)] } fn mean(&self) -> u64 { if self.samples.is_empty() { return 0; } self.samples.iter().sum::() / self.samples.len() as u64 } fn max(&self) -> u64 { self.samples.iter().copied().max().unwrap_or(0) } fn min(&self) -> u64 { self.samples.iter().copied().min().unwrap_or(0) } } /// Complete 54-feature pipeline profiler struct Feature225Profiler { // Wave C pipeline (201 features) wave_c_pipeline: FeatureExtractionPipeline, // Wave D extractors cusum_features: RegimeCUSUMFeatures, adx_features: RegimeADXFeatures, transition_features: RegimeTransitionFeatures, adaptive_features: RegimeAdaptiveFeatures, // Latency tracking per stage wave_c_latencies: LatencyHistogram, cusum_latencies: LatencyHistogram, adx_latencies: LatencyHistogram, transition_latencies: LatencyHistogram, adaptive_latencies: LatencyHistogram, total_latencies: LatencyHistogram, // Feature output buffer (pre-allocated) feature_buffer: Vec, // Historical bars for ATR calculation historical_bars: Vec, } impl Feature225Profiler { fn new() -> Self { Self { wave_c_pipeline: FeatureExtractionPipeline::new(), cusum_features: RegimeCUSUMFeatures::new(0.0, 1.0, 0.5, 4.0), adx_features: RegimeADXFeatures::new(14), transition_features: RegimeTransitionFeatures::new(4, 0.1), // 4 regimes, EMA alpha 0.1 adaptive_features: RegimeAdaptiveFeatures::new(20, 100_000.0, 14), // window 20, max $100K, ATR 14 wave_c_latencies: LatencyHistogram::new(), cusum_latencies: LatencyHistogram::new(), adx_latencies: LatencyHistogram::new(), transition_latencies: LatencyHistogram::new(), adaptive_latencies: LatencyHistogram::new(), total_latencies: LatencyHistogram::new(), feature_buffer: Vec::with_capacity(54), historical_bars: Vec::with_capacity(50), } } /// Extract complete 54-feature vector for a single bar fn extract_features(&mut self, bar: &OHLCVBar) -> Result> { let total_start = Instant::now(); self.feature_buffer.clear(); // Update historical bars for ATR calculation self.historical_bars.push(bar.clone()); if self.historical_bars.len() > 50 { self.historical_bars.remove(0); } // Stage 1: Wave C features (201 features, indices 0-200) let wave_c_start = Instant::now(); // CRITICAL: Update pipeline state before extraction self.wave_c_pipeline.update(bar); let wave_c_features = self .wave_c_pipeline .extract(bar) .context("Failed to extract Wave C features")?; let wave_c_latency = wave_c_start.elapsed().as_micros() as u64; self.wave_c_latencies.record(wave_c_latency); self.feature_buffer.extend_from_slice(&wave_c_features); // Pad to 201 if needed (Wave C may return 65 features) while self.feature_buffer.len() < 201 { self.feature_buffer.push(0.0); } // Stage 2: Wave D CUSUM features (10 features, indices 201-210) let cusum_start = Instant::now(); let log_return = if bar.close > 1e-10 && bar.open > 1e-10 { (bar.close / bar.open).ln() } else { 0.0 }; let cusum_features = self.cusum_features.update(log_return); let cusum_latency = cusum_start.elapsed().as_micros() as u64; self.cusum_latencies.record(cusum_latency); self.feature_buffer.extend_from_slice(&cusum_features); // Stage 3: Wave D ADX features (5 features, indices 211-215) let adx_start = Instant::now(); let adx_bar = ADXBar { timestamp: bar.timestamp.timestamp(), open: bar.open, high: bar.high, low: bar.low, close: bar.close, volume: bar.volume, }; let adx_features = self.adx_features.update(&adx_bar); let adx_latency = adx_start.elapsed().as_micros() as u64; self.adx_latencies.record(adx_latency); self.feature_buffer.extend_from_slice(&adx_features); // Stage 4: Wave D Transition features (5 features, indices 216-220) let transition_start = Instant::now(); // Simple regime detection based on price movement for demo let regime = if log_return > 0.01 { MarketRegime::Bull } else if log_return < -0.01 { MarketRegime::Bear } else { MarketRegime::Sideways }; let transition_features = self.transition_features.update(regime); let transition_latency = transition_start.elapsed().as_micros() as u64; self.transition_latencies.record(transition_latency); self.feature_buffer.extend_from_slice(&transition_features); // Stage 5: Wave D Adaptive features (4 features, indices 221-224) let adaptive_start = Instant::now(); let adaptive_features = self.adaptive_features.update( regime, log_return, 50_000.0, // $50K position &self.historical_bars, ); let adaptive_latency = adaptive_start.elapsed().as_micros() as u64; self.adaptive_latencies.record(adaptive_latency); self.feature_buffer.extend_from_slice(&adaptive_features); // Record total latency let total_latency = total_start.elapsed().as_micros() as u64; self.total_latencies.record(total_latency); // Verify feature count assert_eq!( self.feature_buffer.len(), 54, "Expected 54 features, got {}", self.feature_buffer.len() ); Ok(self.feature_buffer.clone()) } /// Generate comprehensive profiling report fn generate_report(&self) -> ProfilingReport { ProfilingReport { wave_c: LatencyStats::from_histogram(&self.wave_c_latencies), cusum: LatencyStats::from_histogram(&self.cusum_latencies), adx: LatencyStats::from_histogram(&self.adx_latencies), transition: LatencyStats::from_histogram(&self.transition_latencies), adaptive: LatencyStats::from_histogram(&self.adaptive_latencies), total: LatencyStats::from_histogram(&self.total_latencies), } } } /// Latency statistics for a pipeline stage #[derive(Debug, Clone)] struct LatencyStats { p50_us: u64, p90_us: u64, p99_us: u64, mean_us: u64, min_us: u64, max_us: u64, sample_count: usize, } impl LatencyStats { fn from_histogram(hist: &LatencyHistogram) -> Self { Self { p50_us: hist.p50(), p90_us: hist.p90(), p99_us: hist.p99(), mean_us: hist.mean(), min_us: hist.min(), max_us: hist.max(), sample_count: hist.samples.len(), } } fn cpu_percentage(&self, total_mean: u64) -> f64 { if total_mean == 0 { return 0.0; } (self.mean_us as f64 / total_mean as f64) * 100.0 } } /// Complete profiling report with bottleneck analysis #[derive(Debug)] struct ProfilingReport { wave_c: LatencyStats, cusum: LatencyStats, adx: LatencyStats, transition: LatencyStats, adaptive: LatencyStats, total: LatencyStats, } impl ProfilingReport { fn print_summary(&self) { println!("\n╔═══════════════════════════════════════════════════════════════════════════╗"); println!("║ 54-Feature Pipeline Profiling Report (Agent D38) ║"); println!("╚═══════════════════════════════════════════════════════════════════════════╝\n"); println!("📊 Pipeline Stage Breakdown:"); println!("─────────────────────────────────────────────────────────────────────────────"); let total_mean = self.total.mean_us; self.print_stage("Wave C (201 features)", &self.wave_c, 40, total_mean); self.print_stage("CUSUM (10 features)", &self.cusum, 10, total_mean); self.print_stage("ADX (5 features)", &self.adx, 5, total_mean); self.print_stage("Transition (5 features)", &self.transition, 5, total_mean); self.print_stage("Adaptive (4 features)", &self.adaptive, 5, total_mean); println!("\n═══════════════════════════════════════════════════════════════════════════"); println!("📈 TOTAL PIPELINE (54 features)"); println!("═══════════════════════════════════════════════════════════════════════════"); self.print_detailed_stats(&self.total, 100); // Bottleneck identification println!("\n🔍 Bottleneck Analysis:"); println!("─────────────────────────────────────────────────────────────────────────────"); let mut stages = vec![ ("Wave C", &self.wave_c), ("CUSUM", &self.cusum), ("ADX", &self.adx), ("Transition", &self.transition), ("Adaptive", &self.adaptive), ]; // Sort by mean latency (highest first) stages.sort_by(|a, b| b.1.mean_us.cmp(&a.1.mean_us)); println!("Top 3 Hotspots (by mean latency):"); for (i, (name, stats)) in stages.iter().take(3).enumerate() { let cpu_pct = stats.cpu_percentage(total_mean); let status = if cpu_pct > 20.0 { "⚠️ HOTSPOT" } else { "✅ OK" }; println!( " {}. {}: {:.1}μs ({:.1}% of total) {}", i + 1, name, stats.mean_us, cpu_pct, status ); } // Cache performance (placeholder - requires perf integration) println!("\n💾 Memory & Cache Performance:"); println!("─────────────────────────────────────────────────────────────────────────────"); println!(" Note: Run 'perf stat -e cache-references,cache-misses' for detailed metrics"); println!(" Expected: <5% cache miss rate, <1KB allocations per bar"); // Optimization recommendations println!("\n💡 Optimization Recommendations:"); println!("─────────────────────────────────────────────────────────────────────────────"); self.generate_recommendations(&stages); // Overall assessment println!("\n📋 Production Readiness Assessment:"); println!("─────────────────────────────────────────────────────────────────────────────"); let p99_ok = self.total.p99_us <= 100; let outliers_ok = self.total.max_us <= 500; let balanced = stages[0].1.cpu_percentage(total_mean) <= 50.0; println!( " P99 latency: {} ({:>3}μs target, actual: {}μs)", if p99_ok { "✅ PASS" } else { "❌ FAIL" }, 100, self.total.p99_us ); println!( " Max latency: {} ({:>3}μs target, actual: {}μs)", if outliers_ok { "✅ PASS" } else { "❌ FAIL" }, 500, self.total.max_us ); println!( " CPU balance: {} (top stage <50%, actual: {:.1}%)", if balanced { "✅ PASS" } else { "❌ FAIL" }, stages[0].1.cpu_percentage(total_mean) ); let overall_pass = p99_ok && outliers_ok && balanced; println!( "\n Overall: {}", if overall_pass { "✅ PRODUCTION READY" } else { "⚠️ OPTIMIZATION RECOMMENDED" } ); println!(); } fn print_stage(&self, name: &str, stats: &LatencyStats, target_p99: u64, total_mean: u64) { let cpu_pct = stats.cpu_percentage(total_mean); let target_met = if stats.p99_us <= target_p99 { "✅" } else { "❌" }; println!("\n{}", name); println!( " P50: {:>4}μs P90: {:>4}μs P99: {:>4}μs {} (target: <{}μs)", stats.p50_us, stats.p90_us, stats.p99_us, target_met, target_p99 ); println!(" Mean: {:>4}μs CPU%: {:>5.1}%", stats.mean_us, cpu_pct); } fn print_detailed_stats(&self, stats: &LatencyStats, target_p99: u64) { let target_met = if stats.p99_us <= target_p99 { "✅" } else { "❌" }; println!(" Samples: {}", stats.sample_count); println!(" P50: {:>6}μs", stats.p50_us); println!(" P90: {:>6}μs", stats.p90_us); println!( " P99: {:>6}μs {} (target: <{}μs)", stats.p99_us, target_met, target_p99 ); println!(" Mean: {:>6}μs", stats.mean_us); println!(" Min: {:>6}μs", stats.min_us); println!(" Max: {:>6}μs", stats.max_us); } fn generate_recommendations(&self, stages: &[(&str, &LatencyStats)]) { let total_mean = self.total.mean_us; // Recommendation 1: Address top hotspot if let Some((name, stats)) = stages.first() { let cpu_pct = stats.cpu_percentage(total_mean); if cpu_pct > 30.0 { println!(" 1. {} consumes {:.1}% of CPU time", name, cpu_pct); println!(" → Consider algorithmic optimization or SIMD vectorization"); } } // Recommendation 2: P99 latency if self.total.p99_us > 100 { println!( " 2. P99 latency ({}μs) exceeds target (100μs)", self.total.p99_us ); println!(" → Profile with 'cargo flamegraph' to identify outlier causes"); } // Recommendation 3: Outliers if self.total.max_us > 500 { println!(" 3. Max latency ({}μs) has outliers", self.total.max_us); println!(" → Check for cold start effects or unexpected allocations"); } // Recommendation 4: Cache efficiency println!(" 4. Run cache profiling to validate <5% miss rate:"); println!(" → perf stat -e cache-references,cache-misses cargo test ... --release"); } } /// Load OHLCV bars from Databento DBN file fn load_dbn_bars(path: &Path, max_bars: usize) -> Result> { let file = File::open(path).with_context(|| format!("Failed to open DBN file: {}", path.display()))?; let reader = BufReader::new(file); let mut decoder = Decoder::new(reader)?; let mut bars = Vec::with_capacity(max_bars); while let Some(record) = decoder.decode_record::()? { if bars.len() >= max_bars { break; } // Convert Databento OHLCV to internal format let timestamp_secs = (record.hd.ts_event / 1_000_000_000) as i64; let timestamp = Utc.timestamp_opt(timestamp_secs, 0).unwrap(); let bar = OHLCVBar { timestamp, open: record.open as f64 / 1_000_000_000.0, high: record.high as f64 / 1_000_000_000.0, low: record.low as f64 / 1_000_000_000.0, close: record.close as f64 / 1_000_000_000.0, volume: record.volume as f64, }; bars.push(bar); } Ok(bars) } #[test] #[ignore = "Run explicitly with: cargo test -p ml --test wave_d_profiling_test -- --ignored --nocapture"] fn test_wave_d_comprehensive_profiling() -> Result<()> { println!("\n🔍 Starting comprehensive 54-feature pipeline profiling...\n"); // Locate DBN test data let test_data_paths = vec![ "/home/jgrusewski/Work/foxhunt/test_data/real/databento/ml_training_small/6E.FUT_ohlcv-1m_2024-01-02.dbn", "/home/jgrusewski/Work/foxhunt/test_data/real/databento/GC_continuous_ohlcv-1m_2024-01-02_to_2024-01-31.uncompressed.dbn", ]; let mut dbn_path = None; for path_str in &test_data_paths { let path = Path::new(path_str); if path.exists() { dbn_path = Some(path); break; } } let dbn_path = dbn_path.context("No DBN test data found")?; println!("📁 Loading data from: {}", dbn_path.display()); // Load 5000 bars let bars = load_dbn_bars(dbn_path, 5000).context("Failed to load DBN bars")?; println!("✅ Loaded {} bars for profiling\n", bars.len()); // Initialize profiler let mut profiler = Feature225Profiler::new(); // Warmup phase: First populate Wave C pipeline with 50 bars (required minimum) println!("🔥 Warmup phase (150 iterations)..."); println!(" Step 1: Warming up Wave C pipeline (50 bars)..."); for bar in bars.iter().take(50) { profiler.wave_c_pipeline.update(bar); } // Now extract features for the next 100 bars to warm up all components println!(" Step 2: Warming up Wave D features (100 bars)..."); for bar in bars.iter().skip(50).take(100) { profiler.extract_features(bar)?; } // Reset latency tracking after warmup (but keep historical bars for Wave C pipeline) profiler.wave_c_latencies = LatencyHistogram::new(); profiler.cusum_latencies = LatencyHistogram::new(); profiler.adx_latencies = LatencyHistogram::new(); profiler.transition_latencies = LatencyHistogram::new(); profiler.adaptive_latencies = LatencyHistogram::new(); profiler.total_latencies = LatencyHistogram::new(); // Profiling phase (process all bars) println!("📊 Profiling phase ({} iterations)...", bars.len()); let profiling_start = Instant::now(); for (i, bar) in bars.iter().enumerate() { profiler.extract_features(bar)?; if (i + 1) % 1000 == 0 { println!(" Processed {}/{} bars...", i + 1, bars.len()); } } let total_profiling_time = profiling_start.elapsed(); println!( "✅ Profiling complete in {:.2}s\n", total_profiling_time.as_secs_f64() ); // Generate and print report let report = profiler.generate_report(); report.print_summary(); // Write report to file let report_path = "/home/jgrusewski/Work/foxhunt/AGENT_D38_PROFILING_ANALYSIS_REPORT.md"; std::fs::write( report_path, format!( "{:#?}\n\nTotal profiling time: {:.2}s\nBars processed: {}", report, total_profiling_time.as_secs_f64(), bars.len() ), )?; println!("📄 Report saved to: {}\n", report_path); Ok(()) } #[test] fn test_latency_histogram_basic() { let mut hist = LatencyHistogram::new(); // Record test latencies for i in 1..=100 { hist.record(i); } assert_eq!(hist.p50(), 50); assert_eq!(hist.p90(), 90); assert_eq!(hist.p99(), 99); assert_eq!(hist.mean(), 50); assert_eq!(hist.min(), 1); assert_eq!(hist.max(), 100); } #[test] fn test_feature_count_validation() -> Result<()> { let mut profiler = Feature225Profiler::new(); // Create dummy bar let bar = OHLCVBar { timestamp: Utc::now(), open: 4500.0, high: 4505.0, low: 4495.0, close: 4502.0, volume: 1000.0, }; // Extract features and verify count let features = profiler.extract_features(&bar)?; assert_eq!(features.len(), 54, "Expected exactly 54 features"); Ok(()) }