Public entrypoint for change point detection

Author: jan-elastic

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangeDetector.java

@@ -36,18 +36,19 @@ public class ChangeDetector {
 
     private static final Logger logger = LogManager.getLogger(ChangeDetector.class);
 
+    private final MlAggsHelper.DoubleBucketValues bucketValues;
     private final double[] values;
 
-    ChangeDetector(double[] values) {
-        this.values = values;
+    ChangeDetector(MlAggsHelper.DoubleBucketValues bucketValues) {
+        this.bucketValues = bucketValues;
+        this.values = bucketValues.getValues();
     }
 
-    public ChangeType detect(double pValueThreshold, MlAggsHelper.DoubleBucketValues bucketValues) {
+    ChangeType detect(double pValueThreshold) {
         return testForChange(pValueThreshold).changeType(bucketValues, slope(values));
     }
 
-    // visible for testing
-    TestStats testForChange(double pValueThreshold) {
+    private TestStats testForChange(double pValueThreshold) {
 
         int[] candidateChangePoints = computeCandidateChangePoints(values);
         logger.trace("candidatePoints: [{}]", Arrays.toString(candidateChangePoints));
@@ -415,8 +416,7 @@ private static double fDistribSf(double numeratorDegreesOfFreedom, double denomi
         );
     }
 
-    // visible for testing
-    enum Type {
+    private enum Type {
         STATIONARY,
         NON_STATIONARY,
         STEP_CHANGE,
@@ -442,8 +442,7 @@ public String toString() {
         }
     }
 
-    // visible for testing
-    record TestStats(Type type, double pValue, double var, double nParams, int changePoint, DataStats dataStats) {
+    private record TestStats(Type type, double pValue, double var, double nParams, int changePoint, DataStats dataStats) {
         TestStats(Type type, double pValue, int changePoint, DataStats dataStats) {
             this(type, pValue, 0.0, 0.0, changePoint, dataStats);
         }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangePointAggregator.java

@@ -58,30 +58,8 @@ public InternalAggregation doReduce(InternalAggregations aggregations, Aggregati
             );
         }
         MlAggsHelper.DoubleBucketValues bucketValues = maybeBucketValues.get();
-        if (bucketValues.getValues().length < (2 * MINIMUM_BUCKETS) + 2) {
-            return new InternalChangePointAggregation(
-                name(),
-                metadata(),
-                null,
-                new ChangeType.Indeterminable(
-                    "not enough buckets to calculate change_point. Requires at least ["
-                        + ((2 * MINIMUM_BUCKETS) + 2)
-                        + "]; found ["
-                        + bucketValues.getValues().length
-                        + "]"
-                )
-            );
-        }
 
-        ChangeType spikeOrDip = testForSpikeOrDip(bucketValues, P_VALUE_THRESHOLD);
-
-        // Test for change step, trend and distribution changes.
-        ChangeType change = testForChange(bucketValues, changePValueThreshold(bucketValues.getValues().length));
-        logger.trace("change p-value: [{}]", change.pValue());
-
-        if (spikeOrDip.pValue() < change.pValue()) {
-            change = spikeOrDip;
-        }
+        ChangeType change = getChangeType(bucketValues);
 
         ChangePointBucket changePointBucket = null;
         if (change.changePoint() >= 0) {
@@ -93,20 +71,35 @@ public InternalAggregation doReduce(InternalAggregations aggregations, Aggregati
         return new InternalChangePointAggregation(name(), metadata(), changePointBucket, change);
     }
 
+    static ChangeType getChangeType(MlAggsHelper.DoubleBucketValues bucketValues) {
+        if (bucketValues.getValues().length < (2 * MINIMUM_BUCKETS) + 2) {
+            return new ChangeType.Indeterminable(
+                "not enough buckets to calculate change_point. Requires at least ["
+                    + ((2 * MINIMUM_BUCKETS) + 2)
+                    + "]; found ["
+                    + bucketValues.getValues().length
+                    + "]"
+            );
+        }
+
+        ChangeType spikeOrDip = testForSpikeOrDip(bucketValues, P_VALUE_THRESHOLD);
+        ChangeType change = new ChangeDetector(bucketValues).detect(changePValueThreshold(bucketValues.getValues().length));
+        logger.trace("change p-value: [{}]", change.pValue());
+        if (spikeOrDip.pValue() < change.pValue()) {
+            change = spikeOrDip;
+        }
+        return change;
+    }
+
     static ChangeType testForSpikeOrDip(MlAggsHelper.DoubleBucketValues bucketValues, double pValueThreshold) {
         try {
-            SpikeAndDipDetector detect = new SpikeAndDipDetector(bucketValues.getValues());
-            ChangeType result = detect.detect(pValueThreshold, bucketValues);
+            SpikeAndDipDetector detect = new SpikeAndDipDetector(bucketValues);
+            ChangeType result = detect.detect(pValueThreshold);
             logger.trace("spike or dip p-value: [{}]", result.pValue());
             return result;
         } catch (NotStrictlyPositiveException nspe) {
             logger.debug("failure testing for dips and spikes", nspe);
+            return new Indeterminable("failure testing for dips and spikes");
         }
-        return new Indeterminable("failure testing for dips and spikes");
-    }
-
-    static ChangeType testForChange(MlAggsHelper.DoubleBucketValues bucketValues, double pValueThreshold) {
-        double[] timeWindow = bucketValues.getValues();
-        return new ChangeDetector(timeWindow).detect(pValueThreshold, bucketValues);
     }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/aggs/changepoint/SpikeAndDipDetector.java

@@ -92,6 +92,7 @@ private double[] removeIf(ExcludedPredicate should, double[] values) {
         return newValues;
     }
 
+    private final MlAggsHelper.DoubleBucketValues bucketValues;
     private final int numValues;
     private final int dipIndex;
     private final int spikeIndex;
@@ -100,7 +101,10 @@ private double[] removeIf(ExcludedPredicate should, double[] values) {
     private final KDE spikeTestKDE;
     private final KDE dipTestKDE;
 
-    SpikeAndDipDetector(double[] values) {
+    SpikeAndDipDetector(MlAggsHelper.DoubleBucketValues bucketValues) {
+        this.bucketValues = bucketValues;
+        double[] values = bucketValues.getValues();
+
         numValues = values.length;
 
         if (values.length < 4) {
@@ -134,7 +138,7 @@ private double[] removeIf(ExcludedPredicate should, double[] values) {
         spikeTestKDE = new KDE(spikeKDEValues, 1.36);
     }
 
-    ChangeType detect(double pValueThreshold, MlAggsHelper.DoubleBucketValues bucketValues) {
+    ChangeType detect(double pValueThreshold) {
         if (dipIndex == -1 || spikeIndex == -1) {
             return new ChangeType.Indeterminable(
                 "not enough buckets to check for dip or spike. Requires at least [3]; found [" + numValues + "]"