//! PPO Model Evaluation Pipeline //! //! Complete evaluation system for trained PPO models following the DQN evaluation architecture. //! Integrates checkpoint loading, parquet data processing, inference, and metrics calculation. //! //! # Usage //! //! ```bash //! # Evaluate with default settings (auto-detect CUDA) //! cargo run -p ml --example evaluate_ppo --release --features cuda //! //! # Evaluate on CPU with custom model //! cargo run -p ml --example evaluate_ppo --release -- \ //! --actor-checkpoint ml/trained_models/ppo_actor_final.safetensors \ //! --critic-checkpoint ml/trained_models/ppo_critic_final.safetensors \ //! --device cpu //! //! # Evaluate with JSON output for CI/CD //! cargo run -p ml --example evaluate_ppo --release --features cuda -- \ //! --parquet-file test_data/ES_FUT_unseen.parquet \ //! --output-json evaluation_results.json //! //! # Custom warmup period (skip first 30 bars) //! cargo run -p ml --example evaluate_ppo --release --features cuda -- \ //! --warmup-bars 30 \ //! --parquet-file test_data/ES_FUT_validation.parquet //! ``` //! //! # Architecture //! //! ```text //! ┌─────────────────────────────────────────────────────────────────┐ //! │ PPO EVALUATION ORCHESTRATOR │ //! │ │ //! │ 1. INITIALIZATION │ //! │ ├─ Parse CLI args (Configuration) │ //! │ ├─ Setup tracing (stdout + /tmp/ppo_eval.log) │ //! │ ├─ Validate config │ //! │ └─ Setup graceful shutdown (Ctrl+C / SIGTERM) │ //! │ │ //! │ 2. PARALLEL LOADING (tokio::try_join!) │ //! │ ├─ Load Parquet data (225 features) ─────┐ │ //! │ └─ Load PPO checkpoints (actor+critic) ──┴─ Concurrent │ //! │ │ //! │ 3. SEQUENTIAL INFERENCE │ //! │ ├─ Run inference (action selection + value estimation) │ //! │ ├─ Calculate metrics (action distribution, policy stats) │ //! │ └─ Track total elapsed time │ //! │ │ //! │ 4. REPORT GENERATION │ //! │ ├─ Generate report (console output) │ //! │ └─ Export JSON (if configured) │ //! │ │ //! │ 5. GRACEFUL SHUTDOWN │ //! │ ├─ Stop inference loop (if interrupted) │ //! │ ├─ Generate partial report │ //! │ └─ Exit with appropriate code (0=success, 1=error) │ //! └─────────────────────────────────────────────────────────────────┘ //! ``` use anyhow::{Context, Result}; use candle_core::Device; use clap::Parser; use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Instant; use tokio::signal; use tracing::{info, warn}; use tracing_subscriber::FmtSubscriber; use ml::data_loaders::load_parquet_data; use ml::dqn::TradingAction; use ml::ppo::ppo::{PPOConfig, WorkingPPO}; // ============================================================================ // Component 1: CLI Configuration // ============================================================================ /// PPO Model Evaluation Configuration /// /// Parses and validates CLI arguments for evaluating a trained PPO model. /// /// # Validation Rules /// /// - `actor_checkpoint`: Must exist and be a valid SafeTensors file /// - `critic_checkpoint`: Must exist and be a valid SafeTensors file /// - `parquet_file`: Must exist and contain OHLCV data /// - `device`: Must be "cpu", "cuda", or "auto" /// - `warmup_bars`: Must be in range [10, 100] (prevents over/under fitting) #[derive(Parser, Debug)] #[command( name = "evaluate_ppo", about = "Evaluate PPO model on unseen market data", long_about = "Complete PPO evaluation pipeline with parallel loading, inference, \ metrics calculation, and report generation." )] struct EvaluationConfig { /// Path to trained PPO actor (policy) network (SafeTensors format) #[arg(long, default_value = "/tmp/ppo_actor_final.safetensors")] actor_checkpoint: PathBuf, /// Path to trained PPO critic (value) network (SafeTensors format) #[arg(long, default_value = "/tmp/ppo_critic_final.safetensors")] critic_checkpoint: PathBuf, /// Path to Parquet file with unseen OHLCV data #[arg(long, default_value = "test_data/ES_FUT_unseen.parquet")] parquet_file: PathBuf, /// Device selection: cpu, cuda, or auto #[arg(long, default_value = "auto")] device: String, /// Number of warmup bars to skip (insufficient history) #[arg(long, default_value_t = 50)] warmup_bars: usize, /// Optional JSON output path for CI/CD integration #[arg(long)] output_json: Option, /// Verbose logging (DEBUG level) #[arg(short, long)] verbose: bool, } impl EvaluationConfig { /// Validates the configuration parameters pub fn validate(&self) -> Result<()> { // Validate actor_checkpoint exists if !self.actor_checkpoint.exists() { return Err(anyhow::anyhow!( "Actor checkpoint does not exist: {}\n\n\ Suggestion: Train a PPO model first using:\n\ cargo run -p ml --example train_ppo --release --features cuda -- \\\n\ --actor-output {} --critic-output {}", self.actor_checkpoint.display(), self.actor_checkpoint.display(), self.critic_checkpoint.display() )); } if !self.actor_checkpoint.is_file() { return Err(anyhow::anyhow!( "Actor checkpoint path is not a regular file: {}", self.actor_checkpoint.display() )); } // Validate critic_checkpoint exists if !self.critic_checkpoint.exists() { return Err(anyhow::anyhow!( "Critic checkpoint does not exist: {}\n\n\ Suggestion: Train a PPO model first using:\n\ cargo run -p ml --example train_ppo --release --features cuda", self.critic_checkpoint.display() )); } if !self.critic_checkpoint.is_file() { return Err(anyhow::anyhow!( "Critic checkpoint path is not a regular file: {}", self.critic_checkpoint.display() )); } // Validate parquet_file exists if !self.parquet_file.exists() { return Err(anyhow::anyhow!( "Parquet file does not exist: {}\n\n\ Suggestion: Generate unseen data using:\n\ # Download data from Databento API for a different time period\n\ # than your training data (temporal split)", self.parquet_file.display() )); } if !self.parquet_file.is_file() { return Err(anyhow::anyhow!( "Parquet path is not a regular file: {}", self.parquet_file.display() )); } // Validate device string match self.device.as_str() { "cpu" | "cuda" | "auto" => { // Valid device string }, _ => { return Err(anyhow::anyhow!( "Invalid device: '{}'\n\n\ Valid options:\n\ - 'cpu': Force CPU execution\n\ - 'cuda': Force CUDA GPU execution (requires NVIDIA GPU)\n\ - 'auto': Auto-detect CUDA availability (recommended)", self.device )); }, } // Validate warmup_bars range if self.warmup_bars < 10 { return Err(anyhow::anyhow!( "Warmup bars too small: {} (minimum: 10)\n\n\ At least 10 bars are required for basic feature computation.\n\ Recommended: 50 bars for most models.", self.warmup_bars )); } if self.warmup_bars > 100 { return Err(anyhow::anyhow!( "Warmup bars too large: {} (maximum: 100)\n\n\ Using more than 100 warmup bars wastes evaluation data.\n\ Recommended: 50 bars for most models.", self.warmup_bars )); } // Validate output_json path (if specified) if let Some(ref output_path) = self.output_json { // Check if parent directory exists if let Some(parent) = output_path.parent() { if !parent.exists() { return Err(anyhow::anyhow!( "Output JSON parent directory does not exist: {}\n\n\ Suggestion: Create the directory first:\n\ mkdir -p {}", parent.display(), parent.display() )); } } // Check if file already exists (warn, but don't fail) if output_path.exists() { info!( "Output JSON file already exists and will be overwritten: {}", output_path.display() ); } } Ok(()) } } // ============================================================================ // Component 2: Model Loading // ============================================================================ /// Load PPO model from SafeTensors checkpoints with device selection /// /// # Arguments /// * `actor_path` - Path to actor (policy) network SafeTensors file /// * `critic_path` - Path to critic (value) network SafeTensors file /// * `device_str` - Device selection: "cpu", "cuda", or "auto" /// /// # Returns /// WorkingPPO ready for inference with loaded weights /// /// # Errors /// Returns error if: /// - Files not found or not readable /// - SafeTensors deserialization fails /// - CUDA requested but unavailable /// - Model dimensions incorrect (expected: 225 input features, 3 actions) /// /// # Note /// Uses WorkingPPO::load_checkpoint() for production-ready loading. /// Architecture: 225 input → [128, 64] policy / [256, 128, 64] value → 3/1 output fn load_ppo_model(actor_path: &Path, critic_path: &Path, device_str: &str) -> Result { info!("Component 2: Loading PPO model"); info!(" Actor path: {}", actor_path.display()); info!(" Critic path: {}", critic_path.display()); info!(" Device selection: {}", device_str); // 1. Parse device string to create Device let device = match device_str.to_lowercase().as_str() { "cpu" => { info!(" Using CPU device (explicitly requested)"); Device::Cpu }, "cuda" => { info!(" Using CUDA device (explicitly requested)"); Device::new_cuda(0).context( "CUDA device requested but unavailable. \ Suggestions:\n\ - Check nvidia-smi to verify GPU is available\n\ - Try device_str=\"auto\" for automatic fallback to CPU\n\ - Use device_str=\"cpu\" to force CPU execution", )? }, "auto" => match Device::cuda_if_available(0) { Ok(cuda_device) => { info!(" Auto-selected CUDA device (GPU available)"); cuda_device }, Err(e) => { warn!(" CUDA unavailable, falling back to CPU: {}", e); info!(" Using CPU device (auto-fallback)"); Device::Cpu }, }, _ => { return Err(anyhow::anyhow!( "Invalid device_str: '{}'. Must be one of: 'cpu', 'cuda', 'auto'", device_str )); }, }; let device_name = match &device { Device::Cpu => "CPU", Device::Cuda(_) => "CUDA:0", _ => "Unknown", }; // 2. Check if checkpoint files exist and are readable for (path, name) in [(actor_path, "Actor"), (critic_path, "Critic")] { if !path.exists() { return Err(anyhow::anyhow!( "{} checkpoint not found: {}\n\ Suggestions:\n\ - Check the file path is correct\n\ - Verify the model was saved successfully during training\n\ - Look for checkpoint files in ml/trained_models/", name, path.display() )); } if !path.is_file() { return Err(anyhow::anyhow!( "Path exists but is not a file: {}", path.display() )); } // Check file permissions (read access) match std::fs::metadata(path) { Ok(metadata) => { if metadata.permissions().readonly() { warn!(" {} checkpoint is read-only: {}", name, path.display()); } info!( " {} file size: {} bytes ({:.2} KB)", name, metadata.len(), metadata.len() as f64 / 1024.0 ); }, Err(e) => { return Err(anyhow::anyhow!( "Cannot read file metadata for {}: {}", path.display(), e )); }, } } // 3. Create PPOConfig with correct architecture // Architecture: 225 input → [128, 64] policy / [512, 128, 64] value → 3/1 output // NOTE: Using [512, 128, 64] for value network to match production training configs info!(" Creating PPO configuration..."); let config = PPOConfig { state_dim: 225, num_actions: 3, policy_hidden_dims: vec![128, 64], value_hidden_dims: vec![512, 128, 64], // Matches production training (not default [256, 128, 64]) policy_learning_rate: 3e-5, value_learning_rate: 1e-4, clip_epsilon: 0.2, value_loss_coeff: 1.0, entropy_coeff: 0.05, ..Default::default() }; info!(" Model architecture:"); info!(" - Input: 225 features"); info!( " - Policy (Actor): {:?} → 3 actions (BUY, SELL, HOLD)", config.policy_hidden_dims ); info!( " - Value (Critic): {:?} → 1 value estimate", config.value_hidden_dims ); // 4. Load PPO model from checkpoints info!(" Loading weights from SafeTensors checkpoints..."); let actor_path_str = actor_path.to_str().ok_or_else(|| { anyhow::anyhow!( "Actor path contains invalid UTF-8: {}", actor_path.display() ) })?; let critic_path_str = critic_path.to_str().ok_or_else(|| { anyhow::anyhow!( "Critic path contains invalid UTF-8: {}", critic_path.display() ) })?; let ppo = WorkingPPO::load_checkpoint(actor_path_str, critic_path_str, config, device.clone()) .context(format!( "Failed to load PPO checkpoints from actor={}, critic={}\n\ Possible causes:\n\ - Files are corrupted (try retraining)\n\ - Wrong architecture (expected: 225 → [128,64] / [512,128,64] → 3/1)\n\ - Incompatible tensor types or shapes\n\ - Device memory issue ({})", actor_path.display(), critic_path.display(), device_name ))?; info!("Component 2: PPO checkpoints loaded successfully"); info!(" Actor: {}", actor_path.display()); info!(" Critic: {}", critic_path.display()); Ok(ppo) } // ============================================================================ // Component 3: Inference Engine // ============================================================================ /// Result of a single PPO inference #[derive(Debug, Clone)] struct InferenceResult { action: usize, // 0=BUY, 1=SELL, 2=HOLD value_estimate: f32, // Critic's value estimate for the state latency_us: u64, // Microseconds for this inference } /// Run PPO inference on all feature vectors with progress tracking /// /// # Arguments /// * `ppo` - WorkingPPO for inference (uses actor for action selection) /// * `features` - 225-dimensional feature vectors /// * `shutdown_flag` - Atomic flag for graceful shutdown (Ctrl+C) /// /// # Returns /// Vector of inference results (action, value estimate, latency per bar) /// /// # Notes /// - Uses WorkingPPO's predict() method to get action probabilities /// - Selects action with highest probability (greedy policy) /// - Uses critic network to estimate state values /// - Tracks latency per inference in microseconds /// - Progress bar shows real-time inference speed /// - Respects shutdown flag for graceful interruption fn run_inference( ppo: &WorkingPPO, features: Vec<[f64; 225]>, shutdown_flag: &Arc, ) -> Result> { info!("Component 4: Running PPO inference"); let total_bars = features.len(); if total_bars == 0 { return Err(anyhow::anyhow!("No feature vectors provided for inference")); } info!(" Total bars to process: {}", total_bars); let mut results = Vec::with_capacity(total_bars); let mut total_inference_time_us = 0u64; let mut skipped_bars = 0usize; let start_time = Instant::now(); let mut last_progress_update = Instant::now(); // Run inference for each feature vector for (i, feature_vec) in features.iter().enumerate() { // Check for shutdown signal if shutdown_flag.load(Ordering::Relaxed) { warn!( " Shutdown signal received, stopping inference at bar {}/{}", i, total_bars ); break; } // Start timer for this inference let timer = Instant::now(); // Convert f64 features to f32 for PPO networks let state_f32: Vec = feature_vec.iter().map(|&x| x as f32).collect(); // Get action probabilities from policy network let action_probs = match ppo.predict(&state_f32) { Ok(probs) => probs, Err(e) => { if skipped_bars < 10 { warn!(" Bar {}: Policy prediction failed: {}. Skipping.", i, e); } skipped_bars += 1; continue; }, }; // Select action with highest probability (greedy policy) let (action, _max_prob) = action_probs .iter() .enumerate() .max_by(|(_, a), (_, b)| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)) .ok_or_else(|| anyhow::anyhow!("Failed to select action from probabilities"))?; // Get value estimate from critic network using act() method let (_, value_estimate) = match ppo.act(&state_f32) { Ok((_, value)) => (action, value), Err(e) => { if skipped_bars < 10 { warn!(" Bar {}: Value estimation failed: {}. Skipping.", i, e); } skipped_bars += 1; continue; }, }; // Check for NaN/Inf in value estimate if !value_estimate.is_finite() { if skipped_bars < 10 { warn!( " Bar {}: Value estimate is NaN/Inf, skipping. Value: {}", i, value_estimate ); } skipped_bars += 1; continue; } // Calculate latency for this inference let latency_us = timer.elapsed().as_micros() as u64; total_inference_time_us += latency_us; // Store result results.push(InferenceResult { action, value_estimate, latency_us, }); // Update progress every 1 second or every 10% completion let should_update = last_progress_update.elapsed().as_secs() >= 1 || (i + 1) % (total_bars / 10).max(1) == 0 || i == 0 || i == total_bars - 1; if should_update { let elapsed_sec = start_time.elapsed().as_secs_f64(); let avg_speed = if elapsed_sec > 0.0 { (i + 1) as f64 / elapsed_sec } else { 0.0 }; let progress_pct = ((i + 1) as f64 / total_bars as f64) * 100.0; info!( " Progress: {}/{} ({:.1}%) | Speed: {:.1} bars/sec | Skipped: {}", i + 1, total_bars, progress_pct, avg_speed, skipped_bars ); last_progress_update = Instant::now(); } } info!("Component 4: Inference complete"); // Calculate summary statistics let processed_bars = results.len(); let total_time_sec = start_time.elapsed().as_secs_f64(); let avg_latency_us = if processed_bars > 0 { total_inference_time_us / processed_bars as u64 } else { 0 }; let avg_speed = if total_time_sec > 0.0 { processed_bars as f64 / total_time_sec } else { 0.0 }; // Log summary with detailed metrics info!(" Inference Summary:"); info!(" - Total bars: {}", total_bars); info!(" - Processed: {}", processed_bars); info!(" - Skipped (NaN/Inf/errors): {}", skipped_bars); info!( " - Skip rate: {:.2}%", (skipped_bars as f64 / total_bars as f64) * 100.0 ); info!(" - Total time: {:.2}s", total_time_sec); info!( " - Average latency: {}μs ({:.2}ms)", avg_latency_us, avg_latency_us as f64 / 1000.0 ); info!(" - Average speed: {:.1} bars/sec", avg_speed); // Validate results if results.is_empty() { return Err(anyhow::anyhow!( "All {} inference attempts failed (likely NaN/Inf in value estimates or network errors)", total_bars )); } Ok(results) } // ============================================================================ // Component 5: Metrics Calculator // ============================================================================ /// Comprehensive evaluation metrics for PPO model validation #[derive(Debug, Clone, Serialize, Deserialize)] pub struct EvaluationMetrics { pub total_bars: usize, pub action_distribution: ActionDistribution, pub value_statistics: ValueStatistics, pub latency_stats: LatencyStats, pub policy_consistency: PolicyConsistency, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ActionDistribution { pub buy_count: usize, pub sell_count: usize, pub hold_count: usize, pub buy_pct: f64, pub sell_pct: f64, pub hold_pct: f64, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ValueStatistics { pub mean_value: f64, pub median_value: f64, pub std_dev: f64, pub min_value: f64, pub max_value: f64, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct LatencyStats { pub mean_us: f64, pub median_us: u64, pub p50_us: u64, pub p95_us: u64, pub p99_us: u64, pub min_us: u64, pub max_us: u64, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PolicyConsistency { pub total_switches: usize, pub switch_rate: f64, pub interpretation: String, } /// Calculate comprehensive evaluation metrics from inference results fn calculate_metrics(results: &[InferenceResult]) -> Result { info!("Component 5: Calculating metrics"); if results.is_empty() { return Err(anyhow::anyhow!( "Cannot calculate metrics from empty results" )); } let total_bars = results.len(); info!(" Total bars: {}", total_bars); // 1. Action distribution let mut buy_count = 0usize; let mut sell_count = 0usize; let mut hold_count = 0usize; for result in results { match result.action { 0 => buy_count += 1, 1 => sell_count += 1, 2 => hold_count += 1, _ => warn!(" Invalid action: {}", result.action), } } let buy_pct = (buy_count as f64 / total_bars as f64) * 100.0; let sell_pct = (sell_count as f64 / total_bars as f64) * 100.0; let hold_pct = (hold_count as f64 / total_bars as f64) * 100.0; // 2. Value statistics let values: Vec = results.iter().map(|r| r.value_estimate as f64).collect(); let mean_value = values.iter().sum::() / values.len() as f64; let mut sorted_values = values.clone(); sorted_values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); let median_value = sorted_values[sorted_values.len() / 2]; let variance = values .iter() .map(|v| { let diff = v - mean_value; diff * diff }) .sum::() / values.len() as f64; let std_dev = variance.sqrt(); let min_value = sorted_values[0]; let max_value = sorted_values[sorted_values.len() - 1]; // 3. Latency statistics let mut latencies: Vec = results.iter().map(|r| r.latency_us).collect(); latencies.sort_unstable(); let mean_us = latencies.iter().sum::() as f64 / latencies.len() as f64; let median_us = latencies[latencies.len() / 2]; let p50_us = median_us; let p95_us = latencies[(latencies.len() as f64 * 0.95) as usize]; let p99_us = latencies[(latencies.len() as f64 * 0.99) as usize]; let min_us = latencies[0]; let max_us = latencies[latencies.len() - 1]; // 4. Policy consistency let mut total_switches = 0usize; for i in 1..results.len() { if results[i].action != results[i - 1].action { total_switches += 1; } } let switch_rate = if results.len() > 1 { total_switches as f64 / (results.len() - 1) as f64 } else { 0.0 }; let interpretation = if switch_rate < 0.10 { "Stable - Low adaptability".to_string() } else if switch_rate <= 0.30 { "Moderate - Healthy adaptive behavior".to_string() } else { "Volatile - High uncertainty or noise".to_string() }; info!("Component 5: Metrics calculated successfully"); Ok(EvaluationMetrics { total_bars, action_distribution: ActionDistribution { buy_count, sell_count, hold_count, buy_pct, sell_pct, hold_pct, }, value_statistics: ValueStatistics { mean_value, median_value, std_dev, min_value, max_value, }, latency_stats: LatencyStats { mean_us, median_us, p50_us, p95_us, p99_us, min_us, max_us, }, policy_consistency: PolicyConsistency { total_switches, switch_rate, interpretation, }, }) } // ============================================================================ // Component 6: Report Generator // ============================================================================ /// Generate comprehensive evaluation report /// /// # Arguments /// * `metrics` - Evaluation metrics from Component 5 /// * `config` - CLI configuration /// * `elapsed` - Total elapsed time (including data loading, inference, etc.) /// /// # Returns /// Result indicating success/failure of report generation /// /// # Side Effects /// - Prints formatted report to stdout /// - Writes JSON file if config.output_json is specified /// - Validates production readiness thresholds fn generate_report( metrics: &EvaluationMetrics, config: &EvaluationConfig, elapsed: std::time::Duration, ) -> Result<()> { info!("Component 6: Generating evaluation report"); // 1. Print header println!("\n╔══════════════════════════════════════════════════════════════════════╗"); println!("║ PPO MODEL EVALUATION REPORT ║"); println!("╚══════════════════════════════════════════════════════════════════════╝"); println!(); // 2. Configuration summary println!("═══ Configuration ═══"); println!(" Actor checkpoint: {}", config.actor_checkpoint.display()); println!( " Critic checkpoint: {}", config.critic_checkpoint.display() ); println!(" Data file: {}", config.parquet_file.display()); println!(" Device: {}", config.device); println!(" Warmup bars: {}", config.warmup_bars); println!(" Total runtime: {:.2}s", elapsed.as_secs_f64()); println!(); // 3. Action distribution println!("═══ Action Distribution ═══"); println!( " BUY: {:5} ({:5.1}%)", metrics.action_distribution.buy_count, metrics.action_distribution.buy_pct ); println!( " SELL: {:5} ({:5.1}%)", metrics.action_distribution.sell_count, metrics.action_distribution.sell_pct ); println!( " HOLD: {:5} ({:5.1}%)", metrics.action_distribution.hold_count, metrics.action_distribution.hold_pct ); println!(" ────────────────────"); println!(" Total: {} bars", metrics.total_bars); println!(); // 4. Value statistics println!("═══ Value Estimate Statistics ═══"); println!(" Mean: {:8.4}", metrics.value_statistics.mean_value); println!(" Median: {:8.4}", metrics.value_statistics.median_value); println!(" Std Dev: {:8.4}", metrics.value_statistics.std_dev); println!( " Range: {:8.4} - {:8.4}", metrics.value_statistics.min_value, metrics.value_statistics.max_value ); println!(); // 5. Latency statistics println!("═══ Latency Statistics ═══"); println!( " Mean: {:6.1} μs ({:.3} ms)", metrics.latency_stats.mean_us, metrics.latency_stats.mean_us / 1000.0 ); println!( " Median: {:6} μs ({:.3} ms)", metrics.latency_stats.median_us, metrics.latency_stats.median_us as f64 / 1000.0 ); println!( " P95: {:6} μs ({:.3} ms)", metrics.latency_stats.p95_us, metrics.latency_stats.p95_us as f64 / 1000.0 ); println!( " P99: {:6} μs ({:.3} ms)", metrics.latency_stats.p99_us, metrics.latency_stats.p99_us as f64 / 1000.0 ); println!( " Range: {:6} - {:6} μs", metrics.latency_stats.min_us, metrics.latency_stats.max_us ); println!(); // 6. Policy consistency println!("═══ Policy Consistency ═══"); println!( " Switches: {} / {} bars", metrics.policy_consistency.total_switches, metrics.total_bars - 1 ); println!( " Switch rate: {:.1}%", metrics.policy_consistency.switch_rate * 100.0 ); println!( " Interpretation: {}", metrics.policy_consistency.interpretation ); println!(); // 7. Production readiness check println!("═══ Production Readiness Check ═══"); let latency_ok = metrics.latency_stats.p99_us < 5_000; println!( " {} Latency P99 < 5,000μs: {} (actual: {} μs)", if latency_ok { "✅" } else { "❌" }, latency_ok, metrics.latency_stats.p99_us ); let consistency_ok = metrics.policy_consistency.switch_rate >= 0.10 && metrics.policy_consistency.switch_rate <= 0.30; println!( " {} Policy switch rate 10-30%: {} (actual: {:.1}%)", if consistency_ok { "✅" } else { "❌" }, consistency_ok, metrics.policy_consistency.switch_rate * 100.0 ); let balance_ok = metrics.action_distribution.buy_pct >= 5.0 && metrics.action_distribution.sell_pct >= 5.0 && metrics.action_distribution.hold_pct >= 5.0; println!( " {} Balanced actions (each >5%): {}", if balance_ok { "✅" } else { "⚠️ " }, balance_ok ); let values_ok = metrics.value_statistics.mean_value.is_finite() && metrics.value_statistics.std_dev.is_finite(); println!( " {} Value estimates finite: {}", if values_ok { "✅" } else { "❌" }, values_ok ); println!(); let all_ok = latency_ok && consistency_ok && values_ok; if all_ok { println!("Model is PRODUCTION READY!"); } else { println!("Model requires further tuning before production deployment"); } println!(); // 8. Export JSON (if configured) if let Some(ref output_path) = config.output_json { info!(" Exporting metrics to JSON: {}", output_path.display()); let json = serde_json::to_string_pretty(&metrics) .context("Failed to serialize metrics to JSON")?; std::fs::write(output_path, json) .context(format!("Failed to write JSON to {}", output_path.display()))?; println!("Metrics exported to: {}", output_path.display()); println!(); } info!("Component 6: Report generated successfully"); Ok(()) } // ============================================================================ // Component 7: Main Orchestrator // ============================================================================ #[tokio::main] async fn main() -> Result<()> { // ======================================================================== // PHASE 1: INITIALIZATION // ======================================================================== // 1.1: Parse CLI arguments let config = EvaluationConfig::parse(); // 1.2: Setup tracing subscriber (stdout logging) let log_level = if config.verbose { tracing::Level::DEBUG } else { tracing::Level::INFO }; let subscriber = FmtSubscriber::builder() .with_max_level(log_level) .with_target(false) .with_level(true) .finish(); tracing::subscriber::set_global_default(subscriber) .context("Failed to set tracing subscriber")?; // 1.3: Log startup banner info!("╔══════════════════════════════════════════════════════════════════════╗"); info!("║ PPO Model Evaluation Pipeline - Main Orchestrator ║"); info!("║ Version: 1.0.0 ║"); info!( "║ Timestamp: {} ║", chrono::Local::now().format("%Y-%m-%d %H:%M:%S") ); info!("╚══════════════════════════════════════════════════════════════════════╝"); info!(""); // 1.4: Log configuration info!("Configuration:"); info!( " • Actor checkpoint: {}", config.actor_checkpoint.display() ); info!( " • Critic checkpoint: {}", config.critic_checkpoint.display() ); info!(" • Parquet file: {}", config.parquet_file.display()); info!(" • Device: {}", config.device); info!(" • Warmup bars: {}", config.warmup_bars); if let Some(ref output_path) = config.output_json { info!(" • JSON output: {}", output_path.display()); } info!(" • Verbose logging: {}", config.verbose); info!(""); // 1.5: Validate configuration info!("Validating configuration..."); config .validate() .context("Configuration validation failed")?; info!("Configuration validated successfully"); info!(""); // 1.6: Setup graceful shutdown handler let shutdown_flag = Arc::new(AtomicBool::new(false)); let shutdown_clone = shutdown_flag.clone(); tokio::spawn(async move { let ctrl_c = signal::ctrl_c(); #[cfg(unix)] { use tokio::signal::unix::{signal, SignalKind}; let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM handler"); tokio::select! { _ = ctrl_c => { info!("Received Ctrl+C, initiating graceful shutdown..."); } _ = sigterm.recv() => { info!("Received SIGTERM, initiating graceful shutdown..."); } } } #[cfg(not(unix))] { ctrl_c.await.expect("Failed to listen for Ctrl+C"); info!("Received Ctrl+C, initiating graceful shutdown..."); } shutdown_clone.store(true, Ordering::Relaxed); }); info!("Graceful shutdown handler registered (Ctrl+C / SIGTERM)"); info!(""); // Start total elapsed timer let total_start = Instant::now(); // ======================================================================== // PHASE 2: PARALLEL DATA + MODEL LOADING // ======================================================================== info!("Phase 2: Parallel loading (Data + Model)"); info!(""); // Clone paths for async move let parquet_path = config.parquet_file.clone(); let actor_path = config.actor_checkpoint.clone(); let critic_path = config.critic_checkpoint.clone(); let device_str = config.device.clone(); let warmup_bars = config.warmup_bars; // Parallel loading using tokio::try_join! let (data_result, ppo_result) = tokio::try_join!( tokio::task::spawn_blocking(move || { load_parquet_data(&parquet_path, warmup_bars) }), tokio::task::spawn_blocking(move || { load_ppo_model(&actor_path, &critic_path, &device_str) }) ) .context("Parallel loading failed")?; let features = data_result?; let ppo = ppo_result?; info!(""); info!("Phase 2 complete: Data and model loaded in parallel"); info!(""); // ======================================================================== // PHASE 3: SEQUENTIAL INFERENCE // ======================================================================== info!("Phase 3: Sequential inference"); info!(""); // Run inference let inference_results = run_inference(&ppo, features, &shutdown_flag).context("Inference failed")?; // Check if interrupted if shutdown_flag.load(Ordering::Relaxed) { warn!("Evaluation interrupted by shutdown signal"); warn!( " Processed {} bars before interruption", inference_results.len() ); // Still generate partial report info!(""); info!("Generating partial evaluation report..."); if !inference_results.is_empty() { let partial_metrics = calculate_metrics(&inference_results)?; let elapsed = total_start.elapsed(); generate_report(&partial_metrics, &config, elapsed)?; } info!("Partial results saved, safe to terminate"); return Ok(()); } info!(""); info!("Phase 3 complete: Inference finished"); info!(""); // ======================================================================== // PHASE 4: METRICS CALCULATION // ======================================================================== info!("Phase 4: Metrics calculation"); info!(""); let metrics = calculate_metrics(&inference_results).context("Metrics calculation failed")?; info!(""); info!("Phase 4 complete: Metrics calculated"); info!(""); // ======================================================================== // PHASE 5: REPORT GENERATION // ======================================================================== info!("Phase 5: Report generation"); info!(""); let total_elapsed = total_start.elapsed(); generate_report(&metrics, &config, total_elapsed).context("Report generation failed")?; info!(""); info!("Phase 5 complete: Report generated"); info!(""); // ======================================================================== // COMPLETION // ======================================================================== info!("╔══════════════════════════════════════════════════════════════════════╗"); info!("║ EVALUATION COMPLETE ║"); info!("╚══════════════════════════════════════════════════════════════════════╝"); info!(" Total runtime: {:.2}s", total_elapsed.as_secs_f64()); info!(""); Ok(()) }