Add option to enforce minimum segment length

otava/analysis.py

@@ -188,6 +188,88 @@ def fill_missing(data: Sequence[SupportsFloat]):
         prev = data[i]
 
 
+def collapse_short_segments(
+    change_points: TtestCPList,
+    series: Sequence[SupportsFloat],
+    max_pvalue: float,
+    min_magnitude: float,
+    min_segment_len: int,
+) -> TtestCPList:
+    """
+    Normalizes weak change points by removing or collapsing too-short regimes.
+
+    This is an explicit denoising pass that runs before the regular otava merge.
+    If the weak change points induce a regime shorter than ``min_segment_len`` we
+    treat that regime as unstable noise rather than a real stationary segment.
+
+    For an edge segment we simply remove the adjacent change point. For a short
+    middle segment ``A | B | C`` we remove both boundaries around ``B`` and, if
+    the two stable neighbors ``A`` and ``C`` still differ significantly, replace
+    them with a single synthetic change point at the start of ``C``.
+
+    Note that the replacement change point is intentionally a denoised summary:
+    its statistics describe the stable neighbors ``A`` vs ``C``, not the literal
+    contiguous split in the raw series at that index. This keeps one-point spikes
+    from polluting the reported magnitude and p-value while still surfacing the
+    stable post-spike change.
+    """
+    if min_segment_len <= 1:
+        return change_points
+
+    tester = TTestSignificanceTester(max_pvalue)
+
+    def interval_len(interval: slice) -> int:
+        start = 0 if interval.start is None else interval.start
+        stop = len(series) if interval.stop is None else interval.stop
+        return stop - start
+
+    while change_points:
+        intervals = tester.get_intervals(change_points)
+        for interval_index, interval in enumerate(intervals):
+            if interval_len(interval) >= min_segment_len:
+                continue
+
+            if interval_index == 0:
+                # A short leading segment has only one adjacent boundary, so the
+                # most conservative option is to drop that candidate entirely.
+                del change_points[0]
+            elif interval_index == len(intervals) - 1:
+                # Same for a short trailing segment.
+                del change_points[-1]
+            else:
+                left_interval = intervals[interval_index - 1]
+                right_interval = intervals[interval_index + 1]
+                right_change_point = change_points[interval_index]
+                # The short middle regime is treated as transient noise. Compare
+                # the stable neighbors directly and, if they still differ enough,
+                # emit a single replacement change point at the start of the
+                # right-hand stable regime.
+                replacement_stats = tester.compare(
+                    series[left_interval],
+                    series[right_interval],
+                )
+
+                # Remove both boundaries that created the short middle regime.
+                del change_points[interval_index - 1 : interval_index + 1]
+
+                replacement_cp = ChangePoint(
+                    index=right_change_point.index,
+                    qhat=right_change_point.qhat,
+                    stats=replacement_stats,
+                )
+                if (
+                    tester.is_significant(replacement_cp)
+                    and replacement_cp.stats.change_magnitude() > min_magnitude
+                ):
+                    change_points.insert(interval_index - 1, replacement_cp)
+
+            break
+        else:
+            return change_points
+
+    return change_points
+
+
 def merge(
     change_points: TtestCPList, series: Sequence[SupportsFloat], max_pvalue: float, min_magnitude: float
 ) -> TtestCPList:
@@ -201,7 +283,6 @@ def merge(
     """
     tester = TTestSignificanceTester(max_pvalue)
     while change_points:
-
         # Select the change point with weakest unacceptable P-value
         # If all points have acceptable P-values, select the change-point with
         # the least relative change:
@@ -292,8 +373,13 @@ def compute_change_points_orig(series: Sequence[SupportsFloat], max_pvalue: floa
 
 
 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
+    series: Sequence[SupportsFloat],
+    window_len: int = 50,
+    max_pvalue: float = 0.001,
+    min_magnitude: float = 0.0,
+    min_segment_len: int = 1,
+    new_data: Optional[int] = None,
+    old_weak_cp: Optional[GenCPList] = None,
 ) -> Tuple[GenCPList, Optional[GenCPList]]:
     """
     Change Point detection algorithm described in "Hunter: Using Change Point Detection to Hunt for Performance
@@ -320,7 +406,25 @@ def compute_change_points(
         2. Merge step:
             - Filters out weak change points recursively going bottom-up, keeping only high-quality change points, i.e., the
               ones that meet either a p-value threshold criteria or relative magnitude change criteria.
+
+    When ``min_segment_len > 1`` there is an additional normalization step between
+    split and merge which collapses weak change points that form regimes shorter
+    than the requested minimum length.
     """
     first_pass_pvalue = max_pvalue * 10 if max_pvalue < 0.05 else (max_pvalue * 2 if max_pvalue < 0.5 else max_pvalue)
     weak_change_points = split(series, window_len, first_pass_pvalue, new_points=new_data, old_cp=old_weak_cp)
-    return merge(weak_change_points, series, max_pvalue, min_magnitude), weak_change_points
+    return (
+        merge(
+            collapse_short_segments(
+                weak_change_points,
+                series,
+                max_pvalue,
+                min_magnitude,
+                min_segment_len,
+            ),
+            series,
+            max_pvalue,
+            min_magnitude,
+        ),
+        weak_change_points,
+    )

otava/main.py

@@ -433,6 +433,15 @@ def setup_analysis_options_parser(parser: argparse.ArgumentParser):
         help="use the original edivisive algorithm with no windowing "
         "and weak change points analysis improvements",
     )
+    parser.add_argument(
+        "--min-segment-len",
+        default=1,
+        type=int,
+        dest="min_segment_len",
+        help="minimum accepted segment length between change points; "
+        "segments with length >= this value are kept, shorter regimes are "
+        "removed",
+    )
 
 
 def analysis_options_from_args(args: argparse.Namespace) -> AnalysisOptions:
@@ -443,6 +452,8 @@ 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.min_segment_len is not None:
+        conf.min_segment_len = args.min_segment_len
     if args.orig_edivisive is not None:
         conf.orig_edivisive = args.orig_edivisive
     return conf

otava/series.py

@@ -35,19 +35,22 @@ class AnalysisOptions:
     window_len: int
     max_pvalue: float
     min_magnitude: float
+    min_segment_len: int
     orig_edivisive: bool
 
     def __init__(self):
         self.window_len = 50
         self.max_pvalue = 0.001
         self.min_magnitude = 0.0
+        self.min_segment_len = 1
         self.orig_edivisive = False
 
     def to_json(self):
         return {
             "window_len": self.window_len,
             "max_pvalue": self.max_pvalue,
             "min_magnitude": self.min_magnitude,
+            "min_segment_len": self.min_segment_len,
             "orig_edivisive": self.orig_edivisive
         }
 
@@ -255,6 +258,7 @@ def __compute_change_points(
                     window_len=options.window_len,
                     max_pvalue=options.max_pvalue,
                     min_magnitude=options.min_magnitude,
+                    min_segment_len=options.min_segment_len,
                 )
                 for c in weak_cps:
                     weak_change_points[metric].append(
@@ -375,6 +379,7 @@ def append(self, time, new_data, attributes):
                 window_len=self.options.window_len,
                 max_pvalue=self.options.max_pvalue,
                 min_magnitude=self.options.min_magnitude,
+                min_segment_len=self.options.min_segment_len,
                 new_data=len(new_data[metric]),
                 old_weak_cp=self.weak_change_points.get(metric, [])
             )
@@ -479,6 +484,7 @@ def from_json(cls, analyzed_json):
         new_options.window_len = analyzed_json["options"]["window_len"]
         new_options.max_pvalue = analyzed_json["options"]["max_pvalue"]
         new_options.min_magnitude = analyzed_json["options"]["min_magnitude"]
+        new_options.min_segment_len = analyzed_json["options"].get("min_segment_len", 1)
         new_options.orig_edivisive = analyzed_json["options"]["orig_edivisive"]
 
         new_change_points = {}

tests/analysis_test.py

@@ -108,3 +108,101 @@ def test_significance_tester():
     cp = tester.change_point(candidate, series, intervals=[slice(None, None)])
     assert tester.is_significant(cp)
     assert 0.00 < cp.stats.pvalue < 0.001
+
+
+def test_single_point_spike_is_removed_by_min_segment_len():
+    series = [100, 100, 100, 100, 300, 100, 100, 100, 100]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == []
+
+
+def test_clean_step_is_preserved_by_min_segment_len():
+    series = [100, 100, 100, 100, 110, 110, 110, 110, 110]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == [4]
+
+
+def test_spike_then_shift_collapses_to_real_change_point():
+    series = [100, 100, 100, 100, 300, 110, 110, 110, 110]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == [5]
+
+
+def test_later_step_after_short_regime_is_ignored_when_segment_too_short():
+    series = [100, 100, 100, 100, 300, 100, 100, 110, 110]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == []
+
+
+def test_short_regime_is_ignored_when_shorter_than_min_segment_len():
+    series = [100, 100, 100, 100, 300, 300, 100, 100, 100]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == []
+
+
+def test_multiple_sustained_steps_are_preserved_by_min_segment_len():
+    series = [100, 100, 100, 100, 130, 130, 130, 130, 150, 150, 150, 150]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == [4, 8]
+
+
+def test_two_point_middle_regime_is_suppressed_by_min_segment_len():
+    series = [100, 100, 100, 100, 130, 130, 150, 150, 150, 150]
+
+    cps, _ = compute_change_points(
+        series,
+        window_len=5,
+        max_pvalue=0.001,
+        min_magnitude=0.01,
+        min_segment_len=3,
+    )
+
+    assert [cp.index for cp in cps] == [6]

tests/cli_help_test.py

@@ -136,15 +136,16 @@ def test_otava_analyze_help_output():
     # Python 3.13+ formats mutually exclusive group usage and option aliases differently
     if IS_PYTHON_313_PLUS:
         usage_filter_lines = """\
-                     [--attrs LIST] [--since-commit STRING | --since-version STRING |
-                     --since DATE] [--until-commit STRING | --until-version STRING | --until DATE]"""
-        magnitude_option = "  -M, --magnitude MAGNITUDE"
+                     [--attrs LIST]
+                     [--since-commit STRING | --since-version STRING | --since DATE]
+                     [--until-commit STRING | --until-version STRING | --until DATE]"""
     else:
         usage_filter_lines = """\
                      [--attrs LIST]
                      [--since-commit STRING | --since-version STRING | --since DATE]
                      [--until-commit STRING | --until-version STRING | --until DATE]"""
-        magnitude_option = "  -M MAGNITUDE, --magnitude MAGNITUDE"
+
+    magnitude_option = "  -M MAGNITUDE, --magnitude MAGNITUDE"
 
     usage_and_options = f"""\
 usage: otava analyze [-h] [--config-file CONFIG_FILE] [--graphite-url GRAPHITE_URL]
@@ -162,7 +163,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]
+                     [--orig-edivisive ORIG_EDIVISIVE] [--min-segment-len MIN_SEGMENT_LEN]
                      tests [tests ...]
 
 positional arguments:
@@ -218,6 +219,9 @@ def test_otava_analyze_help_output():
   --orig-edivisive ORIG_EDIVISIVE
                         use the original edivisive algorithm with no windowing and weak change
                         points analysis improvements
+  --min-segment-len MIN_SEGMENT_LEN
+                        minimum accepted segment length between change points; segments with
+                        length >= this value are kept, shorter regimes are removed
 
 Graphite Options:
   Options for Graphite configuration