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 EXTENSION: usize = 5;
#[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 valid(&self) -> bool {
self.averaging_window >= 1
&& self.judgement_window >= 1
&& self.threshold.is_finite()
&& self.threshold > 0.0
&& self.min_zscore.is_finite()
&& self.min_zscore >= 0.0
&& self.sensitivity.is_finite()
&& self.sensitivity > 0.0
&& self.sensitivity < 100.0
&& self.batch_size.is_none_or(|b| b >= MIN_OBSERVATIONS)
}
pub fn analyze(&self, series: &[f64]) -> Option<Vec<SrPoint>> {
if !self.valid() || series.len() < MIN_OBSERVATIONS || series.iter().any(|v| !v.is_finite())
{
return None;
}
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;
}
Some(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],
})
.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,
}
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 invalid_input_and_configuration_are_refused() {
assert!(SpectralResidual::new().analyze(&[1.0; 11]).is_none());
let mut series = vec![1.0; 20];
series[7] = f64::INFINITY;
assert!(SpectralResidual::new().analyze(&series).is_none());
let ok = vec![1.0; 20];
assert!(SpectralResidual::new()
.with_threshold(0.0)
.analyze(&ok)
.is_none());
assert!(SpectralResidual::new()
.with_sensitivity(100.0)
.analyze(&ok)
.is_none());
assert!(SpectralResidual::new()
.with_averaging_window(0)
.analyze(&ok)
.is_none());
assert!(SpectralResidual::new()
.with_batch_size(Some(5))
.analyze(&ok)
.is_none());
assert!(SpectralResidual::new()
.with_min_zscore(0.0)
.analyze(&ok)
.is_some());
}
}