From 52f091f98a70a2b52ddf698fbc28557e03f5b354 Mon Sep 17 00:00:00 2001 From: Stephan Dollberg Date: Wed, 22 Apr 2026 14:39:15 +0100 Subject: [PATCH] Add option to enforce minimum segment length One problem we often see in practice is that single spikes cause changepoints. These are just false positive noise which we want to avoid. This patch adds a config to disallow changepoints that only enclose segments of a certain length. Like that we can filter out these one-event changepoints and avoid noise. Of course this will mute true changepoints in a short segment but that's fine if they are followed by another true changepoint. For example [100, 100, 130, 130, 150, 150, 150 ...] would only report the 150 one. This is fine. A single alert is good enough to get someone to look at the data. Default behaviour is unchanged. --- otava/analysis.py | 112 +++++++++++++++++++++++++++++++++++++++-- otava/main.py | 11 ++++ otava/series.py | 6 +++ tests/analysis_test.py | 98 ++++++++++++++++++++++++++++++++++++ tests/cli_help_test.py | 14 ++++-- 5 files changed, 232 insertions(+), 9 deletions(-) diff --git a/otava/analysis.py b/otava/analysis.py index 5ff8975..ef9b062 100644 --- a/otava/analysis.py +++ b/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, + ) diff --git a/otava/main.py b/otava/main.py index 8748ae2..7a8686f 100644 --- a/otava/main.py +++ b/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 diff --git a/otava/series.py b/otava/series.py index bb3bf47..96d633c 100644 --- a/otava/series.py +++ b/otava/series.py @@ -35,12 +35,14 @@ 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): @@ -48,6 +50,7 @@ def to_json(self): "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 = {} diff --git a/tests/analysis_test.py b/tests/analysis_test.py index 5d02716..4f40e2f 100644 --- a/tests/analysis_test.py +++ b/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] diff --git a/tests/cli_help_test.py b/tests/cli_help_test.py index a0dfa42..4550f82 100644 --- a/tests/cli_help_test.py +++ b/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