Fix change point detection for uncertain non-stationary distributions. (elastic#119578)

* Fix change point detection for uncertain non-stationary distributions.

* Replace -1 by ChangeType.NO_CHANGE_POINT

Author: jan-elastic

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

@@ -189,7 +189,7 @@ private TestStats testTrendVs(TestStats H0, double[] values, double[] weights) {
     private TestStats testStepChangeVs(TestStats H0, double[] values, double[] weights, int[] candidateChangePoints) {
 
         double vStep = Double.MAX_VALUE;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -226,7 +226,7 @@ private TestStats testStepChangeVs(TestStats H0, double[] values, double[] weigh
     private TestStats testTrendChangeVs(TestStats H0, double[] values, double[] weights, int[] candidateChangePoints) {
 
         double vChange = Double.MAX_VALUE;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -349,7 +349,7 @@ private TestStats testDistributionChange(
     ) {
 
         double maxDiff = 0.0;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -378,10 +378,12 @@ private TestStats testDistributionChange(
         // before we run the tests.
         SampleData sampleData = sample(values, weights, discoveredChangePoints);
         final double[] sampleValues = sampleData.values();
-        final double[] sampleWeights = sampleData.weights();
 
         double pValue = 1;
         for (int cp : sampleData.changePoints()) {
+            if (cp == ChangeType.NO_CHANGE_POINT) {
+                continue;
+            }
             double[] x = Arrays.copyOfRange(sampleValues, 0, cp);
             double[] y = Arrays.copyOfRange(sampleValues, cp, sampleValues.length);
             double statistic = KOLMOGOROV_SMIRNOV_TEST.kolmogorovSmirnovStatistic(x, y);
@@ -451,7 +453,7 @@ private record TestStats(Type type, double pValue, double var, double nParams, i
         }
 
         TestStats(Type type, double pValue, double var, double nParams, DataStats dataStats) {
-            this(type, pValue, var, nParams, -1, dataStats);
+            this(type, pValue, var, nParams, ChangeType.NO_CHANGE_POINT, dataStats);
         }
 
         boolean accept(double pValueThreshold) {

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

@@ -47,7 +47,7 @@ public InternalAggregation doReduce(InternalAggregations aggregations, Aggregati
         ChangeType change = ChangePointDetector.getChangeType(bucketValues);
 
         ChangePointBucket changePointBucket = null;
-        if (change.changePoint() >= 0) {
+        if (change.changePoint() != ChangeType.NO_CHANGE_POINT) {
             changePointBucket = extractBucket(bucketsPaths()[0], aggregations, change.changePoint()).map(
                 b -> new ChangePointBucket(b.getKey(), b.getDocCount(), b.getAggregations())
             ).orElse(null);

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

@@ -21,8 +21,10 @@
  */
 public interface ChangeType extends NamedWriteable, NamedXContentObject {
 
+    int NO_CHANGE_POINT = -1;
+
     default int changePoint() {
-        return -1;
+        return NO_CHANGE_POINT;
     }
 
     default double pValue() {

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangeDetectorTests.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.DoubleStream;
 
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThan;
@@ -243,4 +244,14 @@ public void testProblemDistributionChange() {
         ChangeType type = new ChangeDetector(bucketValues).detect(0.05);
         assertThat(type, instanceOf(ChangeType.DistributionChange.class));
     }
+
+    public void testUncertainNonStationary() {
+        MlAggsHelper.DoubleBucketValues bucketValues = new MlAggsHelper.DoubleBucketValues(
+            null,
+            new double[] { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 700, 735, 715 }
+        );
+        ChangeType type = new ChangeDetector(bucketValues).detect(0.01);
+        assertThat(type, instanceOf(ChangeType.NonStationary.class));
+        assertThat(((ChangeType.NonStationary) type).getTrend(), equalTo("increasing"));
+    }
 }