use crate::math::u64_f64;
#[cfg(not(feature = "std"))]
use crate::math::FloatExt;
pub const DEFAULT_WINDOW: usize = 100;
#[derive(Clone, Debug)]
pub struct StreamingMedian {
buffer: [f64; DEFAULT_WINDOW],
sorted: [f64; DEFAULT_WINDOW],
pos: usize,
count: usize,
}
impl StreamingMedian {
pub const WINDOW: usize = DEFAULT_WINDOW;
#[must_use]
pub const fn new() -> Self {
Self {
buffer: [0.0; DEFAULT_WINDOW],
sorted: [0.0; DEFAULT_WINDOW],
pos: 0,
count: 0,
}
}
#[inline]
fn binary_search_insert(&self, value: f64, len: usize) -> usize {
let mut left = 0;
let mut right = len;
while left < right {
let mid = left + (right - left) / 2;
if self.sorted[mid] < value {
left = mid + 1;
} else {
right = mid;
}
}
left
}
#[inline]
fn binary_search_remove(&self, value: f64, len: usize) -> Option<usize> {
let mut left = 0;
let mut right = len;
while left < right {
let mid = left + (right - left) / 2;
if self.sorted[mid] < value {
left = mid + 1;
} else if self.sorted[mid] > value {
right = mid;
} else {
return Some(mid);
}
}
if left < len && (self.sorted[left] - value).abs() < f64::EPSILON {
return Some(left);
}
if left > 0 && (self.sorted[left - 1] - value).abs() < f64::EPSILON {
return Some(left - 1);
}
None
}
#[inline]
pub fn push(&mut self, value: f64) {
if self.count < DEFAULT_WINDOW {
let insert_pos = self.binary_search_insert(value, self.count);
if insert_pos < self.count {
self.sorted
.copy_within(insert_pos..self.count, insert_pos + 1);
}
self.sorted[insert_pos] = value;
self.buffer[self.pos] = value;
self.pos = (self.pos + 1) % DEFAULT_WINDOW;
self.count += 1;
} else {
let old_value = self.buffer[self.pos];
if let Some(remove_pos) = self.binary_search_remove(old_value, self.count) {
if remove_pos < self.count - 1 {
self.sorted
.copy_within(remove_pos + 1..self.count, remove_pos);
}
}
let insert_pos = self.binary_search_insert(value, self.count - 1);
if insert_pos < self.count - 1 {
self.sorted
.copy_within(insert_pos..self.count - 1, insert_pos + 1);
}
self.sorted[insert_pos] = value;
self.buffer[self.pos] = value;
self.pos = (self.pos + 1) % DEFAULT_WINDOW;
}
}
#[inline]
pub fn median(&mut self) -> f64 {
if self.count == 0 {
return 0.0;
}
let mid = self.count / 2;
if self.count.is_multiple_of(2) {
f64::midpoint(self.sorted[mid - 1], self.sorted[mid])
} else {
self.sorted[mid]
}
}
#[inline]
#[must_use]
pub const fn count(&self) -> usize {
self.count
}
#[inline]
#[must_use]
pub const fn is_full(&self) -> bool {
self.count >= DEFAULT_WINDOW
}
pub const fn clear(&mut self) {
self.buffer = [0.0; DEFAULT_WINDOW];
self.sorted = [0.0; DEFAULT_WINDOW];
self.pos = 0;
self.count = 0;
}
}
impl Default for StreamingMedian {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone, Debug)]
pub struct MadDetector {
values_median: StreamingMedian,
deviations_median: StreamingMedian,
recent_values: [f64; DEFAULT_WINDOW],
pos: usize,
count: usize,
threshold_k: f64,
cached_median: f64,
cached_mad: f64,
cache_valid: bool,
}
impl MadDetector {
const MAD_SCALE: f64 = 1.4826;
#[must_use]
pub const fn new(threshold_k: f64) -> Self {
Self {
values_median: StreamingMedian::new(),
deviations_median: StreamingMedian::new(),
recent_values: [0.0; DEFAULT_WINDOW],
pos: 0,
count: 0,
threshold_k,
cached_median: 0.0,
cached_mad: 0.0,
cache_valid: false,
}
}
pub fn observe(&mut self, value: f64) {
self.recent_values[self.pos] = value;
self.pos = (self.pos + 1) % DEFAULT_WINDOW;
if self.count < DEFAULT_WINDOW {
self.count += 1;
}
self.values_median.push(value);
self.cache_valid = false;
}
fn update_cache(&mut self) {
if self.cache_valid {
return;
}
self.cached_median = self.values_median.median();
self.deviations_median.clear();
for i in 0..self.count {
let dev = (self.recent_values[i] - self.cached_median).abs();
self.deviations_median.push(dev);
}
self.cached_mad = self.deviations_median.median();
self.cache_valid = true;
}
pub fn is_anomaly(&mut self, value: f64) -> bool {
if self.count < 3 {
return false;
}
self.update_cache();
if self.cached_mad < 1e-10 {
return (value - self.cached_median).abs() > 1e-10;
}
let deviation = (value - self.cached_median).abs();
let threshold = self.threshold_k * self.cached_mad * Self::MAD_SCALE;
deviation > threshold
}
pub fn anomaly_score(&mut self, value: f64) -> f64 {
if self.count < 3 {
return 0.0;
}
self.update_cache();
if self.cached_mad < 1e-10 {
return if (value - self.cached_median).abs() > 1e-10 {
f64::INFINITY
} else {
0.0
};
}
let inv_mad_scale = 1.0 / (self.cached_mad * Self::MAD_SCALE);
(value - self.cached_median).abs() * inv_mad_scale
}
pub fn median(&mut self) -> f64 {
self.update_cache();
self.cached_median
}
pub fn mad(&mut self) -> f64 {
self.update_cache();
self.cached_mad
}
#[inline]
#[must_use]
pub const fn threshold_k(&self) -> f64 {
self.threshold_k
}
#[inline]
pub const fn set_threshold_k(&mut self, k: f64) {
self.threshold_k = k;
}
#[inline]
#[must_use]
pub const fn count(&self) -> usize {
self.count
}
pub const fn clear(&mut self) {
self.values_median.clear();
self.deviations_median.clear();
self.recent_values = [0.0; DEFAULT_WINDOW];
self.pos = 0;
self.count = 0;
self.cache_valid = false;
}
}
#[derive(Clone, Debug)]
pub struct EwmaDetector {
alpha: f64,
ewma: f64,
ewma_var: f64,
threshold_k: f64,
initialized: bool,
count: u64,
}
impl EwmaDetector {
#[must_use]
pub const fn new(alpha: f64, threshold_k: f64) -> Self {
assert!(
alpha > 0.0 && alpha <= 1.0,
"EwmaDetector alpha must be in (0.0, 1.0]"
);
Self {
alpha,
ewma: 0.0,
ewma_var: 0.0,
threshold_k,
initialized: false,
count: 0,
}
}
pub fn observe(&mut self, value: f64) {
self.count += 1;
if !self.initialized {
self.ewma = value;
self.ewma_var = 0.0;
self.initialized = true;
return;
}
let deviation = value - self.ewma;
self.ewma += self.alpha * deviation;
self.ewma_var =
(1.0 - self.alpha) * (self.alpha * deviation).mul_add(deviation, self.ewma_var);
}
#[must_use]
pub fn is_anomaly(&self, value: f64) -> bool {
if !self.initialized || self.count < 3 {
return false;
}
let std_dev = self.ewma_var.sqrt();
if std_dev < 1e-10 {
return (value - self.ewma).abs() > 1e-10;
}
let deviation = (value - self.ewma).abs();
deviation > self.threshold_k * std_dev
}
#[must_use]
pub fn anomaly_score(&self, value: f64) -> f64 {
if !self.initialized {
return 0.0;
}
let std_dev = self.ewma_var.sqrt();
if std_dev < 1e-10 {
return if (value - self.ewma).abs() > 1e-10 {
f64::INFINITY
} else {
0.0
};
}
let inv_std = 1.0 / std_dev;
(value - self.ewma).abs() * inv_std
}
#[inline]
#[must_use]
pub const fn ewma(&self) -> f64 {
self.ewma
}
#[inline]
#[must_use]
pub fn std_dev(&self) -> f64 {
self.ewma_var.sqrt()
}
#[inline]
#[must_use]
pub const fn alpha(&self) -> f64 {
self.alpha
}
#[inline]
pub const fn set_alpha(&mut self, alpha: f64) {
assert!(
alpha > 0.0 && alpha <= 1.0,
"EwmaDetector alpha must be in (0.0, 1.0]"
);
self.alpha = alpha;
}
#[inline]
#[must_use]
pub const fn threshold_k(&self) -> f64 {
self.threshold_k
}
#[inline]
pub const fn set_threshold_k(&mut self, k: f64) {
self.threshold_k = k;
}
#[inline]
#[must_use]
pub const fn count(&self) -> u64 {
self.count
}
pub const fn reset(&mut self) {
self.ewma = 0.0;
self.ewma_var = 0.0;
self.initialized = false;
self.count = 0;
}
}
#[derive(Clone, Debug)]
pub struct ZScoreDetector {
mean: f64,
m2: f64,
count: u64,
threshold_k: f64,
}
impl ZScoreDetector {
#[must_use]
pub const fn new(threshold_k: f64) -> Self {
Self {
mean: 0.0,
m2: 0.0,
count: 0,
threshold_k,
}
}
pub fn observe(&mut self, value: f64) {
self.count += 1;
let inv_count = 1.0 / u64_f64(self.count);
let delta = value - self.mean;
self.mean += delta * inv_count;
let delta2 = value - self.mean;
self.m2 += delta * delta2;
}
#[inline]
#[must_use]
pub fn variance(&self) -> f64 {
if self.count < 2 {
0.0
} else {
self.m2 / u64_f64(self.count - 1)
}
}
#[inline]
#[must_use]
pub fn std_dev(&self) -> f64 {
self.variance().sqrt()
}
#[must_use]
pub fn is_anomaly(&self, value: f64) -> bool {
if self.count < 3 {
return false;
}
let std_dev = self.std_dev();
if std_dev < 1e-10 {
return (value - self.mean).abs() > 1e-10;
}
let inv_std = 1.0 / std_dev;
let z_score = (value - self.mean).abs() * inv_std;
z_score > self.threshold_k
}
#[must_use]
pub fn z_score(&self, value: f64) -> f64 {
let std_dev = self.std_dev();
if std_dev < 1e-10 {
return if (value - self.mean).abs() > 1e-10 {
f64::INFINITY
} else {
0.0
};
}
let inv_std = 1.0 / std_dev;
(value - self.mean) * inv_std
}
#[inline]
#[must_use]
pub const fn mean(&self) -> f64 {
self.mean
}
#[inline]
#[must_use]
pub const fn count(&self) -> u64 {
self.count
}
pub const fn reset(&mut self) {
self.mean = 0.0;
self.m2 = 0.0;
self.count = 0;
}
}
#[derive(Clone, Debug)]
pub struct AnomalyEvent {
pub value: f64,
pub score: f64,
pub expected: f64,
pub timestamp: u64,
pub metric_id: u64,
}
pub trait AnomalyCallback {
fn on_anomaly(&mut self, event: AnomalyEvent);
}
#[derive(Clone, Debug)]
pub struct CompositeDetector {
pub mad: MadDetector,
pub ewma: EwmaDetector,
pub zscore: ZScoreDetector,
pub require_consensus: bool,
}
impl CompositeDetector {
#[must_use]
pub const fn new() -> Self {
Self {
mad: MadDetector::new(3.0),
ewma: EwmaDetector::new(0.1, 3.0),
zscore: ZScoreDetector::new(3.0),
require_consensus: false,
}
}
#[must_use]
pub const fn with_thresholds(mad_k: f64, ewma_alpha: f64, ewma_k: f64, zscore_k: f64) -> Self {
Self {
mad: MadDetector::new(mad_k),
ewma: EwmaDetector::new(ewma_alpha, ewma_k),
zscore: ZScoreDetector::new(zscore_k),
require_consensus: false,
}
}
pub fn observe(&mut self, value: f64) {
self.mad.observe(value);
self.ewma.observe(value);
self.zscore.observe(value);
}
pub fn is_anomaly(&mut self, value: f64) -> bool {
let mad_anomaly = self.mad.is_anomaly(value);
let ewma_anomaly = self.ewma.is_anomaly(value);
let zscore_anomaly = self.zscore.is_anomaly(value);
if self.require_consensus {
mad_anomaly && ewma_anomaly && zscore_anomaly
} else {
mad_anomaly || ewma_anomaly || zscore_anomaly
}
}
pub fn anomaly_score(&mut self, value: f64) -> f64 {
let mad_score = self.mad.anomaly_score(value);
let ewma_score = self.ewma.anomaly_score(value);
let zscore_score = self.zscore.z_score(value).abs();
mad_score.max(ewma_score).max(zscore_score)
}
#[inline]
#[must_use]
pub const fn count(&self) -> u64 {
self.zscore.count()
}
pub const fn reset(&mut self) {
self.mad.clear();
self.ewma.reset();
self.zscore.reset();
}
}
impl Default for CompositeDetector {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
#[allow(clippy::float_cmp)]
mod tests {
use super::*;
use crate::math::usize_f64;
#[test]
fn test_streaming_median() {
let mut sm = StreamingMedian::new();
for v in [5.0, 2.0, 8.0, 1.0, 9.0] {
sm.push(v);
}
let median = sm.median();
assert_eq!(median, 5.0); }
#[test]
fn test_streaming_median_even() {
let mut sm = StreamingMedian::new();
for v in [1.0, 2.0, 3.0, 4.0] {
sm.push(v);
}
let median = sm.median();
assert_eq!(median, 2.5); }
#[test]
fn test_mad_detector_normal() {
let mut detector = MadDetector::new(3.0);
for v in [10.0, 10.5, 9.5, 10.2, 9.8, 10.1, 9.9, 10.3, 9.7, 10.4] {
detector.observe(v);
}
assert!(!detector.is_anomaly(10.0));
assert!(!detector.is_anomaly(10.5));
assert!(!detector.is_anomaly(9.5));
assert!(detector.is_anomaly(100.0));
assert!(detector.is_anomaly(-50.0));
}
#[test]
fn test_ewma_detector() {
let mut detector = EwmaDetector::new(0.1, 3.0);
for i in 0..100 {
let value = 10.0 + f64::from(i % 3) - 1.0; detector.observe(value);
}
assert!((detector.ewma() - 10.0).abs() < 1.0);
assert!(!detector.is_anomaly(10.0));
assert!(!detector.is_anomaly(11.0));
assert!(detector.is_anomaly(50.0));
}
#[test]
fn test_zscore_detector() {
let mut detector = ZScoreDetector::new(3.0);
for v in [10.0, 11.0, 9.0, 10.5, 9.5, 10.2, 9.8, 10.1, 9.9, 10.3] {
detector.observe(v);
}
assert!((detector.mean() - 10.03).abs() < 0.01);
assert!(!detector.is_anomaly(10.0));
assert!(detector.is_anomaly(100.0));
}
#[test]
fn test_composite_detector() {
let mut detector = CompositeDetector::new();
for v in [10.0, 10.5, 9.5, 10.2, 9.8, 10.1, 9.9, 10.3, 9.7, 10.4] {
detector.observe(v);
}
assert!(!detector.is_anomaly(10.0));
assert!(detector.is_anomaly(100.0));
let score = detector.anomaly_score(100.0);
assert!(score > 3.0);
}
#[test]
fn test_mad_detector_constant() {
let mut detector = MadDetector::new(3.0);
for _ in 0..10 {
detector.observe(5.0);
}
assert!(!detector.is_anomaly(5.0));
assert!(detector.is_anomaly(5.1));
}
#[test]
fn test_streaming_median_default() {
let sm = StreamingMedian::default();
assert_eq!(sm.count(), 0);
assert!(!sm.is_full());
}
#[test]
fn test_streaming_median_empty() {
let mut sm = StreamingMedian::new();
assert_eq!(sm.median(), 0.0);
}
#[test]
fn test_streaming_median_clear() {
let mut sm = StreamingMedian::new();
sm.push(1.0);
sm.push(2.0);
assert_eq!(sm.count(), 2);
sm.clear();
assert_eq!(sm.count(), 0);
}
#[test]
fn test_streaming_median_full() {
let mut sm = StreamingMedian::new();
for i in 0..DEFAULT_WINDOW {
sm.push(usize_f64(i));
}
assert!(sm.is_full());
assert_eq!(sm.count(), DEFAULT_WINDOW);
}
#[test]
fn test_streaming_median_wrap() {
let mut sm = StreamingMedian::new();
for i in 0..(DEFAULT_WINDOW + 10) {
sm.push(usize_f64(i));
}
assert_eq!(sm.count(), DEFAULT_WINDOW);
let median = sm.median();
assert!(median > 0.0);
}
#[test]
fn test_mad_detector_insufficient_data() {
let mut detector = MadDetector::new(3.0);
detector.observe(10.0);
detector.observe(20.0);
assert!(!detector.is_anomaly(1000.0));
}
#[test]
fn test_mad_detector_count() {
let mut detector = MadDetector::new(3.0);
assert_eq!(detector.count(), 0);
detector.observe(1.0);
assert_eq!(detector.count(), 1);
}
#[test]
fn test_mad_detector_threshold_k() {
let detector = MadDetector::new(5.0);
assert!((detector.threshold_k() - 5.0).abs() < f64::EPSILON);
}
#[test]
fn test_mad_detector_set_threshold() {
let mut detector = MadDetector::new(3.0);
detector.set_threshold_k(5.0);
assert!((detector.threshold_k() - 5.0).abs() < f64::EPSILON);
}
#[test]
fn test_mad_detector_clear() {
let mut detector = MadDetector::new(3.0);
for v in [10.0, 11.0, 9.0, 10.5] {
detector.observe(v);
}
detector.clear();
assert_eq!(detector.count(), 0);
}
#[test]
fn test_mad_detector_anomaly_score() {
let mut detector = MadDetector::new(3.0);
for v in [10.0, 10.5, 9.5, 10.2, 9.8] {
detector.observe(v);
}
let score_normal = detector.anomaly_score(10.0);
let score_extreme = detector.anomaly_score(100.0);
assert!(score_extreme > score_normal);
}
#[test]
fn test_mad_detector_median_and_mad() {
let mut detector = MadDetector::new(3.0);
for v in [1.0, 2.0, 3.0, 4.0, 5.0] {
detector.observe(v);
}
let median = detector.median();
assert!((median - 3.0).abs() < 0.01);
let mad = detector.mad();
assert!(mad > 0.0);
}
#[test]
fn test_ewma_detector_not_initialized() {
let detector = EwmaDetector::new(0.1, 3.0);
assert!(!detector.is_anomaly(100.0));
assert_eq!(detector.anomaly_score(100.0), 0.0);
}
#[test]
fn test_ewma_detector_count() {
let mut detector = EwmaDetector::new(0.1, 3.0);
assert_eq!(detector.count(), 0);
detector.observe(10.0);
assert_eq!(detector.count(), 1);
}
#[test]
fn test_ewma_detector_alpha() {
let detector = EwmaDetector::new(0.2, 3.0);
assert!((detector.alpha() - 0.2).abs() < f64::EPSILON);
}
#[test]
fn test_ewma_detector_set_alpha() {
let mut detector = EwmaDetector::new(0.1, 3.0);
detector.set_alpha(0.5);
assert!((detector.alpha() - 0.5).abs() < f64::EPSILON);
}
#[test]
fn test_ewma_detector_set_threshold() {
let mut detector = EwmaDetector::new(0.1, 3.0);
detector.set_threshold_k(5.0);
assert!((detector.threshold_k() - 5.0).abs() < f64::EPSILON);
}
#[test]
fn test_ewma_detector_std_dev() {
let mut detector = EwmaDetector::new(0.1, 3.0);
for v in [10.0, 20.0, 30.0, 40.0, 50.0] {
detector.observe(v);
}
assert!(detector.std_dev() > 0.0);
}
#[test]
fn test_ewma_detector_reset() {
let mut detector = EwmaDetector::new(0.1, 3.0);
detector.observe(10.0);
detector.observe(20.0);
detector.reset();
assert_eq!(detector.count(), 0);
}
#[test]
fn test_zscore_detector_insufficient_data() {
let mut detector = ZScoreDetector::new(3.0);
detector.observe(10.0);
assert!(!detector.is_anomaly(1000.0));
}
#[test]
fn test_zscore_detector_variance() {
let mut detector = ZScoreDetector::new(3.0);
for v in [10.0, 20.0, 30.0] {
detector.observe(v);
}
assert!(detector.variance() > 0.0);
assert!(detector.std_dev() > 0.0);
}
#[test]
fn test_zscore_detector_z_score() {
let mut detector = ZScoreDetector::new(3.0);
for v in [10.0, 10.0, 10.0, 10.0, 10.0] {
detector.observe(v);
}
let z = detector.z_score(10.0);
assert!(z.abs() < f64::EPSILON || z == 0.0);
}
#[test]
fn test_zscore_detector_count() {
let mut detector = ZScoreDetector::new(3.0);
detector.observe(1.0);
detector.observe(2.0);
assert_eq!(detector.count(), 2);
}
#[test]
fn test_zscore_detector_reset() {
let mut detector = ZScoreDetector::new(3.0);
detector.observe(1.0);
detector.observe(2.0);
detector.reset();
assert_eq!(detector.count(), 0);
assert_eq!(detector.mean(), 0.0);
}
#[test]
fn test_composite_detector_default() {
let detector = CompositeDetector::default();
assert_eq!(detector.count(), 0);
}
#[test]
fn test_composite_detector_with_thresholds() {
let detector = CompositeDetector::with_thresholds(2.0, 0.2, 2.5, 2.0);
assert!((detector.mad.threshold_k() - 2.0).abs() < f64::EPSILON);
}
#[test]
fn test_composite_detector_consensus() {
let mut detector = CompositeDetector::new();
detector.require_consensus = true;
for v in [10.0, 10.5, 9.5, 10.2, 9.8, 10.1, 9.9] {
detector.observe(v);
}
assert!(!detector.is_anomaly(10.0));
}
#[test]
fn test_composite_detector_reset() {
let mut detector = CompositeDetector::new();
detector.observe(10.0);
detector.observe(20.0);
detector.reset();
assert_eq!(detector.count(), 0);
}
#[test]
fn test_anomaly_event_clone() {
let event = AnomalyEvent {
value: 100.0,
score: 5.0,
expected: 10.0,
timestamp: 12345,
metric_id: 1,
};
let cloned = event;
assert_eq!(cloned.value, 100.0);
assert_eq!(cloned.metric_id, 1);
}
#[cfg(feature = "std")]
#[test]
fn test_anomaly_event_debug() {
let event = AnomalyEvent {
value: 100.0,
score: 5.0,
expected: 10.0,
timestamp: 0,
metric_id: 0,
};
let s = format!("{event:?}");
assert!(s.contains("value"));
}
}