diff --git a/crates/ml/examples/evaluate_baseline.rs b/crates/ml/examples/evaluate_baseline.rs index 10788ffc9..fd2a34264 100644 --- a/crates/ml/examples/evaluate_baseline.rs +++ b/crates/ml/examples/evaluate_baseline.rs @@ -128,6 +128,7 @@ struct Args { /// /// Expected format: `{ "model_key": { "best_params": { ... }, ... } }` /// Returns `None` if the file doesn't exist or can't be parsed. +#[allow(clippy::cognitive_complexity)] fn load_hyperopt_params(hp_path: &Option, model_key: &str) -> Option { let file_path = hp_path.as_ref()?; if !file_path.exists() { @@ -347,6 +348,7 @@ fn evaluate_dqn_fold( } // Create DQN with same config as training (gamma must match train_baseline) + #[allow(clippy::integer_division)] let config = DQNConfig { state_dim: args.feature_dim, num_actions: args.num_actions, @@ -478,6 +480,7 @@ fn evaluate_ppo_fold( } // Create PPO config matching training + #[allow(clippy::integer_division)] let config = PPOConfig { state_dim: args.feature_dim, num_actions: args.num_actions, @@ -636,7 +639,8 @@ fn run_sanity_checks( // --------------------------------------------------------------------------- #[allow(clippy::cognitive_complexity, clippy::too_many_lines)] -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/evaluate_supervised.rs b/crates/ml/examples/evaluate_supervised.rs index ae93f43a3..fa59fc2f0 100644 --- a/crates/ml/examples/evaluate_supervised.rs +++ b/crates/ml/examples/evaluate_supervised.rs @@ -532,7 +532,8 @@ fn evaluate_fold( // --------------------------------------------------------------------------- #[allow(clippy::cognitive_complexity, clippy::too_many_lines)] -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/hyperopt_baseline_rl.rs b/crates/ml/examples/hyperopt_baseline_rl.rs index 01f7301a5..ac47b9db8 100644 --- a/crates/ml/examples/hyperopt_baseline_rl.rs +++ b/crates/ml/examples/hyperopt_baseline_rl.rs @@ -33,7 +33,7 @@ //! } //! ``` -#![allow(unused_crate_dependencies)] +#![allow(unused_crate_dependencies, unsafe_code)] use anyhow::{Context, Result}; use clap::Parser; @@ -261,7 +261,8 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) } #[allow(clippy::cognitive_complexity)] -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( @@ -275,7 +276,8 @@ fn main() -> Result<()> { metrics_server::start_metrics_server(9094); training_metrics::set_active_workers(1.0); - // Pre-allocate CUBLAS workspace for deterministic + faster tensor core ops + // Pre-allocate CUBLAS workspace for deterministic + faster tensor core ops. + // SAFETY: called once at startup before any multi-threading or CUDA work begins. unsafe { std::env::set_var("CUBLAS_WORKSPACE_CONFIG", ":4096:8"); } let args = Args::parse(); @@ -309,6 +311,7 @@ fn main() -> Result<()> { // Smart cap: on GPU instances with few vCPUs (e.g. L40S-1-48G has 8 vCPU), // more threads than vCPU/2 causes CPU contention that starves GPU. let cpu_cap = cpus.saturating_sub(2).max(1); + #[allow(clippy::integer_division)] let cpu_smart_cap = (cpus / 2).max(1); let parallel = if args.parallel == 0 { cpu_cap.min(cpu_smart_cap) @@ -319,18 +322,18 @@ fn main() -> Result<()> { // Cap by VRAM budget with corrected constants (DQN ~50MB overhead, ~0.0005MB/sample) let budget = ml::hyperopt::HardwareBudget::detect(); let vram_cap = budget.plan_hyperopt(50.0, 0.0005, 64.0, 4096.0).max_concurrent_trials; - let parallel = parallel.min(vram_cap); - info!("Parallel: {} threads (CPU: {}, CPU smart cap: {}, VRAM cap: {})", parallel, cpus, cpu_smart_cap, vram_cap); + let num_threads = parallel.min(vram_cap); + info!("Parallel: {} threads (CPU: {}, CPU smart cap: {}, VRAM cap: {})", num_threads, cpus, cpu_smart_cap, vram_cap); // Configure rayon thread pool for parallel trial evaluation - if parallel > 1 { + if num_threads > 1 { rayon::ThreadPoolBuilder::new() - .num_threads(parallel) + .num_threads(num_threads) .build_global() .unwrap_or_else(|e| { warn!("Failed to set rayon thread pool size: {}", e); }); - info!("Rayon thread pool configured: {} threads", parallel); + info!("Rayon thread pool configured: {} threads", num_threads); } // Scope data directory to symbol subdirectory to avoid mixing instruments @@ -383,7 +386,7 @@ fn main() -> Result<()> { // Run DQN hyperopt if run_dqn { - match run_dqn_hyperopt(&args, parallel, &device) { + match run_dqn_hyperopt(&args, num_threads, &device) { Ok(dqn_result) => { results.insert("dqn".to_owned(), dqn_result); }, @@ -400,7 +403,7 @@ fn main() -> Result<()> { // Run PPO hyperopt if run_ppo { - match run_ppo_hyperopt(&args, parallel, &device) { + match run_ppo_hyperopt(&args, num_threads, &device) { Ok(ppo_result) => { results.insert("ppo".to_owned(), ppo_result); }, diff --git a/crates/ml/examples/hyperopt_baseline_supervised.rs b/crates/ml/examples/hyperopt_baseline_supervised.rs index 742338b02..3a76d2692 100644 --- a/crates/ml/examples/hyperopt_baseline_supervised.rs +++ b/crates/ml/examples/hyperopt_baseline_supervised.rs @@ -543,7 +543,8 @@ const VALID_MODELS: &[&str] = &[ ]; #[allow(clippy::cognitive_complexity)] -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 717211363..72a912cca 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -188,9 +188,10 @@ fn hp_usize(params: &Option, key: &str) -> Option { // Fold data preparation (extracted for prefetching) // --------------------------------------------------------------------------- -/// Prepared fold data: (train_features, val_features, train_bars, val_bars) +/// Prepared fold data: (`train_features`, `val_features`, `train_bars`, `val_bars`) type FoldData = (Vec<[f64; 51]>, Vec<[f64; 51]>, Vec, Vec); +#[allow(clippy::cognitive_complexity)] /// Prepare a fold's data for training: extract features, normalize, align bars. /// /// Returns `None` if feature extraction fails or produces empty features. @@ -268,6 +269,7 @@ fn features_to_trainer_format( /// fold structure stays in this binary; only per-fold training delegates to `DQNTrainer`. /// /// Returns the best validation loss achieved. +#[allow(clippy::cognitive_complexity, clippy::too_many_arguments)] fn train_dqn_fold( fold: usize, train_features: &[[f64; 51]], @@ -400,18 +402,19 @@ fn train_dqn_fold( /// production trainer. The walk-forward fold structure stays in this binary; /// only per-fold training delegates to `PpoTrainer`. /// -/// Returns the best validation loss achieved (approximated by final value_loss). +/// Returns the best validation loss achieved (approximated by final `value_loss`). +#[allow(clippy::cognitive_complexity)] fn train_ppo_fold( fold: usize, train_features: &[[f64; 51]], - _val_features: &[[f64; 51]], + val_features: &[[f64; 51]], train_bars: &[OHLCVBar], _val_bars: &[OHLCVBar], args: &Args, output_dir: &Path, hp: &Option, ) -> Result { - info!(" [PPO] Fold {} -- {} train, {} val features", fold, train_features.len(), _val_features.len()); + info!(" [PPO] Fold {} -- {} train, {} val features", fold, train_features.len(), val_features.len()); let n_train = train_features.len(); if n_train < 2 { @@ -652,7 +655,7 @@ fn run_training(args: &Args) -> Result> { .spawn(move || { info!(" [Prefetch] Loading fold {} data on background thread", nw.fold); let result = prepare_fold_data(&nw, &out_dir); - let _ = tx.send(result); + drop(tx.send(result)); }); Some(rx) } else { @@ -824,7 +827,8 @@ fn run_training(args: &Args) -> Result> { // Main // --------------------------------------------------------------------------- -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/train_baseline_supervised.rs b/crates/ml/examples/train_baseline_supervised.rs index ee8c54c16..5203bfad9 100644 --- a/crates/ml/examples/train_baseline_supervised.rs +++ b/crates/ml/examples/train_baseline_supervised.rs @@ -796,7 +796,8 @@ fn run_training(args: &Args) -> Result> { // Main // --------------------------------------------------------------------------- -fn main() -> Result<()> { +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability(