From 1a6834fff2bb1551cd8b2bd03c637b3334246b4d Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Sun, 1 Mar 2026 21:09:10 +0100
Subject: [PATCH] =?UTF-8?q?feat(ml):=20parallel=20ensemble=20inference=20v?=
 =?UTF-8?q?ia=20rayon=20=E2=80=94=20sub-ms=20multi-model=20predictions?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace sequential for-loop over ModelInferenceAdapters with rayon
par_iter(). Each adapter's predict() runs on a separate thread,
then results are aggregated sequentially (fast arithmetic).

ModelInferenceAdapter: Send + Sync makes this safe for parallel execution.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 crates/ml/src/ensemble/inference_ensemble.rs | 85 +++++++++++---------
 1 file changed, 48 insertions(+), 37 deletions(-)

diff --git a/crates/ml/src/ensemble/inference_ensemble.rs b/crates/ml/src/ensemble/inference_ensemble.rs
index 359bf267c..bc72f9ba1 100644
--- a/crates/ml/src/ensemble/inference_ensemble.rs
+++ b/crates/ml/src/ensemble/inference_ensemble.rs
@@ -6,6 +6,8 @@
 
 use std::collections::HashMap;
 
+use rayon::prelude::*;
+
 use crate::ensemble::inference_adapter::{
     EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta,
 };
@@ -66,50 +68,59 @@ impl InferenceEnsemble {
             ));
         }
 
+        // Run inference in parallel across all ready adapters
+        let results: Vec<(String, EnsemblePrediction)> = ready_adapters
+            .par_iter()
+            .filter_map(|adapter| {
+                let model_name = adapter.model_name().to_string();
+                match adapter.predict(features) {
+                    Ok(pred) => {
+                        if !pred.direction.is_finite() || !pred.confidence.is_finite() {
+                            tracing::warn!(
+                                model = %model_name,
+                                direction = %pred.direction,
+                                confidence = %pred.confidence,
+                                "Model returned NaN/Inf prediction, skipping (circuit breaker)"
+                            );
+                            None
+                        } else {
+                            Some((model_name, pred))
+                        }
+                    }
+                    Err(e) => {
+                        tracing::warn!(
+                            model = %model_name,
+                            error = %e,
+                            "Model prediction failed, skipping"
+                        );
+                        None
+                    }
+                }
+            })
+            .collect();
+
+        // Aggregate results sequentially (fast arithmetic)
         let mut weighted_direction_sum = 0.0_f64;
         let mut weight_confidence_sum = 0.0_f64;
         let mut confidence_sum = 0.0_f64;
-        let mut successful_count = 0_usize;
         let mut model_names: Vec<String> = Vec::new();
 
-        for adapter in &ready_adapters {
-            let model_name = adapter.model_name().to_string();
-            match adapter.predict(features) {
-                Ok(pred) => {
-                    // Circuit breaker: skip NaN/Inf predictions
-                    if !pred.direction.is_finite() || !pred.confidence.is_finite() {
-                        tracing::warn!(
-                            model = %model_name,
-                            direction = %pred.direction,
-                            confidence = %pred.confidence,
-                            "Model returned NaN/Inf prediction, skipping (circuit breaker)"
-                        );
-                        continue;
-                    }
-                    // Clamp confidence to valid range
-                    let confidence = pred.confidence.clamp(0.0, 1.0);
-                    let w = self
-                        .weights
-                        .get(&model_name)
-                        .copied()
-                        .unwrap_or(1.0);
-                    let wc = w * confidence;
-                    weighted_direction_sum += pred.direction * wc;
-                    weight_confidence_sum += wc;
-                    confidence_sum += confidence;
-                    successful_count += 1;
-                    model_names.push(model_name);
-                }
-                Err(e) => {
-                    tracing::warn!(
-                        model = %model_name,
-                        error = %e,
-                        "Model prediction failed, skipping"
-                    );
-                }
-            }
+        for (model_name, pred) in &results {
+            let confidence = pred.confidence.clamp(0.0, 1.0);
+            let w = self
+                .weights
+                .get(model_name)
+                .copied()
+                .unwrap_or(1.0);
+            let wc = w * confidence;
+            weighted_direction_sum += pred.direction * wc;
+            weight_confidence_sum += wc;
+            confidence_sum += confidence;
+            model_names.push(model_name.clone());
         }
 
+        let successful_count = results.len();
+
         if successful_count == 0 {
             return Err(MLError::InferenceError(
                 "All ready models failed during inference".to_owned(),