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 - 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());
}
}
#[cfg(test)]
mod numpy_reference {
use super::*;
const SERIES: [f64; 64] = [
0.061, 0.585, 0.52, 1.134, 1.216, 0.847, 0.806, 0.369, -0.109, -0.265, -0.526, -1.008,
-1.139, -0.838, -0.869, -0.672, -0.068, 0.575, 0.594, 0.806, 0.924, 0.716, 0.839, 0.411,
0.217, -0.158, -0.583, -1.146, -1.238, -0.975, 3.168, -0.527, 0.231, 0.418, 0.492, 1.129,
0.898, 1.212, 0.745, 0.17, -0.134, -0.308, -0.865, -0.867, -0.867, -0.913, -0.746, -0.503,
0.119, 0.423, 0.811, 0.793, 0.878, 1.089, 0.49, 0.568, 0.154, -0.484, -0.677, -1.11,
-0.991, -0.84, -0.504, -0.44,
];
const SALIENCY: [f64; 64] = [
0.0369614318108603,
0.0796307697333925,
0.0271892659280146,
0.152493866186463,
0.166141948303089,
0.0816656918320778,
0.107174248120332,
0.0556332303417036,
0.0251300576938508,
0.00299138253005822,
0.00594392958654813,
0.106106695034459,
0.119126565861323,
0.0424769851751049,
0.0893964151003082,
0.0920927965046129,
0.0198418341809379,
0.0882438287553024,
0.0549823475501431,
0.067869463097445,
0.113994918150788,
0.105398733553807,
0.119888725456503,
0.0746552451898384,
0.109038966340197,
0.0173152365850345,
0.0229006048210958,
0.132971301432821,
0.205056469142638,
0.150938526823712,
0.830517907514111,
0.128639067122104,
0.0093039215691647,
0.0611436432409572,
0.0458141563355618,
0.143893509514701,
0.129454085999137,
0.179180669344178,
0.105280578308881,
0.0679545755367448,
0.0275280318011067,
0.0125531569333055,
0.0908993020495842,
0.0747621046819059,
0.0497022369555018,
0.0789105027673606,
0.0665323274213843,
0.0500154087434401,
0.0336638535386336,
0.0524451689724362,
0.0975047615363696,
0.0534062948099107,
0.0728496986560913,
0.133492464750329,
0.0136337226253535,
0.068520339745976,
0.0280483371477404,
0.102625433672842,
0.106450145288677,
0.183534057756558,
0.135468059382387,
0.0877555778478272,
0.0278059231072768,
0.0277019054815613,
];
const SCORE: [f64; 64] = [
0.0,
0.365970771264122,
0.0,
1.05881285226553,
0.796450465672768,
0.0,
0.15195611134601,
0.0,
0.0,
0.0,
0.0,
0.503171625125652,
0.602838806095091,
0.0,
0.22119300766101,
0.238060922935183,
0.0,
0.223493163232391,
0.0,
0.0,
0.559450528883167,
0.413466232376651,
0.566393797404165,
0.0,
0.402200065392971,
0.0,
0.0,
0.758500430509727,
1.5606529881759,
0.830851346829295,
6.79293349706241,
0.199289129287858,
0.0,
0.0,
0.0,
0.402864454850786,
0.253212692235468,
0.701706509860138,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
0.315186058686334,
0.0,
0.0,
0.0,
0.0305605858088266,
0.0553323044375994,
0.76882472342237,
0.298864697731819,
0.0,
0.0,
0.0,
];
fn close(got: f64, want: f64) -> bool {
(got - want).abs() <= 1e-8 * want.abs().max(1.0)
}
#[test]
fn saliency_and_scores_reproduce_the_published_method() {
let points = SpectralResidual::new()
.analyze(&SERIES)
.expect("a valid series");
for (i, p) in points.iter().enumerate() {
assert!(
close(p.saliency, SALIENCY[i]),
"saliency[{i}]: got {}, reference {}",
p.saliency,
SALIENCY[i]
);
assert!(
close(p.score, SCORE[i]),
"score[{i}]: got {}, reference {}",
p.score,
SCORE[i]
);
}
let flagged: Vec<usize> = points
.iter()
.filter(|p| p.is_anomaly)
.map(|p| p.index)
.collect();
assert_eq!(flagged, vec![30]);
}
}