A threshold on the raw samples can only catch a change that is large compared with the noise. + * CUSUM instead accumulates evidence. Every sample is normalised into + * {@code z = (x - target) / sigma} and pushed into two one-sided sums: + * + *
+ * upper <- max(0, upper + z - allowance) + * lower <- max(0, lower - z - allowance) + *+ * + *
The allowance {@code k} is a toll paid on every step. While the stream sits at its + * target the toll exceeds the average evidence, both sums are pinned at zero and the detector stays + * quiet no matter how long it runs. As soon as the mean shifts by more than {@code k} standard + * deviations, the corresponding sum starts drifting upwards, and it keeps drifting: a small but + * persistent bias accumulates until it crosses the decision threshold {@code h}. That is the whole + * point of the method - a shift of half a standard deviation is invisible in any single sample, yet + * unmistakable after twenty of them. + * + *
The classic tuning is {@code k = delta / 2} for the shift size {@code delta} one wants to catch + * quickly, together with {@code h} between 4 and 5, which keeps false alarms rare while detecting a + * one sigma shift within roughly ten samples. Both sums are cleared whenever an alarm fires, so the + * detector immediately starts looking for the next change instead of latching. + * + *
{@code
+ * // Watch for a shift of one standard deviation around a target of 20.0.
+ * CusumDetector detector = new CusumDetector(20.0, 0.5, 0.5, 5.0);
+ * for (double sample : stream) {
+ * ShiftSignal signal = detector.accept(sample);
+ * if (signal.isAlarm()) {
+ * alert(signal, detector.count());
+ * }
+ * }
+ * }
+ *
+ * Each sample costs O(1) time and the detector keeps O(1) state. This class is not thread-safe. + * + * @see CUSUM + */ +public final class CusumDetector { + + /** Allowance used when none is given; it targets shifts of one standard deviation. */ + public static final double DEFAULT_ALLOWANCE = 0.5; + + /** Decision threshold used when none is given. */ + public static final double DEFAULT_THRESHOLD = 5.0; + + private final double target; + private final double standardDeviation; + private final double allowance; + private final double threshold; + + private double upperSum; + private double lowerSum; + private long count; + private long alarmCount; + private ShiftSignal lastSignal = ShiftSignal.NONE; + + /** + * Creates a detector tuned for shifts of about one standard deviation. + * + * @param target the level the stream is expected to sit at + * @param standardDeviation the noise level of the stream, strictly positive + * @throws IllegalArgumentException if {@code target} is not finite or {@code standardDeviation} is not strictly positive + */ + public CusumDetector(double target, double standardDeviation) { + this(target, standardDeviation, DEFAULT_ALLOWANCE, DEFAULT_THRESHOLD); + } + + /** + * Creates a detector. + * + * @param target the level the stream is expected to sit at + * @param standardDeviation the noise level of the stream, strictly positive + * @param allowance the toll subtracted on every step, in standard deviations; half of the shift + * size one wants to detect quickly + * @param threshold how much accumulated evidence raises an alarm, in standard deviations + * @throws IllegalArgumentException if any argument is not finite, if {@code standardDeviation} or + * {@code threshold} is not strictly positive, or if {@code allowance} is negative + */ + public CusumDetector(double target, double standardDeviation, double allowance, double threshold) { + requireFinite(target, "target"); + if (!(standardDeviation > 0.0) || !Double.isFinite(standardDeviation)) { + throw new IllegalArgumentException("The standard deviation must be finite and strictly positive, but was " + standardDeviation); + } + if (!(allowance >= 0.0) || !Double.isFinite(allowance)) { + throw new IllegalArgumentException("The allowance must be finite and non-negative, but was " + allowance); + } + if (!(threshold > 0.0) || !Double.isFinite(threshold)) { + throw new IllegalArgumentException("The threshold must be finite and strictly positive, but was " + threshold); + } + this.target = target; + this.standardDeviation = standardDeviation; + this.allowance = allowance; + this.threshold = threshold; + } + + /** + * Feeds one sample into the detector. + * + * @param value the incoming sample + * @return {@link ShiftSignal#NONE} while the stream stays in control, otherwise the direction of + * the detected shift; the accumulated sums are cleared on an alarm + * @throws IllegalArgumentException if {@code value} is NaN or infinite + */ + public ShiftSignal accept(double value) { + requireFinite(value, "sample"); + count++; + + double normalized = (value - target) / standardDeviation; + upperSum = Math.max(0.0, upperSum + normalized - allowance); + lowerSum = Math.max(0.0, lowerSum - normalized - allowance); + + if (upperSum > threshold) { + lastSignal = ShiftSignal.UPWARD; + } else if (lowerSum > threshold) { + lastSignal = ShiftSignal.DOWNWARD; + } else { + lastSignal = ShiftSignal.NONE; + } + + if (lastSignal.isAlarm()) { + alarmCount++; + upperSum = 0.0; + lowerSum = 0.0; + } + return lastSignal; + } + + /** + * Runs the detector over a whole signal. + * + * @param signal the samples to inspect + * @return a new array of the same length holding the verdict for every sample + * @throws IllegalArgumentException if any sample is NaN or infinite + * @throws NullPointerException if {@code signal} is {@code null} + */ + public ShiftSignal[] scan(double[] signal) { + ShiftSignal[] signals = new ShiftSignal[signal.length]; + for (int i = 0; i < signal.length; i++) { + signals[i] = accept(signal[i]); + } + return signals; + } + + /** + * Returns the evidence accumulated in favour of an upward shift. + * + * @return the one-sided upper sum, never negative + */ + public double upperSum() { + return upperSum; + } + + /** + * Returns the evidence accumulated in favour of a downward shift. + * + * @return the one-sided lower sum, never negative + */ + public double lowerSum() { + return lowerSum; + } + + /** + * Returns the verdict on the most recent sample. + * + * @return the last signal, {@link ShiftSignal#NONE} before the first sample + */ + public ShiftSignal lastSignal() { + return lastSignal; + } + + /** + * Returns how many samples have been inspected since the last reset. + * + * @return the sample count + */ + public long count() { + return count; + } + + /** + * Returns how many alarms have been raised since the last reset. + * + * @return the alarm count + */ + public long alarmCount() { + return alarmCount; + } + + /** + * Returns the expected level of the stream. + * + * @return the target given at construction time + */ + public double target() { + return target; + } + + /** + * Returns the assumed noise level. + * + * @return the standard deviation given at construction time + */ + public double standardDeviation() { + return standardDeviation; + } + + /** + * Returns the per-step allowance. + * + * @return the allowance given at construction time + */ + public double allowance() { + return allowance; + } + + /** + * Returns the decision threshold. + * + * @return the threshold given at construction time + */ + public double threshold() { + return threshold; + } + + /** + * Clears the accumulated evidence and the counters. + */ + public void reset() { + upperSum = 0.0; + lowerSum = 0.0; + count = 0; + alarmCount = 0; + lastSignal = ShiftSignal.NONE; + } + + @Override + public String toString() { + return "CusumDetector{target=" + target + ", upperSum=" + upperSum + ", lowerSum=" + lowerSum + ", alarms=" + alarmCount + '}'; + } + + private static void requireFinite(double value, String name) { + if (!Double.isFinite(value)) { + throw new IllegalArgumentException("The " + name + " must be finite, but was " + value); + } + } + + /** + * What the detector reports after looking at one sample: either the stream still behaves as + * expected, or its level has shifted, in one direction or the other. + */ + public enum ShiftSignal { + + /** No evidence of a change; the stream is in control. */ + NONE, + + /** The level of the stream has moved above the target. */ + UPWARD, + + /** The level of the stream has moved below the target. */ + DOWNWARD; + + /** + * Tells whether this signal reports a change. + * + * @return {@code true} for {@link #UPWARD} and {@link #DOWNWARD} + */ + public boolean isAlarm() { + return this != NONE; + } + } +} diff --git a/src/test/java/com/thealgorithms/streaming/CusumDetectorTest.java b/src/test/java/com/thealgorithms/streaming/CusumDetectorTest.java new file mode 100644 index 000000000000..9c0e4a49f56a --- /dev/null +++ b/src/test/java/com/thealgorithms/streaming/CusumDetectorTest.java @@ -0,0 +1,191 @@ +package com.thealgorithms.streaming; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.thealgorithms.streaming.CusumDetector.ShiftSignal; +import java.util.Random; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class CusumDetectorTest { + + /** + * Index of the first alarm in a sequence of verdicts, or {@code -1} if there is none. + */ + private static int firstAlarm(ShiftSignal[] signals) { + for (int i = 0; i < signals.length; i++) { + if (signals[i].isAlarm()) { + return i; + } + } + return -1; + } + + @Test + void rejectsInvalidConfiguration() { + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(Double.NaN, 1.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 0.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, -1.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, -0.5, 5.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, 0.5, 0.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, 0.5, Double.NaN)); + } + + @ParameterizedTest + @ValueSource(doubles = {Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY}) + void rejectsNonFiniteSamples(double value) { + CusumDetector detector = new CusumDetector(0.0, 1.0); + assertThrows(IllegalArgumentException.class, () -> detector.accept(value)); + } + + @Test + void exposesItsConfiguration() { + CusumDetector detector = new CusumDetector(20.0, 0.5, 0.4, 4.0); + assertEquals(20.0, detector.target()); + assertEquals(0.5, detector.standardDeviation()); + assertEquals(0.4, detector.allowance()); + assertEquals(4.0, detector.threshold()); + assertEquals(ShiftSignal.NONE, detector.lastSignal()); + assertEquals(0L, detector.count()); + assertEquals(0L, detector.alarmCount()); + } + + @Test + @DisplayName("a stream sitting exactly on target never raises an alarm") + void staysQuietOnTarget() { + CusumDetector detector = new CusumDetector(10.0, 1.0); + for (int i = 0; i < 100_000; i++) { + assertEquals(ShiftSignal.NONE, detector.accept(10.0)); + } + assertEquals(0.0, detector.upperSum()); + assertEquals(0.0, detector.lowerSum()); + assertEquals(0L, detector.alarmCount()); + assertEquals(100_000L, detector.count()); + } + + @Test + @DisplayName("the toll keeps in-control noise from accumulating") + void staysMostlyQuietOnInControlNoise() { + Random random = new Random(20240517L); + CusumDetector detector = new CusumDetector(0.0, 1.0, 0.5, 8.0); + for (int i = 0; i < 2_000; i++) { + detector.accept(random.nextGaussian()); + } + assertTrue(detector.alarmCount() <= 2, "raised " + detector.alarmCount() + " false alarms in 2000 samples"); + } + + @Test + @DisplayName("a persistent upward shift is detected within a handful of samples") + void detectsAnUpwardShift() { + double[] signal = new double[60]; + Random random = new Random(11L); + for (int i = 0; i < signal.length; i++) { + signal[i] = (i < 30 ? 0.0 : 2.0) + 0.1 * random.nextGaussian(); + } + + CusumDetector detector = new CusumDetector(0.0, 1.0, 0.5, 5.0); + ShiftSignal[] verdicts = detector.scan(signal); + int alarm = firstAlarm(verdicts); + assertTrue(alarm >= 30, "alarmed before the shift, at index " + alarm); + assertTrue(alarm <= 35, "took too long to alarm, index " + alarm); + assertEquals(ShiftSignal.UPWARD, verdicts[alarm]); + } + + @Test + void detectsADownwardShift() { + double[] signal = new double[60]; + Random random = new Random(12L); + for (int i = 0; i < signal.length; i++) { + signal[i] = (i < 30 ? 100.0 : 96.0) + 0.5 * random.nextGaussian(); + } + + CusumDetector detector = new CusumDetector(100.0, 1.0, 0.5, 5.0); + ShiftSignal[] verdicts = detector.scan(signal); + int alarm = firstAlarm(verdicts); + assertTrue(alarm >= 30 && alarm <= 35, "alarm at index " + alarm); + assertEquals(ShiftSignal.DOWNWARD, verdicts[alarm]); + } + + @Test + @DisplayName("catches a drift far too small to see in any single sample") + void detectsASmallPersistentDrift() { + double[] signal = new double[400]; + Random random = new Random(13L); + for (int i = 0; i < signal.length; i++) { + // The stream is noise free until the drift starts, so an early alarm can only be a + // reaction to the drift itself. + signal[i] = i < 200 ? 0.0 : 0.5 + random.nextGaussian(); + } + + CusumDetector detector = new CusumDetector(0.0, 1.0, 0.25, 5.0); + int alarm = firstAlarm(detector.scan(signal)); + assertTrue(alarm >= 200, "alarmed before the drift, at index " + alarm); + assertTrue(alarm < 260, "a half sigma drift should be caught quickly, but took until " + alarm); + } + + @Test + void sumsNeverGoNegativeAndClearOnAlarm() { + CusumDetector detector = new CusumDetector(0.0, 1.0, 0.5, 3.0); + for (int i = 0; i < 50; i++) { + detector.accept(-5.0); + assertTrue(detector.upperSum() >= 0.0); + assertTrue(detector.lowerSum() >= 0.0); + } + assertTrue(detector.alarmCount() > 1, "a sustained shift should keep alarming"); + } + + @Test + void scanReportsOneVerdictPerSample() { + CusumDetector detector = new CusumDetector(0.0, 1.0); + ShiftSignal[] signals = detector.scan(new double[] {0.0, 0.0, 0.0}); + assertEquals(3, signals.length); + for (ShiftSignal signal : signals) { + assertEquals(ShiftSignal.NONE, signal); + assertFalse(signal.isAlarm()); + } + } + + @Test + void resetClearsTheEvidence() { + CusumDetector detector = new CusumDetector(0.0, 1.0, 0.5, 5.0); + for (int i = 0; i < 10; i++) { + detector.accept(1.5); + } + assertTrue(detector.upperSum() > 0.0 || detector.alarmCount() > 0); + + detector.reset(); + assertEquals(0.0, detector.upperSum()); + assertEquals(0.0, detector.lowerSum()); + assertEquals(0L, detector.count()); + assertEquals(0L, detector.alarmCount()); + assertEquals(ShiftSignal.NONE, detector.lastSignal()); + } + + @Test + void toStringMentionsTheState() { + CusumDetector detector = new CusumDetector(7.0, 1.0); + assertTrue(detector.toString().contains("target=7.0"), detector.toString()); + } + + @Test + void shiftSignalDescribesItself() { + assertFalse(ShiftSignal.NONE.isAlarm()); + assertTrue(ShiftSignal.UPWARD.isAlarm()); + assertTrue(ShiftSignal.DOWNWARD.isAlarm()); + assertEquals(3, ShiftSignal.values().length); + assertEquals(ShiftSignal.UPWARD, ShiftSignal.valueOf("UPWARD")); + } + + @Test + void rejectsNonFiniteConfiguration() { + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, Double.NaN)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, Double.NaN, 5.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, Double.POSITIVE_INFINITY, 5.0)); + assertThrows(IllegalArgumentException.class, () -> new CusumDetector(0.0, 1.0, 0.5, Double.POSITIVE_INFINITY)); + } +}