1use u_numflow::fourier::{fft, ifft, Complex};
52use u_numflow::special::inverse_normal_cdf;
53use u_numflow::stats::median;
54
55pub const MIN_OBSERVATIONS: usize = 12;
57
58const _: () = assert!(
62 MIN_OBSERVATIONS == 12,
63 "the batch_size requirement message spells this number"
64);
65
66const EXTENSION: usize = 5;
70
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78#[non_exhaustive]
79pub enum SpectralResidualError {
80 OptionOutOfRange {
82 option: &'static str,
85 requirement: &'static str,
87 },
88 TooFewObservations {
90 needed: usize,
92 got: usize,
94 },
95 ValueNotFinite {
97 index: usize,
99 },
100}
101
102impl core::fmt::Display for SpectralResidualError {
103 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
104 match self {
105 SpectralResidualError::OptionOutOfRange {
106 option,
107 requirement,
108 } => write!(f, "{option} must be {requirement}"),
109 SpectralResidualError::TooFewObservations { needed, got } => {
110 write!(f, "needs at least {needed} observations, got {got}")
111 }
112 SpectralResidualError::ValueNotFinite { index } => {
113 write!(f, "data[{index}] is not a finite number")
114 }
115 }
116 }
117}
118
119impl std::error::Error for SpectralResidualError {}
120
121#[derive(Debug, Clone, PartialEq)]
126pub struct SpectralResidual {
127 pub averaging_window: usize,
130 pub judgement_window: usize,
133 pub threshold: f64,
135 pub min_zscore: f64,
140 pub sensitivity: f64,
144 pub batch_size: Option<usize>,
149}
150
151impl Default for SpectralResidual {
152 fn default() -> Self {
153 Self::new()
154 }
155}
156
157impl SpectralResidual {
158 pub fn new() -> Self {
161 Self {
162 averaging_window: 3,
163 judgement_window: 40,
164 threshold: 3.0,
165 min_zscore: 1.5,
166 sensitivity: 70.0,
167 batch_size: None,
168 }
169 }
170
171 pub fn with_averaging_window(mut self, q: usize) -> Self {
173 self.averaging_window = q;
174 self
175 }
176
177 pub fn with_judgement_window(mut self, z: usize) -> Self {
179 self.judgement_window = z;
180 self
181 }
182
183 pub fn with_threshold(mut self, threshold: f64) -> Self {
185 self.threshold = threshold;
186 self
187 }
188
189 pub fn with_min_zscore(mut self, min_zscore: f64) -> Self {
191 self.min_zscore = min_zscore;
192 self
193 }
194
195 pub fn with_sensitivity(mut self, sensitivity: f64) -> Self {
197 self.sensitivity = sensitivity;
198 self
199 }
200
201 pub fn with_batch_size(mut self, batch_size: Option<usize>) -> Self {
203 self.batch_size = batch_size;
204 self
205 }
206
207 fn check_options(&self) -> Result<(), SpectralResidualError> {
209 let out_of_range = |option, requirement| {
210 Err(SpectralResidualError::OptionOutOfRange {
211 option,
212 requirement,
213 })
214 };
215 if self.averaging_window < 1 {
216 return out_of_range("averaging_window", ">= 1");
217 }
218 if self.judgement_window < 1 {
219 return out_of_range("judgement_window", ">= 1");
220 }
221 if !self.threshold.is_finite() || self.threshold <= 0.0 {
222 return out_of_range("threshold", "a finite number > 0");
223 }
224 if !self.min_zscore.is_finite() || self.min_zscore < 0.0 {
225 return out_of_range("min_zscore", "a finite number >= 0");
226 }
227 if !self.sensitivity.is_finite() || self.sensitivity <= 0.0 || self.sensitivity >= 100.0 {
228 return out_of_range("sensitivity", "a finite number strictly between 0 and 100");
229 }
230 if self.batch_size.is_some_and(|b| b < MIN_OBSERVATIONS) {
231 return out_of_range("batch_size", "unset, or at least 12");
232 }
233 Ok(())
234 }
235
236 pub fn analyze(&self, series: &[f64]) -> Result<Vec<SrPoint>, SpectralResidualError> {
269 self.check_options()?;
270 if series.len() < MIN_OBSERVATIONS {
271 return Err(SpectralResidualError::TooFewObservations {
272 needed: MIN_OBSERVATIONS,
273 got: series.len(),
274 });
275 }
276 if let Some(index) = series.iter().position(|v| !v.is_finite()) {
277 return Err(SpectralResidualError::ValueNotFinite { index });
278 }
279 let n = series.len();
280 let batch = self.batch_size.unwrap_or(n).min(n);
281 let mut points = Vec::with_capacity(n);
282 let mut start = 0;
283 while start < n {
284 let end = if n - start < 2 * batch {
286 n
287 } else {
288 start + batch
289 };
290 points.extend(self.analyze_batch(&series[start..end], start));
291 start = end;
292 }
293 Ok(points)
294 }
295
296 fn analyze_batch(&self, data: &[f64], offset: usize) -> Vec<SrPoint> {
297 let n = data.len();
298 let saliency = saliency_map(data, self.averaging_window);
299
300 let scores: Vec<f64> = (0..n)
303 .map(|i| {
304 let from = i.saturating_sub(self.judgement_window - 1);
305 let avg = saliency[from..=i].iter().sum::<f64>() / (i + 1 - from) as f64;
306 let avg = avg.max(1e-8);
307 ((saliency[i] - avg) / avg).max(0.0)
308 })
309 .collect();
310
311 let flagged: Vec<bool> = (0..n)
315 .map(|i| {
316 if scores[i] <= self.threshold {
317 return false;
318 }
319 if self.min_zscore == 0.0 || i == 0 {
320 return true;
321 }
322 let from = i.saturating_sub(self.judgement_window);
323 let window = &data[from..i];
324 let mean = window.iter().sum::<f64>() / window.len() as f64;
325 let std = (window.iter().map(|v| (v - mean).powi(2)).sum::<f64>()
326 / window.len() as f64)
327 .sqrt();
328 if std > 0.0 {
329 (data[i] - mean).abs() / std >= self.min_zscore
330 } else {
331 data[i] != mean
332 }
333 })
334 .collect();
335
336 let expected = expected_values(data, &flagged);
337 let residuals: Vec<f64> = (0..n)
338 .filter(|&i| !flagged[i])
339 .map(|i| (data[i] - expected[i]).abs())
340 .collect();
341 let spread = median(&residuals).unwrap_or(0.0) * 1.4826;
345 let z = inverse_normal_cdf(0.5 + self.sensitivity / 200.0);
346 let margin = z * spread;
347
348 (0..n)
349 .map(|i| SrPoint {
350 index: offset + i,
351 value: data[i],
352 saliency: saliency[i],
353 score: scores[i],
354 expected: expected[i],
355 lower: expected[i] - margin,
356 upper: expected[i] + margin,
357 is_anomaly: flagged[i],
358 near_edge: i < EXTENSION || i + EXTENSION >= n,
359 })
360 .collect()
361 }
362}
363
364#[derive(Debug, Clone, PartialEq)]
366pub struct SrPoint {
367 pub index: usize,
369 pub value: f64,
371 pub saliency: f64,
373 pub score: f64,
375 pub expected: f64,
377 pub lower: f64,
379 pub upper: f64,
381 pub is_anomaly: bool,
383 pub near_edge: bool,
399}
400
401fn saliency_map(data: &[f64], q: usize) -> Vec<f64> {
406 let n = data.len();
407 let mut extended = data.to_vec();
408 let m = EXTENSION.min(n - 1);
409 let last = data[n - 1];
410 let slope = (1..=m)
411 .map(|i| (last - data[n - 1 - i]) / i as f64)
412 .sum::<f64>()
413 / m as f64;
414 let predicted = data[n - 1 - m] + slope * m as f64;
415 extended.extend(std::iter::repeat_n(predicted, EXTENSION));
416
417 let spectrum = fft(&extended
418 .iter()
419 .map(|&x| Complex::real(x))
420 .collect::<Vec<_>>());
421 let len = spectrum.len();
422 let log_amp: Vec<f64> = spectrum.iter().map(|z| (z.norm() + 1e-8).ln()).collect();
423 let mut avg = vec![0.0; len];
425 let mut cum = 0.0;
426 let mut cums = Vec::with_capacity(len + 1);
427 cums.push(0.0);
428 for &v in &log_amp {
429 cum += v;
430 cums.push(cum);
431 }
432 for i in 0..len {
433 let from = (i + 1).saturating_sub(q);
434 avg[i] = (cums[i + 1] - cums[from]) / (i + 1 - from) as f64;
435 }
436 let residual_spectrum: Vec<Complex> = spectrum
437 .iter()
438 .enumerate()
439 .map(|(i, z)| {
440 let r = (log_amp[i] - avg[i]).exp();
441 let phase = z.arg();
442 Complex::new(r * phase.cos(), r * phase.sin())
443 })
444 .collect();
445 ifft(&residual_spectrum)
446 .into_iter()
447 .take(n)
448 .map(|z| z.norm())
449 .collect()
450}
451
452fn expected_values(data: &[f64], flagged: &[bool]) -> Vec<f64> {
457 let n = data.len();
458 let mut cleaned = data.to_vec();
459 for i in 0..n {
460 if !flagged[i] {
461 continue;
462 }
463 let left = (0..i).rev().find(|&j| !flagged[j]).map(|j| data[j]);
464 let right = (i + 1..n).find(|&j| !flagged[j]).map(|j| data[j]);
465 cleaned[i] = match (left, right) {
466 (Some(a), Some(b)) => (a + b) / 2.0,
467 (Some(a), None) | (None, Some(a)) => a,
468 (None, None) => data[i],
469 };
470 }
471 let mut spectrum = fft(&cleaned
472 .iter()
473 .map(|&x| Complex::real(x))
474 .collect::<Vec<_>>());
475 let lo = n * 3 / 8;
476 let hi = n * 5 / 8;
477 for (i, z) in spectrum.iter_mut().enumerate() {
478 if i > lo && i < hi {
479 *z = Complex::default();
480 }
481 }
482 ifft(&spectrum).into_iter().map(|z| z.re).collect()
483}
484
485#[cfg(test)]
486mod tests {
487 use super::*;
488
489 fn noise(seed: u64, n: usize) -> Vec<f64> {
490 let mut state = seed;
491 (0..n)
492 .map(|_| {
493 state = state
494 .wrapping_mul(6364136223846793005)
495 .wrapping_add(1442695040888963407);
496 ((state >> 11) as f64 / (1u64 << 53) as f64) * 2.0 - 1.0
497 })
498 .collect()
499 }
500
501 fn check_shape(points: &[SrPoint], n: usize) {
502 assert_eq!(points.len(), n);
503 for (i, p) in points.iter().enumerate() {
504 assert_eq!(p.index, i);
505 assert!(p.saliency >= 0.0 && p.saliency.is_finite(), "{p:?}");
506 assert!(p.score >= 0.0 && p.score.is_finite(), "{p:?}");
507 assert!(
508 p.expected.is_finite() && p.lower <= p.expected && p.expected <= p.upper,
509 "{p:?}"
510 );
511 }
512 }
513
514 #[test]
515 fn a_spike_on_a_sine_is_the_anomaly() {
516 let mut series: Vec<f64> = noise(3, 120)
517 .iter()
518 .enumerate()
519 .map(|(t, e)| (t as f64 * 0.25).sin() + 0.1 * e)
520 .collect();
521 series[80] += 5.0;
522 let points = SpectralResidual::new().analyze(&series).unwrap();
523 check_shape(&points, 120);
524 assert!(points[80].is_anomaly, "{:?}", points[80]);
525 let flagged: Vec<usize> = points
526 .iter()
527 .filter(|p| p.is_anomaly)
528 .map(|p| p.index)
529 .collect();
530 assert!(flagged.len() <= 3, "flagged {flagged:?}");
531 assert!(points[80].value > points[80].upper);
532 assert!(
534 (points[80].expected - (80.0f64 * 0.25).sin()).abs() < 0.6,
535 "{:?}",
536 points[80]
537 );
538 }
539
540 #[test]
541 fn a_step_is_flagged_where_it_happens() {
542 let series: Vec<f64> = (0..100).map(|t| if t < 60 { 1.0 } else { 6.0 }).collect();
543 let points = SpectralResidual::new().analyze(&series).unwrap();
544 check_shape(&points, 100);
545 let first = points.iter().position(|p| p.is_anomaly);
546 assert_eq!(
547 first,
548 Some(60),
549 "{:?}",
550 points
551 .iter()
552 .filter(|p| p.is_anomaly)
553 .map(|p| p.index)
554 .collect::<Vec<_>>()
555 );
556 }
557
558 #[test]
559 fn a_constant_and_plain_noise_raise_no_anomaly() {
560 let constant = vec![3.0; 50];
561 let points = SpectralResidual::new().analyze(&constant).unwrap();
562 check_shape(&points, 50);
563 assert!(points.iter().all(|p| !p.is_anomaly));
564 assert!(points.iter().all(|p| (p.expected - 3.0).abs() < 1e-9));
565
566 let series = noise(9, 200);
567 let points = SpectralResidual::new().analyze(&series).unwrap();
568 check_shape(&points, 200);
569 let flagged = points.iter().filter(|p| p.is_anomaly).count();
570 assert!(flagged <= 4, "{flagged} of 200 noise points flagged");
571 }
572
573 #[test]
574 fn scoring_is_deterministic_and_batching_keeps_the_shape() {
575 let mut series: Vec<f64> = noise(5, 300).iter().map(|e| 10.0 + e).collect();
576 series[150] = 20.0;
577 series[260] = -5.0;
578 let sr = SpectralResidual::new();
579 let a = sr.analyze(&series).unwrap();
580 let b = sr.analyze(&series).unwrap();
581 assert_eq!(a, b);
582 let batched = sr.with_batch_size(Some(100)).analyze(&series).unwrap();
583 check_shape(&batched, 300);
584 assert!(batched[150].is_anomaly && batched[260].is_anomaly);
585 assert_eq!(batched[200].index, 200);
588 }
589
590 #[test]
591 fn the_band_widens_with_sensitivity() {
592 let series: Vec<f64> = noise(1, 80).iter().map(|e| 5.0 + e).collect();
593 let narrow = SpectralResidual::new()
594 .with_sensitivity(50.0)
595 .analyze(&series)
596 .unwrap();
597 let wide = SpectralResidual::new()
598 .with_sensitivity(99.0)
599 .analyze(&series)
600 .unwrap();
601 for (a, b) in narrow.iter().zip(&wide) {
602 assert!(b.upper - b.lower > a.upper - a.lower);
603 assert_eq!(a.expected, b.expected);
604 }
605 }
606
607 #[test]
611 fn points_near_either_end_of_a_batch_are_marked() {
612 let series: Vec<f64> = (0..200).map(|i| (i % 7) as f64).collect();
613 let points = SpectralResidual::new().analyze(&series).expect("ok");
614 let marked: Vec<usize> = points
615 .iter()
616 .filter(|p| p.near_edge)
617 .map(|p| p.index)
618 .collect();
619 assert_eq!(marked, vec![0, 1, 2, 3, 4, 195, 196, 197, 198, 199]);
620
621 let batched = SpectralResidual::new()
623 .with_batch_size(Some(50))
624 .analyze(&series)
625 .expect("ok");
626 let marked: Vec<usize> = batched
627 .iter()
628 .filter(|p| p.near_edge)
629 .map(|p| p.index)
630 .collect();
631 assert!(marked.contains(&49) && marked.contains(&50), "{marked:?}");
634 assert!(!marked.contains(&75), "{marked:?}");
635 assert_eq!(
636 marked.len(),
637 4 * 10,
638 "four batches, ten marks each: {marked:?}"
639 );
640
641 let mut spiked: Vec<f64> = (0..60).map(|i| (i % 7) as f64).collect();
644 spiked[57] += 30.0;
645 let points = SpectralResidual::new().analyze(&spiked).expect("ok");
646 assert!(points[57].is_anomaly);
647 assert!(points[57].near_edge);
648 }
649
650 #[test]
653 fn a_step_can_flag_the_point_before_the_transition() {
654 let step: Vec<f64> = (0..120)
655 .map(|i| {
656 let jitter = ((i as f64 * 7.3891).sin() * 10000.0).fract().abs() * 0.2;
657 (if i < 60 { 1.0 } else { 6.0 }) + jitter
658 })
659 .collect();
660 let points = SpectralResidual::new().analyze(&step).expect("ok");
661 let flagged: Vec<usize> = points
662 .iter()
663 .filter(|p| p.is_anomaly && !p.near_edge)
664 .map(|p| p.index)
665 .collect();
666 assert!(!flagged.is_empty(), "the transition is detected");
667 assert!(
668 flagged.iter().all(|&i| (59..=61).contains(&i)),
669 "detections sit at the transition or its neighbour: {flagged:?}"
670 );
671 }
672
673 #[test]
676 fn a_refusal_names_the_condition_that_failed() {
677 let ok = vec![1.0; 20];
678 let out_of_range = |option, requirement| {
679 Err(SpectralResidualError::OptionOutOfRange {
680 option,
681 requirement,
682 })
683 };
684
685 assert_eq!(
686 SpectralResidual::new().analyze(&[1.0; 11]),
687 Err(SpectralResidualError::TooFewObservations {
688 needed: MIN_OBSERVATIONS,
689 got: 11
690 })
691 );
692
693 let mut series = vec![1.0; 20];
694 series[7] = f64::INFINITY;
695 assert_eq!(
696 SpectralResidual::new().analyze(&series),
697 Err(SpectralResidualError::ValueNotFinite { index: 7 })
698 );
699
700 assert_eq!(
704 SpectralResidual::new().with_threshold(0.0).analyze(&ok),
705 out_of_range("threshold", "a finite number > 0")
706 );
707 assert_eq!(
708 SpectralResidual::new().with_sensitivity(100.0).analyze(&ok),
709 out_of_range("sensitivity", "a finite number strictly between 0 and 100")
710 );
711 assert_eq!(
712 SpectralResidual::new().with_sensitivity(0.0).analyze(&ok),
713 out_of_range("sensitivity", "a finite number strictly between 0 and 100")
714 );
715 assert_eq!(
716 SpectralResidual::new()
717 .with_batch_size(Some(5))
718 .analyze(&ok),
719 out_of_range("batch_size", "unset, or at least 12")
720 );
721 assert_eq!(
722 SpectralResidual::new()
723 .with_averaging_window(0)
724 .analyze(&ok),
725 out_of_range("averaging_window", ">= 1")
726 );
727 assert_eq!(
728 SpectralResidual::new()
729 .with_judgement_window(0)
730 .analyze(&ok),
731 out_of_range("judgement_window", ">= 1")
732 );
733 assert_eq!(
734 SpectralResidual::new().with_min_zscore(-1.0).analyze(&ok),
735 out_of_range("min_zscore", "a finite number >= 0")
736 );
737 assert_eq!(
738 SpectralResidual::new()
739 .with_threshold(f64::NAN)
740 .analyze(&ok),
741 out_of_range("threshold", "a finite number > 0")
742 );
743
744 let message = SpectralResidual::new()
746 .with_threshold(0.0)
747 .analyze(&ok)
748 .expect_err("refused")
749 .to_string();
750 assert_eq!(message, "threshold must be a finite number > 0");
751 assert!(!message.contains("sensitivity"));
752
753 assert!(SpectralResidual::new()
755 .with_min_zscore(0.0)
756 .analyze(&ok)
757 .is_ok());
758 }
759}