optirs_core/streaming/adaptive_streaming/
anomaly_ensemble.rs

1// Ensemble voting for the streaming anomaly detector.
2//
3// Each strategy computes the quantity it names. Two of them used to fall
4// through to a `_` arm and silently behave as plain min-consensus voting; that
5// is gone. `Adaptive` is now backed by real per-detector confusion counters fed
6// from the ground-truth feedback path, and `Stacking` — the one strategy that
7// genuinely needs a trained second-stage model — returns an honest error.
8
9use super::anomaly_detection::{AnomalyDetectionResult, AnomalySeverity, AnomalyType};
10use super::anomaly_scoring::DetectionCounters;
11
12use crate::utils::try_scalar_str;
13use scirs2_core::numeric::Float;
14use std::collections::HashMap;
15
16/// Ensemble anomaly detector combining multiple methods
17pub struct EnsembleAnomalyDetector<A: Float + Send + Sync> {
18    /// Ensemble voting strategy
19    voting_strategy: EnsembleVotingStrategy,
20    /// Per-detector weights used by
21    /// [`EnsembleVotingStrategy::Weighted`]. Detectors with no explicit weight
22    /// count as `1`, so an unconfigured ensemble weights every detector
23    /// equally rather than ignoring them.
24    detector_weights: HashMap<String, A>,
25    /// Per-detector confusion counters, the measurement
26    /// [`EnsembleVotingStrategy::Adaptive`] derives its weights from.
27    ///
28    /// These are the ensemble's own counters, not the ML detectors': the ML
29    /// detectors are told the *ensemble's* verdict when ground truth arrives
30    /// (that is what "the outcome of the most recent detection" means to them),
31    /// and the statistical detectors carry no counters at all. Weighting a vote
32    /// needs each member's own verdict scored against the truth, which is
33    /// exactly what these record.
34    detector_counters: HashMap<String, DetectionCounters>,
35    /// Each member's verdict on the most recently combined point, awaiting
36    /// ground truth. Cleared once labelled, so one detection is scored once.
37    pending_verdicts: HashMap<String, bool>,
38    /// Ensemble configuration
39    ensemble_config: EnsembleConfig<A>,
40}
41
42/// Ensemble voting strategies
43#[derive(Debug, Clone)]
44pub enum EnsembleVotingStrategy {
45    /// Simple majority voting
46    Majority,
47    /// Weighted voting using the operator-supplied per-detector weights
48    Weighted,
49    /// Maximum anomaly score
50    MaxScore,
51    /// Average anomaly score
52    AverageScore,
53    /// Median anomaly score
54    MedianScore,
55    /// Weighted voting whose weights are *measured* from each detector's own
56    /// confusion matrix (balanced accuracy), rather than configured.
57    Adaptive,
58    /// Stacking with meta-learner
59    Stacking,
60}
61
62/// Ensemble configuration
63#[derive(Debug, Clone)]
64pub struct EnsembleConfig<A: Float + Send + Sync> {
65    /// Minimum number of detectors that must agree
66    pub min_consensus: usize,
67    /// Threshold for ensemble anomaly score
68    pub ensemble_threshold: A,
69    /// Enable dynamic detector weighting
70    pub dynamic_weighting: bool,
71    /// Performance evaluation window
72    pub evaluation_window: usize,
73    /// Enable detector selection based on context
74    pub context_based_selection: bool,
75}
76
77impl<A: Float + Default + Clone + Send + Sync + std::iter::Sum> EnsembleAnomalyDetector<A> {
78    pub(super) fn new(voting_strategy: EnsembleVotingStrategy) -> Result<Self, String> {
79        Ok(Self {
80            voting_strategy,
81            detector_weights: HashMap::new(),
82            detector_counters: HashMap::new(),
83            pending_verdicts: HashMap::new(),
84            ensemble_config: EnsembleConfig {
85                min_consensus: 2,
86                ensemble_threshold: try_scalar_str::<A, _>(0.5)?,
87                dynamic_weighting: true,
88                evaluation_window: 100,
89                context_based_selection: false,
90            },
91        })
92    }
93
94    /// Replaces the voting strategy.
95    pub(super) fn set_voting_strategy(&mut self, strategy: EnsembleVotingStrategy) {
96        self.voting_strategy = strategy;
97    }
98
99    /// Strategy currently in force.
100    pub fn voting_strategy(&self) -> &EnsembleVotingStrategy {
101        &self.voting_strategy
102    }
103
104    /// Sets the weight [`EnsembleVotingStrategy::Weighted`] gives one detector.
105    pub(super) fn set_detector_weight(&mut self, detector_name: &str, weight: A) {
106        self.detector_weights
107            .insert(detector_name.to_string(), weight);
108    }
109
110    /// Records ground truth for the most recently combined point against every
111    /// member's own verdict.
112    ///
113    /// This is the ground-truth feedback path the adaptive weighting needs: a
114    /// detector that flagged the point is scored as a true or false positive,
115    /// one that did not is scored as a true or false negative.
116    pub(super) fn record_outcome(&mut self, was_true_anomaly: bool) {
117        for (name, verdict) in self.pending_verdicts.drain() {
118            self.detector_counters
119                .entry(name)
120                .or_default()
121                .record_outcome(verdict, was_true_anomaly);
122        }
123    }
124
125    /// Balanced accuracy measured for one member, or `None` before any ground
126    /// truth has been recorded for it.
127    pub fn detector_balanced_accuracy(&self, detector_name: &str) -> Option<f64> {
128        self.detector_counters
129            .get(detector_name)?
130            .balanced_accuracy()
131    }
132
133    /// Weights derived from the measured per-detector confusion matrices, or
134    /// `None` while no member has any labelled outcome.
135    ///
136    /// The weight is balanced accuracy `(TPR + TNR) / 2`, which is insensitive
137    /// to how rare anomalies are; plain accuracy would hand a detector that
138    /// flags nothing on a 1%-anomaly stream a weight of 0.99.
139    pub fn measured_weights(&self) -> Option<HashMap<String, A>> {
140        let mut weights = HashMap::new();
141        let mut measured_any = false;
142        for (name, counters) in &self.detector_counters {
143            if let Some(balanced) = counters.balanced_accuracy() {
144                measured_any = true;
145                if let Some(weight) = A::from(balanced) {
146                    weights.insert(name.clone(), weight);
147                }
148            }
149        }
150        measured_any.then_some(weights)
151    }
152
153    /// Weighted vote and weighted mean score under an arbitrary weighting.
154    ///
155    /// Shared by [`EnsembleVotingStrategy::Weighted`] (configured weights) and
156    /// [`EnsembleVotingStrategy::Adaptive`] (measured weights), so the two
157    /// cannot drift apart.
158    fn weighted_vote(
159        results: &HashMap<String, AnomalyDetectionResult<A>>,
160        weight_of: &dyn Fn(&str) -> A,
161    ) -> Result<(bool, A), String> {
162        let mut weight_sum = A::zero();
163        let mut weighted_score = A::zero();
164        let mut weighted_votes = A::zero();
165        for (name, result) in results {
166            let weight = weight_of(name);
167            weight_sum = weight_sum + weight;
168            weighted_score = weighted_score + weight * result.anomaly_score;
169            if result.is_anomaly {
170                weighted_votes = weighted_votes + weight;
171            }
172        }
173        if weight_sum <= A::zero() {
174            return Err(
175                "weighted ensemble voting needs a positive total detector weight".to_string(),
176            );
177        }
178        let score = weighted_score / weight_sum;
179        let vote_share = weighted_votes / weight_sum;
180        Ok((vote_share > try_scalar_str::<A, _>(0.5)?, score))
181    }
182
183    pub(super) fn combine_results(
184        &mut self,
185        results: HashMap<String, AnomalyDetectionResult<A>>,
186    ) -> Result<AnomalyDetectionResult<A>, String> {
187        if results.is_empty() {
188            return Ok(AnomalyDetectionResult {
189                is_anomaly: false,
190                anomaly_score: A::zero(),
191                confidence: A::zero(),
192                anomaly_type: None,
193                severity: AnomalySeverity::Low,
194                metadata: HashMap::new(),
195            });
196        }
197
198        // Record each member's own verdict, so that a later
199        // `record_outcome` can score it against the truth. This is the only
200        // place the ensemble sees the individual verdicts.
201        self.pending_verdicts.clear();
202        for (name, result) in &results {
203            self.detector_counters
204                .entry(name.clone())
205                .or_default()
206                .record_prediction(result.is_anomaly, result.anomaly_score);
207            self.pending_verdicts
208                .insert(name.clone(), result.is_anomaly);
209        }
210
211        let anomaly_count = results.values().filter(|r| r.is_anomaly).count();
212        let total_count = results.len();
213
214        let avg_score = results.values().map(|r| r.anomaly_score).sum::<A>()
215            / try_scalar_str::<A, _>(total_count)?;
216        let avg_confidence = results.values().map(|r| r.confidence).sum::<A>()
217            / try_scalar_str::<A, _>(total_count)?;
218
219        // The ensemble the detector actually builds is `Weighted`, which used
220        // to fall through to the `_` arm and behave as plain min-consensus
221        // voting -- the weights were never consulted at all. Each strategy now
222        // computes the quantity it names, and the one that has no
223        // implementation behind it says so instead of silently pretending to
224        // be a different strategy.
225        let threshold = self.ensemble_config.ensemble_threshold;
226        let (is_anomaly, ensemble_score) = match self.voting_strategy {
227            EnsembleVotingStrategy::Majority => (anomaly_count > total_count / 2, avg_score),
228            EnsembleVotingStrategy::MaxScore => {
229                let max_score = results
230                    .values()
231                    .map(|r| r.anomaly_score)
232                    .fold(A::zero(), |acc, s| if s > acc { s } else { acc });
233                (max_score > threshold, max_score)
234            }
235            EnsembleVotingStrategy::AverageScore => (avg_score > threshold, avg_score),
236            EnsembleVotingStrategy::MedianScore => {
237                let mut scores: Vec<A> = results.values().map(|r| r.anomaly_score).collect();
238                scores.sort_by(crate::utils::total_order);
239                let median = if scores.len().is_multiple_of(2) {
240                    (scores[scores.len() / 2 - 1] + scores[scores.len() / 2])
241                        / try_scalar_str::<A, _>(2.0)?
242                } else {
243                    scores[scores.len() / 2]
244                };
245                (median > threshold, median)
246            }
247            EnsembleVotingStrategy::Weighted => {
248                let one = A::one();
249                let weights = &self.detector_weights;
250                Self::weighted_vote(&results, &|name| *weights.get(name).unwrap_or(&one))?
251            }
252            EnsembleVotingStrategy::Adaptive => {
253                // Measured weights when ground truth has arrived; uniform
254                // otherwise. Uniform is the honest default: before any label
255                // exists there is no evidence that one member deserves more say
256                // than another, and inventing a prior ranking would be a
257                // fabrication dressed as a measurement.
258                match self.measured_weights() {
259                    Some(measured) if measured.values().any(|w| *w > A::zero()) => {
260                        Self::weighted_vote(&results, &|name| {
261                            measured.get(name).copied().unwrap_or_else(A::zero)
262                        })?
263                    }
264                    // Either no ground truth yet, or every member has measured
265                    // zero skill — in both cases there is nothing to rank the
266                    // members by, so the vote stays uniform.
267                    _ => Self::weighted_vote(&results, &|_| A::one())?,
268                }
269            }
270            EnsembleVotingStrategy::Stacking => {
271                return Err(
272                    "ensemble voting strategy `Stacking` is not implemented: stacking \
273                     feeds the member scores into a second-stage model trained on held-out \
274                     labelled data, and this detector carries no meta-learner and no \
275                     held-out split to train one on. Use `Adaptive`, which weights the \
276                     members by their measured balanced accuracy from the same feedback \
277                     path"
278                        .to_string(),
279                );
280            }
281        };
282        let avg_score = ensemble_score;
283
284        Ok(AnomalyDetectionResult {
285            is_anomaly,
286            anomaly_score: avg_score,
287            confidence: avg_confidence,
288            anomaly_type: if is_anomaly {
289                Some(AnomalyType::StatisticalOutlier)
290            } else {
291                None
292            },
293            severity: if avg_score > try_scalar_str::<A, _>(0.8)? {
294                AnomalySeverity::High
295            } else if avg_score > try_scalar_str::<A, _>(0.5)? {
296                AnomalySeverity::Medium
297            } else {
298                AnomalySeverity::Low
299            },
300            metadata: HashMap::new(),
301        })
302    }
303
304    pub(super) fn adjust_sensitivity(&mut self, adjustment: A) -> Result<(), String> {
305        self.ensemble_config.ensemble_threshold = (self.ensemble_config.ensemble_threshold
306            + adjustment)
307            .max(try_scalar_str::<A, _>(0.1)?)
308            .min(try_scalar_str::<A, _>(0.9)?);
309        Ok(())
310    }
311}
312
313#[cfg(test)]
314#[path = "anomaly_ensemble_tests.rs"]
315mod anomaly_ensemble_tests;
optirs_core/streaming/adaptive_streaming/anomaly_ensemble.rs

optirs_core/streaming/adaptive_streaming/
anomaly_ensemble.rs
