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