use u_numflow::fourier::{fft, ifft, Complex};
use u_numflow::special::inverse_normal_cdf;
use u_numflow::stats::median;
pub const MIN_OBSERVATIONS: usize = 12;
const _: () = assert!(
MIN_OBSERVATIONS == 12,
"the batch_size requirement message spells this number"
);
const EXTENSION: usize = 5;
#[derive(Debug, Clone, Copy, PartialEq)]
#[non_exhaustive]
pub enum SpectralResidualError {
OptionOutOfRange {
option: &'static str,
requirement: &'static str,
got: f64,
min: Option<f64>,
exclusive_min: Option<f64>,
max: Option<f64>,
exclusive_max: Option<f64>,
},
TooFewObservations {
needed: usize,
got: usize,
},
ValueNotFinite {
index: usize,
},
}
impl core::fmt::Display for SpectralResidualError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
SpectralResidualError::OptionOutOfRange {
option,
requirement,
..
} => write!(f, "{option} must be {requirement}"),
SpectralResidualError::TooFewObservations { needed, got } => {
write!(f, "needs at least {needed} observations, got {got}")
}
SpectralResidualError::ValueNotFinite { index } => {
write!(f, "data[{index}] is not a finite number")
}
}
}
}
impl std::error::Error for SpectralResidualError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CountOption {
AveragingWindow,
JudgementWindow,
BatchSize,
}
impl CountOption {
pub(crate) fn check(self, got: f64) -> Result<(), SpectralResidualError> {
let (option, requirement, min) = match self {
CountOption::AveragingWindow => ("averaging_window", ">= 1", 1),
CountOption::JudgementWindow => ("judgement_window", ">= 1", 1),
CountOption::BatchSize => ("batch_size", "unset, or at least 12", MIN_OBSERVATIONS),
};
if got < min as f64 {
Err(SpectralResidualError::below(
option,
requirement,
got,
min as f64,
))
} else {
Ok(())
}
}
}
impl SpectralResidualError {
fn below(option: &'static str, requirement: &'static str, got: f64, min: f64) -> Self {
Self::OptionOutOfRange {
option,
requirement,
got,
min: Some(min),
exclusive_min: None,
max: None,
exclusive_max: None,
}
}
fn not_above(
option: &'static str,
requirement: &'static str,
got: f64,
exclusive_min: f64,
) -> Self {
Self::OptionOutOfRange {
option,
requirement,
got,
min: None,
exclusive_min: Some(exclusive_min),
max: None,
exclusive_max: None,
}
}
fn not_between(
option: &'static str,
requirement: &'static str,
got: f64,
exclusive_min: f64,
exclusive_max: f64,
) -> Self {
Self::OptionOutOfRange {
option,
requirement,
got,
min: None,
exclusive_min: Some(exclusive_min),
max: None,
exclusive_max: Some(exclusive_max),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct SpectralResidual {
pub averaging_window: usize,
pub judgement_window: usize,
pub threshold: f64,
pub min_zscore: f64,
pub sensitivity: f64,
pub batch_size: Option<usize>,
}
impl Default for SpectralResidual {
fn default() -> Self {
Self::new()
}
}
impl SpectralResidual {
pub fn new() -> Self {
Self {
averaging_window: 3,
judgement_window: 40,
threshold: 3.0,
min_zscore: 1.5,
sensitivity: 70.0,
batch_size: None,
}
}
pub fn with_averaging_window(mut self, q: usize) -> Self {
self.averaging_window = q;
self
}
pub fn with_judgement_window(mut self, z: usize) -> Self {
self.judgement_window = z;
self
}
pub fn with_threshold(mut self, threshold: f64) -> Self {
self.threshold = threshold;
self
}
pub fn with_min_zscore(mut self, min_zscore: f64) -> Self {
self.min_zscore = min_zscore;
self
}
pub fn with_sensitivity(mut self, sensitivity: f64) -> Self {
self.sensitivity = sensitivity;
self
}
pub fn with_batch_size(mut self, batch_size: Option<usize>) -> Self {
self.batch_size = batch_size;
self
}
fn check_options(&self) -> Result<(), SpectralResidualError> {
use SpectralResidualError as E;
CountOption::AveragingWindow.check(self.averaging_window as f64)?;
CountOption::JudgementWindow.check(self.judgement_window as f64)?;
if !self.threshold.is_finite() || self.threshold <= 0.0 {
return Err(E::not_above(
"threshold",
"a finite number > 0",
self.threshold,
0.0,
));
}
if !self.min_zscore.is_finite() || self.min_zscore < 0.0 {
return Err(E::below(
"min_zscore",
"a finite number >= 0",
self.min_zscore,
0.0,
));
}
if !self.sensitivity.is_finite() || self.sensitivity <= 0.0 || self.sensitivity >= 100.0 {
return Err(E::not_between(
"sensitivity",
"a finite number strictly between 0 and 100",
self.sensitivity,
0.0,
100.0,
));
}
if let Some(b) = self.batch_size {
CountOption::BatchSize.check(b as f64)?;
}
Ok(())
}
pub fn analyze(&self, series: &[f64]) -> Result<Vec<SrPoint>, SpectralResidualError> {
self.check_options()?;
if series.len() < MIN_OBSERVATIONS {
return Err(SpectralResidualError::TooFewObservations {
needed: MIN_OBSERVATIONS,
got: series.len(),
});
}
if let Some(index) = series.iter().position(|v| !v.is_finite()) {
return Err(SpectralResidualError::ValueNotFinite { index });
}
let n = series.len();
let batch = self.batch_size.unwrap_or(n).min(n);
let mut points = Vec::with_capacity(n);
let mut start = 0;
while start < n {
let end = if n - start < 2 * batch {
n
} else {
start + batch
};
points.extend(self.analyze_batch(&series[start..end], start));
start = end;
}
Ok(points)
}
fn analyze_batch(&self, data: &[f64], offset: usize) -> Vec<SrPoint> {
let n = data.len();
let saliency = saliency_map(data, self.averaging_window);
let scores: Vec<f64> = (0..n)
.map(|i| {
let from = i.saturating_sub(self.judgement_window - 1);
let avg = saliency[from..=i].iter().sum::<f64>() / (i + 1 - from) as f64;
let avg = avg.max(1e-8);
((saliency[i] - avg) / avg).max(0.0)
})
.collect();
let flagged: Vec<bool> = (0..n)
.map(|i| {
if scores[i] <= self.threshold {
return false;
}
if self.min_zscore == 0.0 || i == 0 {
return true;
}
let from = i.saturating_sub(self.judgement_window);
let window = &data[from..i];
let mean = window.iter().sum::<f64>() / window.len() as f64;
let std = (window.iter().map(|v| (v - mean).powi(2)).sum::<f64>()
/ window.len() as f64)
.sqrt();
if std > 0.0 {
(data[i] - mean).abs() / std >= self.min_zscore
} else {
data[i] != mean
}
})
.collect();
let expected = expected_values(data, &flagged);
let residuals: Vec<f64> = (0..n)
.filter(|&i| !flagged[i])
.map(|i| (data[i] - expected[i]).abs())
.collect();
let spread = median(&residuals).unwrap_or(0.0) * 1.4826;
let z = inverse_normal_cdf(0.5 + self.sensitivity / 200.0);
let margin = z * spread;
(0..n)
.map(|i| SrPoint {
index: offset + i,
value: data[i],
saliency: saliency[i],
score: scores[i],
expected: expected[i],
lower: expected[i] - margin,
upper: expected[i] + margin,
is_anomaly: flagged[i],
near_edge: i < EXTENSION || i + EXTENSION >= n,
})
.collect()
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct SrPoint {
pub index: usize,
pub value: f64,
pub saliency: f64,
pub score: f64,
pub expected: f64,
pub lower: f64,
pub upper: f64,
pub is_anomaly: bool,
pub near_edge: bool,
}
fn saliency_map(data: &[f64], q: usize) -> Vec<f64> {
let n = data.len();
let mut extended = data.to_vec();
let m = EXTENSION.min(n - 1);
let last = data[n - 1];
let slope = (1..=m)
.map(|i| (last - data[n - 1 - i]) / i as f64)
.sum::<f64>()
/ m as f64;
let predicted = data[n - 1 - m] + slope * m as f64;
extended.extend(std::iter::repeat_n(predicted, EXTENSION));
let spectrum = fft(&extended
.iter()
.map(|&x| Complex::real(x))
.collect::<Vec<_>>());
let len = spectrum.len();
let log_amp: Vec<f64> = spectrum.iter().map(|z| (z.norm() + 1e-8).ln()).collect();
let mut avg = vec![0.0; len];
let mut cum = 0.0;
let mut cums = Vec::with_capacity(len + 1);
cums.push(0.0);
for &v in &log_amp {
cum += v;
cums.push(cum);
}
for i in 0..len {
let from = (i + 1).saturating_sub(q);
avg[i] = (cums[i + 1] - cums[from]) / (i + 1 - from) as f64;
}
let residual_spectrum: Vec<Complex> = spectrum
.iter()
.enumerate()
.map(|(i, z)| {
let r = (log_amp[i] - avg[i]).exp();
let phase = z.arg();
Complex::new(r * phase.cos(), r * phase.sin())
})
.collect();
ifft(&residual_spectrum)
.into_iter()
.take(n)
.map(|z| z.norm())
.collect()
}
fn expected_values(data: &[f64], flagged: &[bool]) -> Vec<f64> {
let n = data.len();
let mut cleaned = data.to_vec();
for i in 0..n {
if !flagged[i] {
continue;
}
let left = (0..i).rev().find(|&j| !flagged[j]).map(|j| data[j]);
let right = (i + 1..n).find(|&j| !flagged[j]).map(|j| data[j]);
cleaned[i] = match (left, right) {
(Some(a), Some(b)) => (a + b) / 2.0,
(Some(a), None) | (None, Some(a)) => a,
(None, None) => data[i],
};
}
let mut spectrum = fft(&cleaned
.iter()
.map(|&x| Complex::real(x))
.collect::<Vec<_>>());
let lo = n * 3 / 8;
let hi = n * 5 / 8;
for (i, z) in spectrum.iter_mut().enumerate() {
if i > lo && i < hi {
*z = Complex::default();
}
}
ifft(&spectrum).into_iter().map(|z| z.re).collect()
}
#[cfg(test)]
mod tests {
use super::*;
fn noise(seed: u64, n: usize) -> Vec<f64> {
let mut state = seed;
(0..n)
.map(|_| {
state = state
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
((state >> 11) as f64 / (1u64 << 53) as f64) * 2.0 - 1.0
})
.collect()
}
fn check_shape(points: &[SrPoint], n: usize) {
assert_eq!(points.len(), n);
for (i, p) in points.iter().enumerate() {
assert_eq!(p.index, i);
assert!(p.saliency >= 0.0 && p.saliency.is_finite(), "{p:?}");
assert!(p.score >= 0.0 && p.score.is_finite(), "{p:?}");
assert!(
p.expected.is_finite() && p.lower <= p.expected && p.expected <= p.upper,
"{p:?}"
);
}
}
#[test]
fn a_spike_on_a_sine_is_the_anomaly() {
let mut series: Vec<f64> = noise(3, 120)
.iter()
.enumerate()
.map(|(t, e)| (t as f64 * 0.25).sin() + 0.1 * e)
.collect();
series[80] += 5.0;
let points = SpectralResidual::new().analyze(&series).unwrap();
check_shape(&points, 120);
assert!(points[80].is_anomaly, "{:?}", points[80]);
let flagged: Vec<usize> = points
.iter()
.filter(|p| p.is_anomaly)
.map(|p| p.index)
.collect();
assert!(flagged.len() <= 3, "flagged {flagged:?}");
assert!(points[80].value > points[80].upper);
assert!(
(points[80].expected - (80.0f64 * 0.25).sin()).abs() < 0.6,
"{:?}",
points[80]
);
}
#[test]
fn a_step_is_flagged_where_it_happens() {
let series: Vec<f64> = (0..100).map(|t| if t < 60 { 1.0 } else { 6.0 }).collect();
let points = SpectralResidual::new().analyze(&series).unwrap();
check_shape(&points, 100);
let first = points.iter().position(|p| p.is_anomaly);
assert_eq!(
first,
Some(60),
"{:?}",
points
.iter()
.filter(|p| p.is_anomaly)
.map(|p| p.index)
.collect::<Vec<_>>()
);
}
#[test]
fn a_constant_and_plain_noise_raise_no_anomaly() {
let constant = vec![3.0; 50];
let points = SpectralResidual::new().analyze(&constant).unwrap();
check_shape(&points, 50);
assert!(points.iter().all(|p| !p.is_anomaly));
assert!(points.iter().all(|p| (p.expected - 3.0).abs() < 1e-9));
let series = noise(9, 200);
let points = SpectralResidual::new().analyze(&series).unwrap();
check_shape(&points, 200);
let flagged = points.iter().filter(|p| p.is_anomaly).count();
assert!(flagged <= 4, "{flagged} of 200 noise points flagged");
}
#[test]
fn scoring_is_deterministic_and_batching_keeps_the_shape() {
let mut series: Vec<f64> = noise(5, 300).iter().map(|e| 10.0 + e).collect();
series[150] = 20.0;
series[260] = -5.0;
let sr = SpectralResidual::new();
let a = sr.analyze(&series).unwrap();
let b = sr.analyze(&series).unwrap();
assert_eq!(a, b);
let batched = sr.with_batch_size(Some(100)).analyze(&series).unwrap();
check_shape(&batched, 300);
assert!(batched[150].is_anomaly && batched[260].is_anomaly);
assert_eq!(batched[200].index, 200);
}
#[test]
fn the_band_widens_with_sensitivity() {
let series: Vec<f64> = noise(1, 80).iter().map(|e| 5.0 + e).collect();
let narrow = SpectralResidual::new()
.with_sensitivity(50.0)
.analyze(&series)
.unwrap();
let wide = SpectralResidual::new()
.with_sensitivity(99.0)
.analyze(&series)
.unwrap();
for (a, b) in narrow.iter().zip(&wide) {
assert!(b.upper - b.lower > a.upper - a.lower);
assert_eq!(a.expected, b.expected);
}
}
#[test]
fn points_near_either_end_of_a_batch_are_marked() {
let series: Vec<f64> = (0..200).map(|i| (i % 7) as f64).collect();
let points = SpectralResidual::new().analyze(&series).expect("ok");
let marked: Vec<usize> = points
.iter()
.filter(|p| p.near_edge)
.map(|p| p.index)
.collect();
assert_eq!(marked, vec![0, 1, 2, 3, 4, 195, 196, 197, 198, 199]);
let batched = SpectralResidual::new()
.with_batch_size(Some(50))
.analyze(&series)
.expect("ok");
let marked: Vec<usize> = batched
.iter()
.filter(|p| p.near_edge)
.map(|p| p.index)
.collect();
assert!(marked.contains(&49) && marked.contains(&50), "{marked:?}");
assert!(!marked.contains(&75), "{marked:?}");
assert_eq!(
marked.len(),
4 * 10,
"four batches, ten marks each: {marked:?}"
);
let mut spiked: Vec<f64> = (0..60).map(|i| (i % 7) as f64).collect();
spiked[57] += 30.0;
let points = SpectralResidual::new().analyze(&spiked).expect("ok");
assert!(points[57].is_anomaly);
assert!(points[57].near_edge);
}
#[test]
fn a_step_can_flag_the_point_before_the_transition() {
let step: Vec<f64> = (0..120)
.map(|i| {
let jitter = ((i as f64 * 7.3891).sin() * 10000.0).fract().abs() * 0.2;
(if i < 60 { 1.0 } else { 6.0 }) + jitter
})
.collect();
let points = SpectralResidual::new().analyze(&step).expect("ok");
let flagged: Vec<usize> = points
.iter()
.filter(|p| p.is_anomaly && !p.near_edge)
.map(|p| p.index)
.collect();
assert!(!flagged.is_empty(), "the transition is detected");
assert!(
flagged.iter().all(|&i| (59..=61).contains(&i)),
"detections sit at the transition or its neighbour: {flagged:?}"
);
}
#[test]
fn a_refusal_names_the_condition_that_failed() {
let ok = vec![1.0; 20];
use SpectralResidualError as E;
assert_eq!(
SpectralResidual::new().analyze(&[1.0; 11]),
Err(SpectralResidualError::TooFewObservations {
needed: MIN_OBSERVATIONS,
got: 11
})
);
let mut series = vec![1.0; 20];
series[7] = f64::INFINITY;
assert_eq!(
SpectralResidual::new().analyze(&series),
Err(SpectralResidualError::ValueNotFinite { index: 7 })
);
assert_eq!(
SpectralResidual::new().with_threshold(0.0).analyze(&ok),
Err(E::not_above("threshold", "a finite number > 0", 0.0, 0.0))
);
let sensitivity = |got| {
Err(E::not_between(
"sensitivity",
"a finite number strictly between 0 and 100",
got,
0.0,
100.0,
))
};
assert_eq!(
SpectralResidual::new().with_sensitivity(100.0).analyze(&ok),
sensitivity(100.0)
);
assert_eq!(
SpectralResidual::new().with_sensitivity(0.0).analyze(&ok),
sensitivity(0.0)
);
assert_eq!(
SpectralResidual::new()
.with_batch_size(Some(5))
.analyze(&ok),
Err(E::below("batch_size", "unset, or at least 12", 5.0, 12.0))
);
assert_eq!(
SpectralResidual::new()
.with_averaging_window(0)
.analyze(&ok),
Err(E::below("averaging_window", ">= 1", 0.0, 1.0))
);
assert_eq!(
SpectralResidual::new()
.with_judgement_window(0)
.analyze(&ok),
Err(E::below("judgement_window", ">= 1", 0.0, 1.0))
);
assert_eq!(
SpectralResidual::new().with_min_zscore(-1.0).analyze(&ok),
Err(E::below("min_zscore", "a finite number >= 0", -1.0, 0.0))
);
assert!(matches!(
SpectralResidual::new()
.with_threshold(f64::NAN)
.analyze(&ok),
Err(E::OptionOutOfRange {
option: "threshold",
min: None,
exclusive_min,
max: None,
exclusive_max: None,
got,
..
}) if got.is_nan() && exclusive_min == Some(0.0)
));
let message = SpectralResidual::new()
.with_threshold(0.0)
.analyze(&ok)
.expect_err("refused")
.to_string();
assert_eq!(message, "threshold must be a finite number > 0");
assert!(!message.contains("sensitivity"));
assert!(SpectralResidual::new()
.with_min_zscore(0.0)
.analyze(&ok)
.is_ok());
}
}