Add --deterministic-edivisive variation

otava/analysis.py

@@ -291,6 +291,44 @@ def compute_change_points_orig(series: Sequence[SupportsFloat], max_pvalue: floa
     return change_points, None
 
 
+def compute_change_points_deterministic(series: Sequence[SupportsFloat], max_pvalue: float = 0.001, min_magnitude: float = 0.0) -> Tuple[PermCPList, Optional[PermCPList]]:
+    """
+    Same as the original algorithm but with deterministic Student T significance test at the end.
+
+    The motivation for this variation follows from fixing the bug explained at the top of https://github.com/apache/otava/pull/96
+    The intuition is that the split-merge approach introduced by Datastax is addressing the same problem that the _kappa_ variable
+    does in the original paper. Now that we compute correctly over all values of kappa, the split-merge part should be unnecessary,
+    as the original algorithm with kappa bug fixed, will find the same change points, and more. Therefore the conclusion is we want
+    to go back as much as possible to the original and real algorithm from the Matteson & James paper. But even then, we find that
+    Student T as significance test is both much faster but also qualitatively produces better results for the use case we're in at least,
+    that we want to continue using T test and not random permutations for the significance test.
+
+    TBD: Whether weak change points are still helpful or not. By reading the problem they fix appears unrelated from the split-merge vs **kappa** symptoms.
+
+    TODO: Support incremental e-divisive. This was easy to implement on top of the split-merge variation. Not clear what is the correct way here.
+    An easy solution is to rerun from the last change-point, but the problem is the last change point could itself be influenced by the new data
+    appended.
+    """
+    tester = TTestSignificanceTester(max_pvalue=max_pvalue)
+    detector = ChangePointDetector(significance_tester=tester, calculator=PairDistanceCalculator)
+    all_change_points = detector.get_change_points(series=series)
+    if min_magnitude > 0.0:
+        above_threshold_change_points = [cp for cp in all_change_points if cp.stats.change_magnitude() >= min_magnitude]
+    else:
+        above_threshold_change_points = all_change_points
+    return above_threshold_change_points, all_change_points
+
+
+def compute_change_points_split(
+    series: Sequence[SupportsFloat], window_len: int = 50, max_pvalue: float = 0.001, min_magnitude: float = 0.0,
+    new_data: Optional[int] = None, old_weak_cp: Optional[GenCPList] = None
+) -> Tuple[GenCPList, Optional[GenCPList]]:
+    """
+    This function added mainly for symmetry and anticipating a future where this is no longer the default
+    """
+    return compute_change_points(series, window_len, max_pvalue, min_magnitude, new_data, old_weak_cp)
+
+
 def compute_change_points(
     series: Sequence[SupportsFloat], window_len: int = 50, max_pvalue: float = 0.001, min_magnitude: float = 0.0,
     new_data: Optional[int] = None, old_weak_cp: Optional[GenCPList] = None

otava/main.py

@@ -425,14 +425,28 @@ def setup_analysis_options_parser(parser: argparse.ArgumentParser):
         "as noise so it is best to keep it short enough to include not more "
         "than a few change points (optimally at most 1)",
     )
-    parser.add_argument(
+    ediv_group = parser.add_mutually_exclusive_group()
+    ediv_group.add_argument(
         "--orig-edivisive",
         action="store_true",
-        default=False,
         dest="orig_edivisive",
         help="use the original edivisive algorithm with no windowing "
         "and weak change points analysis improvements",
     )
+    ediv_group.add_argument(
+        "--deterministic-edivisive",
+        action="store_true",
+        dest="deterministic_edivisive",
+        help="EXPERIMENTAL: use the original edivisive algorithm, but using "
+        "Student T for significance test. (TBD: May include weak change points later.)",
+    )
+    ediv_group.add_argument(
+        "--split-edivisive",
+        action="store_true",
+        dest="split_edivisive",
+        help="use 'hunter' version of this algorithm, from 2023, featuring "
+        "split of data into smaller windows, weak change points and Student T test. (Default)",
+    )
 
 
 def analysis_options_from_args(args: argparse.Namespace) -> AnalysisOptions:
@@ -443,8 +457,14 @@ def analysis_options_from_args(args: argparse.Namespace) -> AnalysisOptions:
         conf.min_magnitude = args.magnitude
     if args.window is not None:
         conf.window_len = args.window
-    if args.orig_edivisive is not None:
-        conf.orig_edivisive = args.orig_edivisive
+
+    conf.orig_edivisive = args.orig_edivisive
+    conf.deterministic_edivisive = args.deterministic_edivisive
+    conf.split_edivisive = args.split_edivisive
+    if not (args.split_edivisive or args.deterministic_edivisive or args.orig_edivisive):
+        # Default:
+        conf.split_edivisive = True
+
     return conf
 
 

otava/series.py

@@ -25,6 +25,7 @@
     TTestSignificanceTester,
     TTestStats,
     compute_change_points,
+    compute_change_points_deterministic,
     compute_change_points_orig,
     fill_missing,
 )
@@ -37,19 +38,25 @@ class AnalysisOptions:
     max_pvalue: float
     min_magnitude: float
     orig_edivisive: bool
+    deterministic_edivisive: bool
+    split_edivisive: bool
 
     def __init__(self):
         self.window_len = 50
         self.max_pvalue = 0.001
         self.min_magnitude = 0.0
         self.orig_edivisive = False
+        self.deterministic_edivisive = False
+        self.split_edivisive = True
 
     def to_json(self):
         return {
             "window_len": self.window_len,
             "max_pvalue": self.max_pvalue,
             "min_magnitude": self.min_magnitude,
-            "orig_edivisive": self.orig_edivisive
+            "orig_edivisive": self.orig_edivisive,
+            "deterministic_edivisive": self.deterministic_edivisive,
+            "split_edivisive": self.split_edivisive,
         }
 
 
@@ -259,6 +266,20 @@ def __compute_change_points(
                             time=series.time[cp_ttest.index], metric=metric, stats=cp_ttest.stats
                         )
                     )
+            elif options.deterministic_edivisive:
+                # weak_change_points == change_points when min_magnitude == 0.
+                # when min_magnitude > 0 then change_points is the subset where the change was >= min_magnitude.
+                change_points, weak_cps = compute_change_points_deterministic(values, max_pvalue=options.max_pvalue, min_magnitude=options.min_magnitude)
+                tester = TTestSignificanceTester(options.max_pvalue)
+                intervals = tester.get_intervals(change_points)
+                for c in change_points:
+                    cp_ttest = tester.change_point(c.to_candidate(), values, intervals)
+                    result[metric].append(
+                        ChangePoint(
+                            index=cp_ttest.index, qhat=cp_ttest.qhat,
+                            time=series.time[cp_ttest.index], metric=metric, stats=cp_ttest.stats
+                        )
+                    )
             else:
                 change_points, weak_cps = compute_change_points(
                     values,
@@ -278,8 +299,7 @@ def __compute_change_points(
                             index=c.index, qhat=0.0, time=series.time[c.index], metric=metric, stats=c.stats
                         )
                     )
-        # If you got an exception and are wondering about the next row...
-        # weak_cps is an optimization which you can ignore
+
         return result, weak_change_points
 
     @staticmethod

tests/cli_help_test.py

@@ -162,7 +162,7 @@ def test_otava_analyze_help_output():
                      [--output {{log,json,regressions_only}}] [--branch [STRING]] [--metrics LIST]
 {usage_filter_lines}
                      [--last COUNT] [-P, --p-value PVALUE] [-M MAGNITUDE] [--window WINDOW]
-                     [--orig-edivisive]
+                     [--orig-edivisive | --deterministic-edivisive | --split-edivisive]
                      tests [tests ...]
 
 positional arguments:
@@ -217,6 +217,11 @@ def test_otava_analyze_help_output():
                         enough to include not more than a few change points (optimally at most 1)
   --orig-edivisive      use the original edivisive algorithm with no windowing and weak change
                         points analysis improvements
+  --deterministic-edivisive
+                        EXPERIMENTAL: use the original edivisive algorithm, but using Student T
+                        for significance test. (TBD: May include weak change points later.)
+  --split-edivisive     use 'hunter' version of this algorithm, from 2023, featuring split of data
+                        into smaller windows, weak change points and Student T test. (Default)
 
 Graphite Options:
   Options for Graphite configuration

tests/cli_options_test.py

@@ -0,0 +1,146 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import csv
+import tempfile
+import textwrap
+import unittest
+from datetime import datetime, timedelta
+from pathlib import Path
+from unittest.mock import patch
+
+import otava.series
+from otava.main import script_main
+
+
+class CliOptionsTest(unittest.TestCase):
+
+    # Test --deterministic-edivisive in various ways
+    def test_no_cli_option(self):
+        with patch('otava.series.compute_change_points') as mock_split:
+            script_main(args=[])
+            mock_split.assert_not_called()
+
+    def test_default_cli_option(self):
+        with patch('otava.series.compute_change_points') as mock_split:
+            mock_split.return_value = ([], [])
+            with tempfile.TemporaryDirectory() as td:
+                td_path = Path(td)
+                csv_path, timestamps, config_path, test_name = _create_files_in_temp_dir(td_path)
+                # _uv_run(td_path, test_name)
+                config_path_str = "" + str(config_path)
+                script_main(args=["analyze", "--config", config_path_str, test_name])
+
+            assert otava.series.compute_change_points.call_count == 2
+
+    def test_split_cli_option(self):
+        with patch('otava.series.compute_change_points') as mock_split:
+            mock_split.return_value = ([], [])
+            with tempfile.TemporaryDirectory() as td:
+                td_path = Path(td)
+                csv_path, timestamps, config_path, test_name = _create_files_in_temp_dir(td_path)
+                # _uv_run(td_path, test_name)
+                config_path_str = "" + str(config_path)
+                script_main(args=["analyze", "--config", config_path_str, "--split-edivisive", test_name])
+
+            assert otava.series.compute_change_points.call_count == 2
+
+    def test_orig_cli_option(self):
+        with patch('otava.series.compute_change_points_orig') as mock_orig:
+            mock_orig.return_value = ([], None)
+            with tempfile.TemporaryDirectory() as td:
+                td_path = Path(td)
+                csv_path, timestamps, config_path, test_name = _create_files_in_temp_dir(td_path)
+                # _uv_run(td_path, test_name)
+                config_path_str = "" + str(config_path)
+                script_main(args=["analyze", "--config", config_path_str, "--orig-edivisive", test_name])
+
+            assert otava.series.compute_change_points_orig.call_count == 2
+
+    def test_deterministic_cli_option(self):
+        with patch('otava.series.compute_change_points_deterministic') as mock_deterministic:
+            mock_deterministic.return_value = ([], None)
+            with tempfile.TemporaryDirectory() as td:
+                td_path = Path(td)
+                csv_path, timestamps, config_path, test_name = _create_files_in_temp_dir(td_path)
+                # _uv_run(td_path, test_name)
+                config_path_str = "" + str(config_path)
+                script_main(args=["analyze", "--config", config_path_str, "--deterministic-edivisive", test_name])
+
+            assert otava.series.compute_change_points_deterministic.call_count == 2
+
+
+def _create_files_in_temp_dir(td_path: Path):
+    data_dir = td_path / "data"
+    data_dir.mkdir(parents=True, exist_ok=True)
+
+    csv_path, timestamps = _create_csv_data_file_for_test(td_path)
+    config_path, test_name = _create_csv_config_file_for_test(td_path)
+
+    return csv_path, timestamps, config_path, test_name
+
+
+def _create_csv_data_file_for_test(td_path: Path):
+    # create data directory and write CSV
+    data_dir = td_path / "data"
+    csv_path = data_dir / "local_sample.csv"
+
+    # Generate some CSV content
+    now = datetime.now()
+    n = 10
+    timestamps = [now - timedelta(days=i) for i in range(n)]
+    metrics1 = [154023, 138455, 143112, 149190, 132098, 151344, 155145, 148889, 149466, 148209]
+    metrics2 = [10.43, 10.23, 10.29, 10.91, 10.34, 10.69, 9.23, 9.11, 9.13, 9.03]
+    data_points = []
+    for i in range(n):
+        data_points.append(
+            (
+                timestamps[i].strftime("%Y.%m.%d %H:%M:%S %z"),  # time
+                "aaa" + str(i),  # commit
+                metrics1[i],
+                metrics2[i],
+            )
+        )
+
+    with open(csv_path, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow(["time", "commit", "metric1", "metric2"])
+        writer.writerows(data_points)
+
+    return csv_path, timestamps
+
+
+def _create_csv_config_file_for_test(td_path: Path):
+    data_dir = td_path / "data"
+    csv_path = str(data_dir / "local_sample.csv")
+
+    config_content = textwrap.dedent(
+        """\
+        tests:
+            sample_for_test:
+                type: csv
+                file: """ + csv_path + """
+                time_column: time
+                attributes: [commit]
+                metrics: [metric1, metric2]
+                csv_options:
+                    delimiter: ","
+                    quotechar: "'"
+        """
+    )
+    config_path = td_path / "otava.yaml"
+    config_path.write_text(config_content, encoding="utf-8")
+    return config_path, "sample_for_test"