From 3c46d5c8a7dd28499fca1e7ef6c6e05423152290 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 14 Jul 2026 16:20:15 +0200 Subject: [PATCH 1/7] reduce AGENTS.md size --- AGENTS.md | 16 +--------------- 1 file changed, 1 insertion(+), 15 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index dd994543..68ee8c0d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,10 +14,7 @@ uv sync --dev uv run prek install # Run tests -uv run pytest -q -uv run pytest -s # verbose with stdout -uv run pytest --cov=. --cov-report=term-missing # with coverage -uv run pytest tests/test_foo.py # single test file +uv run pytest # Run linting/formatting (all pre-commit hooks) uv run prek run -a @@ -182,17 +179,6 @@ def set_configuration(self) -> None: Omitting either step means a `persist:` block in the YAML is silently ignored with no error. -## Code Quality - -Pre-commit hooks enforce: -- **mypy** strict mode -- **flake8** linting, **autopep8** formatting (max line 110) -- **bandit** security checks, **vulture** dead-code detection (70% threshold) -- **docformatter** docstring style - -Python 3.12 is required (see `.python-version`). - - # Git NEVER include "Co-Authored-By ..." in your commit or PR messages. From 04ca291369c9b06760d42ef2d012ebaae0fd7f14 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Wed, 15 Jul 2026 15:14:57 +0200 Subject: [PATCH 2/7] feat: opt-in time-dependent stability classification Add a time_dependent config that cuts StabilityClassifier segments at equal-duration boundaries of the observed time span instead of equal-count boundaries of the change series. - StabilityClassifier.is_stable gains optional timestamps; boundaries via np.searchsorted on equal-duration cuts, count-mode fallback on missing/mismatched/non-finite timestamps or zero span (never raises) - SingleStabilityTracker(time_dependent=...) stores exact per-occurrence timestamps; state round-trips via msgpack, legacy snapshots load - timestamp threaded EventPersistency.ingest_event -> add_data -> add_value as optional trailing param; dataframe backends accept+ignore - EventStabilityTracker(time_dependent=...) forwards via closure factory (survives load(), mirroring expand_value) Default off is bit-for-bit unchanged. 17 new tests; full suite 550 pass. Co-Authored-By: Claude Fable 5 --- .../persistency/event_data_structures/base.py | 2 +- .../dataframes/chunked_event_dataframe.py | 2 +- .../dataframes/event_dataframe.py | 2 +- .../trackers/base/event_tracker.py | 4 +- .../trackers/base/multi_tracker.py | 4 +- .../trackers/base/single_tracker.py | 2 +- .../stability/stability_classifier.py | 63 +++++- .../trackers/stability/stability_tracker.py | 32 ++- .../utils/persistency/event_persistency.py | 5 +- .../test_time_dependent_stability.py | 182 ++++++++++++++++++ 10 files changed, 275 insertions(+), 23 deletions(-) create mode 100644 tests/test_persistency/test_time_dependent_stability.py diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py index 08e4909b..31eb768a 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py @@ -11,7 +11,7 @@ class EventDataStructure(ABC): template: str = "" @abstractmethod - def add_data(self, data_object: Any) -> None: ... + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: ... @abstractmethod def get_data(self) -> Any: ... diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py index 53c32306..fac9c73c 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py @@ -24,7 +24,7 @@ class ChunkedEventDataFrame(EventDataStructure): chunks: list[pl.DataFrame] = field(default_factory=list) _rows: int = 0 - def add_data(self, data: pl.DataFrame) -> None: + def add_data(self, data: pl.DataFrame, timestamp: float | None = None) -> None: if data.height == 0: return self.chunks.append(data) diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py index 2a024883..5519393e 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py @@ -17,7 +17,7 @@ class EventDataFrame(EventDataStructure): """ data: pd.DataFrame = field(default_factory=pd.DataFrame) - def add_data(self, data: pd.DataFrame) -> None: + def add_data(self, data: pd.DataFrame, timestamp: float | None = None) -> None: if len(self.data) > 0: self.data = pd.concat([self.data, data], ignore_index=True) else: diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py index f43a22a7..05b1c6d7 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py @@ -27,9 +27,9 @@ def __init__( self.converter_function = converter_function self.multi_tracker = self.multi_tracker_type(single_tracker_type=self.single_tracker_type) - def add_data(self, data_object: Any) -> None: + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: """Add data to the variable trackers.""" - self.multi_tracker.add_data(data_object) + self.multi_tracker.add_data(data_object, timestamp=timestamp) def get_data(self) -> Dict[str, SingleTracker]: """Retrieve the tracker's stored data.""" diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py index 08df8218..19aa2d28 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py @@ -13,12 +13,12 @@ def __init__(self, single_tracker_type: Callable[[], SingleTracker] = SingleTrac self.single_trackers: Dict[str, SingleTracker] = {} self.single_tracker_type: Callable[[], SingleTracker] = single_tracker_type - def add_data(self, data_object: Dict[str, Any]) -> None: + def add_data(self, data_object: Dict[str, Any], timestamp: float | None = None) -> None: """Add data to the appropriate feature trackers.""" for name, value in data_object.items(): if name not in self.single_trackers: self.single_trackers[name] = self.single_tracker_type() - self.single_trackers[name].add_value(value) + self.single_trackers[name].add_value(value, timestamp=timestamp) def get_trackers(self) -> Dict[str, SingleTracker]: """Get the current feature trackers.""" diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py index 01808d89..c5c12a43 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py @@ -15,7 +15,7 @@ class SingleTracker(ABC): """Tracks whether a single variable is converging to a constant value.""" @abstractmethod - def add_value(self, value: Any) -> None: + def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" @abstractmethod diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index 25db8bc2..c756db9f 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -18,10 +18,51 @@ def __init__(self, segment_thresholds: List[float], min_samples: int = 10): # for lists self.segment_means: List[float] = [] - def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: + def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = None) -> List[int]: + """Index boundaries of n_segments segments over total_len items. + + Equal-count by default. When timestamps are given (one per item, + chronological, non-zero span), boundaries are equal-DURATION cuts of + the observed time span, mapped back to indices. Falls back to + equal-count on missing/mismatched/non-finite timestamps or zero span. + """ + try: + use_time = ( + timestamps is not None + and len(timestamps) == total_len + and total_len > 0 + and bool(np.all(np.isfinite(timestamps))) + and timestamps[-1] > timestamps[0] + ) + except TypeError: + # e.g. a None entry: not comparable/convertible -> equal-count + use_time = False + if use_time: + t_first, t_last = timestamps[0], timestamps[-1] + cuts = [ + t_first + k * (t_last - t_first) / self.n_segments + for k in range(self.n_segments + 1) + ] + boundaries = [int(np.searchsorted(timestamps, t, side="left")) for t in cuts] + boundaries[0] = 0 + boundaries[-1] = total_len + return boundaries + segment_size = total_len / self.n_segments + boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] + boundaries[-1] = total_len + return boundaries + + def is_stable( + self, + change_series: RLEList[bool] | List[bool], + timestamps: List[float] | None = None, + ) -> bool: """Determine if a list of segment means is stable. Works efficiently with RLEList without expanding to a full list. + When timestamps are given (one per occurrence, chronological), + segments are equal-duration cuts of the time span instead of + equal-count cuts of the series. """ # Handle both RLEList and regular list if isinstance(change_series, RLEList): @@ -29,10 +70,7 @@ def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: if total_len == 0: return True - # Calculate segment boundaries - segment_size = total_len / self.n_segments - segment_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] - segment_boundaries[-1] = total_len + segment_boundaries = self._segment_boundaries(total_len, timestamps) # Compute segment means directly from RLE runs segment_sums = [0.0] * self.n_segments @@ -65,8 +103,15 @@ def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: for i in range(self.n_segments) ] else: - # Original implementation for regular lists - self.segment_means = self._compute_segment_means(change_series) + if timestamps is not None and len(timestamps) == len(change_series): + b = self._segment_boundaries(len(change_series), timestamps) + self.segment_means = [ + float(np.mean(change_series[b[i]:b[i + 1]])) if b[i + 1] > b[i] else np.nan + for i in range(self.n_segments) + ] + else: + # Original implementation for regular lists + self.segment_means = self._compute_segment_means(change_series) return all([not q >= thresh for q, thresh in zip(self.segment_means, self.segment_threshs)]) def _compute_segment_means(self, change_series: List[bool]) -> List[float]: @@ -80,8 +125,8 @@ def get_last_segment_means(self) -> List[float]: def get_segment_thresholds(self) -> List[float]: return self.segment_threshs - def __call__(self, change_series: RLEList[bool] | List[bool]) -> bool: - return self.is_stable(change_series) + def __call__(self, change_series: RLEList[bool] | List[bool], timestamps: List[float] | None = None) -> bool: + return self.is_stable(change_series, timestamps=timestamps) def __repr__(self) -> str: return ( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index 8184cc0a..fbfb3625 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -12,27 +12,43 @@ class SingleStabilityTracker(SingleTracker): """Tracks stability of a single feature.""" - def __init__(self, min_samples: int = 3, expand_value: bool = False) -> None: + def __init__( + self, + min_samples: int = 3, + expand_value: bool = False, + time_dependent: bool = False, + ) -> None: self.min_samples = min_samples self.expand_value = expand_value + self.time_dependent = time_dependent self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( segment_thresholds=[1.1, 0.3, 0.1, 0.01], ) self._accum = set.update if expand_value else set.add + # ponytail: O(N) timestamps; switch to fixed-width time buckets if + # this ever runs unbounded/streaming. + self.timestamps: List[float] = [] # Opaque slot for detectors to stash per-variable model state that # must survive save/load. Schema-free; the tracker does not interpret it. self.extra_state: Dict[str, Any] = {} - def add_value(self, value: Any) -> None: + def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" before = len(self.unique_set) self._accum(self.unique_set, value) self.change_series.append(len(self.unique_set) > before) + if self.time_dependent and timestamp is not None: + self.timestamps.append(float(timestamp)) def classify(self) -> Classification: """Classify the variable.""" + timestamps = ( + self.timestamps + if self.time_dependent and len(self.timestamps) == len(self.change_series) + else None + ) if len(self.change_series) < self.min_samples: return Classification( type="INSUFFICIENT_DATA", @@ -48,7 +64,7 @@ def classify(self) -> Classification: type="RANDOM", reason=f"Unique set size equals number of samples ({len(self.change_series)})" ) - elif self.stability_classifier.is_stable(self.change_series): + elif self.stability_classifier.is_stable(self.change_series, timestamps=timestamps): return Classification( type="STABLE", reason=( @@ -70,6 +86,8 @@ def to_state(self) -> Dict[str, Any]: "module": self.__class__.__module__, "min_samples": self.min_samples, "expand_value": self.expand_value, + "time_dependent": self.time_dependent, + "timestamps": self.timestamps, "runs": self.change_series.runs(), "unique_set": list(self.unique_set), "segment_thresholds": self.stability_classifier.segment_threshs, @@ -82,6 +100,7 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": tracker = cls( min_samples=state["min_samples"], expand_value=state.get("expand_value", False), + time_dependent=state.get("time_dependent", False), ) runs = [(bool(r[0]), int(r[1])) for r in state["runs"]] tracker.change_series._runs = runs @@ -92,6 +111,7 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": tracker.stability_classifier = StabilityClassifier( segment_thresholds=state["segment_thresholds"] ) + tracker.timestamps = [float(t) for t in state.get("timestamps", [])] tracker.extra_state = state.get("extra_state", {}) return tracker @@ -132,11 +152,15 @@ def __init__( self, converter_function: Callable[[Any], Any] = lambda x: x, expand_value: bool = False, + time_dependent: bool = False, ) -> None: self.multi_tracker: MultiStabilityTracker # for type hinting def make_tracker() -> SingleStabilityTracker: - return SingleStabilityTracker(expand_value=expand_value) + return SingleStabilityTracker( + expand_value=expand_value, + time_dependent=time_dependent, + ) # Mirror class identity onto the closure so dump()/load() can resolve # the underlying SingleStabilityTracker via its module + qualname. diff --git a/src/detectmatelibrary/utils/persistency/event_persistency.py b/src/detectmatelibrary/utils/persistency/event_persistency.py index ba9ae18c..8dd22d39 100644 --- a/src/detectmatelibrary/utils/persistency/event_persistency.py +++ b/src/detectmatelibrary/utils/persistency/event_persistency.py @@ -44,7 +44,8 @@ def ingest_event( event_id: int | str, event_template: str, variables: list[Any] = [], - named_variables: Dict[str, Any] = {} + named_variables: Dict[str, Any] = {}, + timestamp: float | None = None, ) -> None: """Ingest event data into the appropriate EventData store.""" with self._lock: @@ -60,7 +61,7 @@ def ingest_event( self.events_data[event_id] = data_structure data = data_structure.to_data(all_variables) - data_structure.add_data(data) + data_structure.add_data(data, timestamp=timestamp) # ponytail: fire callbacks outside the lock so a count-triggered save # doesn't hold the ingest lock across serialize + file I/O. for _cb in self._on_ingest_callbacks: diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py new file mode 100644 index 00000000..683e7098 --- /dev/null +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -0,0 +1,182 @@ +"""Tests for the time_dependent option of the stability trackers.""" + +from detectmatelibrary.utils.persistency.rle_list import RLEList +from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_classifier import ( + StabilityClassifier, +) + +THRESHOLDS = [1.1, 0.3, 0.1, 0.01] # same defaults SingleStabilityTracker uses + + +def make_classifier() -> StabilityClassifier: + return StabilityClassifier(segment_thresholds=THRESHOLDS) + + +# Divergence fixture: 3 changes up front, then a quiet tail of 37. +# In *time*, the changes span most of the observed window and the quiet +# tail is a burst compressed into ~0.04s at the end. +DIVERGENT_SERIES = [True, True, True] + [False] * 37 +DIVERGENT_TIMES = [0.0, 40.0, 80.0] + [100.0 + 0.001 * i for i in range(37)] + + +class TestClassifierTimeBoundaries: + def test_count_mode_is_stable_on_divergent_fixture(self): + clf = make_classifier() + # count segments of 10: means [0.3, 0, 0, 0] -> all below thresholds + assert clf.is_stable(RLEList(DIVERGENT_SERIES)) is True + + def test_time_mode_is_unstable_on_divergent_fixture(self): + clf = make_classifier() + # time quarters put a lone change (mean 1.0) into segment 2 (thresh 0.3) + assert clf.is_stable(RLEList(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_uniform_timestamps_match_count_mode(self): + # N divisible by n_segments -> boundaries coincide exactly + series = [True, False, False, True, False, False, False, False] + ts = [float(i) for i in range(8)] + clf_count = make_classifier() + clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + clf_time.is_stable(RLEList(series), timestamps=ts) + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_plain_list_path_supports_timestamps(self): + clf = make_classifier() + assert clf.is_stable(list(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_zero_span_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + result = clf_time.is_stable(RLEList(series), timestamps=[5.0] * 8) + assert result == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_length_mismatch_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(series), timestamps=[1.0, 2.0]) == expected + + def test_none_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, None, 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_nan_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, float("nan"), 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + +from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_tracker import ( + SingleStabilityTracker, +) + + +def feed_divergent(tracker: SingleStabilityTracker) -> None: + """3 new values spread over ~80s, then a repeated value bursting at t~100.""" + values = ["a", "b", "c"] + ["c"] * 37 + for value, ts in zip(values, DIVERGENT_TIMES): + tracker.add_value(value, timestamp=ts) + + +class TestSingleStabilityTrackerTimeDependent: + def test_timestamps_stored_only_when_enabled(self): + on = SingleStabilityTracker(time_dependent=True) + on.add_value("a", timestamp=1.0) + on.add_value("b", timestamp=2.0) + assert on.timestamps == [1.0, 2.0] + + off = SingleStabilityTracker() # default time_dependent=False + off.add_value("a", timestamp=1.0) + assert off.timestamps == [] + + def test_classification_diverges_between_modes(self): + count_mode = SingleStabilityTracker() + feed_divergent(count_mode) + assert count_mode.classify().type == "STABLE" + + time_mode = SingleStabilityTracker(time_dependent=True) + feed_divergent(time_mode) + assert time_mode.classify().type == "UNSTABLE" + + def test_missing_timestamps_fall_back_to_count_mode(self): + # time_dependent on, but values arrive without timestamps + tracker = SingleStabilityTracker(time_dependent=True) + for value in ["a", "b", "c"] + ["c"] * 37: + tracker.add_value(value) + reference = SingleStabilityTracker() + for value in ["a", "b", "c"] + ["c"] * 37: + reference.add_value(value) + assert tracker.classify().type == reference.classify().type + + def test_round_trip_preserves_time_state(self): + tracker = SingleStabilityTracker(time_dependent=True) + feed_divergent(tracker) + restored = SingleStabilityTracker.from_state(tracker.to_state()) + assert restored.time_dependent is True + assert restored.timestamps == tracker.timestamps + assert restored.classify().type == "UNSTABLE" + + def test_legacy_state_without_time_keys_defaults_off(self): + tracker = SingleStabilityTracker() + tracker.add_value("hello") + state = tracker.to_state() + state.pop("time_dependent", None) # simulate pre-flag snapshot + state.pop("timestamps", None) + restored = SingleStabilityTracker.from_state(state) + assert restored.time_dependent is False + assert restored.timestamps == [] + + +from detectmatelibrary.utils.persistency import EventPersistency +from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_tracker import ( + EventStabilityTracker, +) + + +class TestTimeDependentPlumbing: + def test_event_tracker_propagates_flag_and_timestamp(self): + event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + single = event_tracker.get_data()["var1"] + assert single.time_dependent is True + assert single.timestamps == [1.0, 2.0] + + def test_ingest_event_forwards_timestamp(self): + storage = EventPersistency( + EventStabilityTracker, + event_data_kwargs={"time_dependent": True}, + ) + storage.ingest_event(1, "tpl <*>", variables=["a"], timestamp=10.0) + storage.ingest_event(1, "tpl <*>", variables=["b"], timestamp=20.0) + single = storage.get_events_data()[1].get_data()["var_0"] + assert single.timestamps == [10.0, 20.0] + + def test_ingest_event_without_timestamp_still_works(self): + storage = EventPersistency(EventStabilityTracker) + storage.ingest_event(1, "tpl <*>", variables=["a"]) + single = storage.get_events_data()[1].get_data()["var_0"] + assert list(single.change_series) == [True] + assert single.timestamps == [] + + def test_event_tracker_dump_load_preserves_timestamps(self): + event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + restored = EventStabilityTracker.load(event_tracker.dump(), time_dependent=True) + single = restored.get_data()["var1"] + assert single.time_dependent is True + assert single.timestamps == [1.0, 2.0] From 7d4d3960decab6e234aa06c2a68337d0232feca9 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Thu, 23 Jul 2026 11:20:44 +0200 Subject: [PATCH 3/7] Fix botched development merge in stability trackers Merge left both sides' __init__ stacked instead of combined and a duplicate return in make_tracker. Merge the params (expand_value/ time_dependent + add_value_fn/detector_config) into one __init__, use self._accum in add_value so expand_value takes effect, and drop the dead return. Narrow timestamps for mypy without assert and consolidate test imports via the trackers re-export. Co-Authored-By: Claude Opus 4.8 --- .../stability/stability_classifier.py | 13 ++++++++----- .../trackers/stability/stability_tracker.py | 10 +++++----- .../test_time_dependent_stability.py | 19 ++++++------------- 3 files changed, 19 insertions(+), 23 deletions(-) diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index c756db9f..bf907758 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -22,9 +22,10 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N """Index boundaries of n_segments segments over total_len items. Equal-count by default. When timestamps are given (one per item, - chronological, non-zero span), boundaries are equal-DURATION cuts of - the observed time span, mapped back to indices. Falls back to - equal-count on missing/mismatched/non-finite timestamps or zero span. + chronological, non-zero span), boundaries are equal-DURATION + cuts of the observed time span, mapped back to indices. Falls + back to equal-count on missing/mismatched/non-finite timestamps + or zero span. """ try: use_time = ( @@ -37,7 +38,7 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N except TypeError: # e.g. a None entry: not comparable/convertible -> equal-count use_time = False - if use_time: + if use_time and timestamps is not None: # 2nd clause narrows for mypy t_first, t_last = timestamps[0], timestamps[-1] cuts = [ t_first + k * (t_last - t_first) / self.n_segments @@ -125,7 +126,9 @@ def get_last_segment_means(self) -> List[float]: def get_segment_thresholds(self) -> List[float]: return self.segment_threshs - def __call__(self, change_series: RLEList[bool] | List[bool], timestamps: List[float] | None = None) -> bool: + def __call__( + self, change_series: RLEList[bool] | List[bool], timestamps: List[float] | None = None + ) -> bool: return self.is_stable(change_series, timestamps=timestamps) def __repr__(self) -> str: diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index cd103e6a..9ad5dbd5 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -19,13 +19,12 @@ def __init__( min_samples: int = 3, expand_value: bool = False, time_dependent: bool = False, + add_value_fn: str = "default", + detector_config: "CoreDetectorConfig | None" = None, ) -> None: self.min_samples = min_samples self.expand_value = expand_value self.time_dependent = time_dependent - def __init__(self, min_samples: int = 3, add_value_fn: str = "default", - detector_config: "CoreDetectorConfig | None" = None) -> None: - self.min_samples = min_samples self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( @@ -51,7 +50,7 @@ def __init__(self, min_samples: int = 3, add_value_fn: str = "default", def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" before = len(self.unique_set) - self.unique_set.add(value) + self._accum(self.unique_set, value) self.change_series.append(len(self.unique_set) > before) if self.time_dependent and timestamp is not None: self.timestamps.append(float(timestamp)) @@ -181,8 +180,9 @@ def make_tracker() -> SingleStabilityTracker: return SingleStabilityTracker( expand_value=expand_value, time_dependent=time_dependent, + add_value_fn=add_value_fn, + detector_config=detector_config, ) - return SingleStabilityTracker(add_value_fn=add_value_fn, detector_config=detector_config) # Mirror class identity onto the closure so dump()/load() can resolve # the underlying SingleStabilityTracker via its module + qualname. diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py index 683e7098..c497fa80 100644 --- a/tests/test_persistency/test_time_dependent_stability.py +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -1,8 +1,11 @@ """Tests for the time_dependent option of the stability trackers.""" from detectmatelibrary.utils.persistency.rle_list import RLEList -from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_classifier import ( +from detectmatelibrary.utils.persistency import EventPersistency +from detectmatelibrary.utils.persistency.event_data_structures.trackers import ( StabilityClassifier, + SingleStabilityTracker, + EventStabilityTracker, ) THRESHOLDS = [1.1, 0.3, 0.1, 0.01] # same defaults SingleStabilityTracker uses @@ -79,13 +82,9 @@ def test_nan_timestamp_entry_falls_back_to_count_mode(self): assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() -from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_tracker import ( - SingleStabilityTracker, -) - - def feed_divergent(tracker: SingleStabilityTracker) -> None: - """3 new values spread over ~80s, then a repeated value bursting at t~100.""" + """3 new values spread over ~80s, then a repeated value bursting at + t~100.""" values = ["a", "b", "c"] + ["c"] * 37 for value, ts in zip(values, DIVERGENT_TIMES): tracker.add_value(value, timestamp=ts) @@ -140,12 +139,6 @@ def test_legacy_state_without_time_keys_defaults_off(self): assert restored.timestamps == [] -from detectmatelibrary.utils.persistency import EventPersistency -from detectmatelibrary.utils.persistency.event_data_structures.trackers.stability.stability_tracker import ( - EventStabilityTracker, -) - - class TestTimeDependentPlumbing: def test_event_tracker_propagates_flag_and_timestamp(self): event_tracker = EventStabilityTracker(time_dependent=True) From 7345ef5e73e0837fe3f06e50d24c5c353a213e51 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 4 Aug 2026 16:20:10 +0200 Subject: [PATCH 4/7] revert: restore AGENTS.md Code Quality section Unrelated to the time-dependent feature; the deleted section documents the mypy/flake8/bandit/vulture gates and the Python 3.12 requirement. --- AGENTS.md | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 815bd2bb..4b089d4b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,10 @@ uv sync --dev uv run prek install # Run tests -uv run pytest +uv run pytest -q +uv run pytest -s # verbose with stdout +uv run pytest --cov=. --cov-report=term-missing # with coverage +uv run pytest tests/test_foo.py # single test file # Run linting/formatting (all pre-commit hooks) uv run prek run -a @@ -179,6 +182,17 @@ def set_configuration(self) -> None: Omitting either step means a `persist:` block in the YAML is silently ignored with no error. +## Code Quality + +Pre-commit hooks enforce: +- **mypy** strict mode +- **flake8** linting, **autopep8** formatting (max line 110) +- **bandit** security checks, **vulture** dead-code detection (70% threshold) +- **docformatter** docstring style + +Python 3.12 is required (see `.python-version`). + + # Git NEVER include "Co-Authored-By ..." in your commit or PR messages. From bc112045bcfbd32841f9af50ea64df110ce3de87 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 4 Aug 2026 16:20:22 +0200 Subject: [PATCH 5/7] feat: time-dependent stability classification, reconciled with add_value_fn Segments a variable's change series by equal time spans instead of equal sample counts when timestamps are available, so a variable that churns in a burst and then settles is judged on when its values changed rather than on how many samples separated them. Reconciles the feature with development's add_value_fn hook, which a merge had collided with. A detector no longer replaces the tracker's add_value; the tracker calls it through _value_fn and keeps ownership of the two lists that must stay aligned. A detector that records nothing for a value (ValueRangeDetector on non-numeric input) therefore cannot desync timestamps from change_series. That collision was failing 35 tests. Drops expand_value, which 895a0d4 replaced with add_value_fn and a botched merge resurrected: nothing passed it, CharsetDetector already provides its semantics, and it was silently inert whenever add_value_fn was set. Config surface: time_dependent, timestamp_variable and timestamp_format on VariableDetectorConfig. The event time is read from logFormatVariables and parsed by the existing TimeFormatHandler; a missing or unparseable value warns once per detector and falls back to count segmentation, so a typo'd variable name is not mistaken for a working time-dependent run. The flag is merged into the tracker kwargs in VariableDetector.__init__ rather than in _stability_kwargs, so it also reaches NewValueDetector and NewValueComboDetector, and is preserved across the set_configuration reassignment that auto_config performs before training. StabilityClassifier now has one segment-boundary rule for both the RLE and plain-list paths; _compute_segment_means used np.array_split (remainder front-loaded) while _segment_boundaries floored. Equal-duration cuts fall back to count boundaries when they would leave a segment empty or when timestamps are not monotonic, since an empty segment yields a nan mean that passes every threshold unconditionally. Downstream: SingleTracker.add_value and EventDataStructure.add_data gain an optional trailing timestamp: float | None = None, so out-of-tree subclasses of those ABCs must accept it. Detector add_value(self, tracker, value) signatures are unaffected. --- docs/detectors.md | 59 ++++ .../common/variable_detector.py | 70 +++- .../detectors/new_value_combo_detector.py | 25 +- .../trackers/base/event_tracker.py | 2 +- .../stability/stability_classifier.py | 68 ++-- .../trackers/stability/stability_tracker.py | 53 +-- .../test_new_value_combo_detector.py | 103 +++++- .../test_time_dependent_stability.py | 328 +++++++++++++++++- 8 files changed, 651 insertions(+), 57 deletions(-) diff --git a/docs/detectors.md b/docs/detectors.md index e3e744a3..b787342b 100644 --- a/docs/detectors.md +++ b/docs/detectors.md @@ -222,6 +222,65 @@ def set_configuration(self): When `auto_config` is `False`, steps 1 and 2 are skipped entirely. +### Time-dependent stability (optional) + +Stability classification splits a variable's change history into four segments and +compares each segment's rate of change against a threshold. By default the segments +are **equal-count**: each holds the same number of observations, regardless of how much +time they cover. For bursty log sources that is misleading — a variable that changed +constantly during a quiet night and then went silent under a flood of daytime traffic +looks stable, because the flood supplies enough samples to dominate the later segments. + +Setting `time_dependent: true` switches the segmentation to **equal-duration** cuts of +the observed time span, so each segment covers the same amount of wall-clock time. The +detector then needs an event time per record, which it reads from the log's named +variables (`logFormatVariables`, i.e. the fields declared in the parser's `log_format`) +under the name given by `timestamp_variable`. + +These three parameters live on every `VariableDetector` subclass (`NewValueDetector`, +`NewValueComboDetector`, `ValueRangeDetector`, `CharsetDetector`, `BigramDetector`, …) +and go in the detector's top-level `params` block: + +```yaml +detectors: + NewValueDetector: + method_type: new_value_detector + auto_config: True + params: + time_dependent: True + timestamp_variable: Time # a field name from the parser's log_format + timestamp_format: "%y%m%d %H%M%S" # optional; omit to auto-detect +``` + +#### Fields + +| Field | Type | Default | Description | +|---|---|---|---| +| `time_dependent` | `bool` | `false` | Segment the change history by equal time spans instead of equal sample counts. When `false` (the default) the other two fields are ignored and no timestamps are recorded. | +| `timestamp_variable` | `str \| null` | `null` | Name of the field in `logFormatVariables` holding the record's event time. Required for `time_dependent` to have any effect. Only named log-format fields are consulted — never the positional `variables` list. | +| `timestamp_format` | `str \| null` | `null` | Explicit [`strftime`](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes) pattern for parsing that field. When unset, `TimeFormatHandler` auto-detects the format (ISO 8601, Apache, syslog, numeric epoch seconds/milliseconds, and other common layouts). | + +Set `timestamp_format` when the source uses a layout the auto-detection does not +know. The HDFS loghub corpus, for example, stamps records as `081109 203615`, which +only parses with an explicit `"%y%m%d %H%M%S"`. + +#### Fallback behaviour + +Time-dependent segmentation is best-effort and never fails a run: + +* If `time_dependent` is `true` but `timestamp_variable` is unset, or the named field + is absent from a record, or its value cannot be parsed, the detector logs a + **single** warning (once per detector, so a bad config cannot flood the log) and + falls back to count-based segmentation. +* If timestamps stop lining up with the recorded observations, or the observed time + span is zero, or the equal-duration cut would leave a segment with no observations + in it, the classifier silently falls back to count-based segmentation for that + variable. + +In every fallback case classification still runs and produces a result — only the +segmentation rule changes back to the default. + + ### Saving state (persist) Detectors can persist their training state to disk (or cloud storage) so it diff --git a/src/detectmatelibrary/common/variable_detector.py b/src/detectmatelibrary/common/variable_detector.py index 8b57395a..bc012f0b 100644 --- a/src/detectmatelibrary/common/variable_detector.py +++ b/src/detectmatelibrary/common/variable_detector.py @@ -13,6 +13,7 @@ ) from detectmatelibrary.utils.persistency.event_persistency import EventPersistency from detectmatelibrary.utils.data_buffer import BufferMode +from detectmatelibrary.utils.time_format_handler import TimeFormatHandler from detectmatelibrary.schemas import ParserSchema, DetectorSchema from detectmatelibrary.constants import GLOBAL_EVENT_ID from detectmatelibrary.tools.logging import logger @@ -25,6 +26,13 @@ class VariableDetectorConfig(CoreDetectorConfig): use_stable_vars: bool = True use_static_vars: bool = True + # Time-dependent stability: cut the classifier's segments at equal-duration + # boundaries instead of equal-count ones. Needs a per-record event time, + # named here and read from the record's logFormatVariables. + time_dependent: bool = False + timestamp_variable: str | None = None + timestamp_format: str | None = None # None -> TimeFormatHandler auto-detect + class VariableDetector(CoreDetector): """Abstract base for detectors that learn a per-variable model from @@ -44,17 +52,31 @@ class VariableDetector(CoreDetector): def __init__(self, name: str, config: VariableDetectorConfig) -> None: super().__init__(name=name, buffer_mode=BufferMode.NO_BUF, config=config) self.config: VariableDetectorConfig # type narrowing for IDE + self._time_handler = TimeFormatHandler() + self._warned_bad_timestamp = False self.persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._event_data_kwargs(), + event_data_kwargs=self._with_time_flag(self._event_data_kwargs()), ) # auto config checks individual-variable stability to select features self.auto_conf_persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._auto_conf_kwargs(), + event_data_kwargs=self._with_time_flag(self._auto_conf_kwargs()), ) self._register_persistency(self.persistency) + def _with_time_flag(self, kwargs: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Add time_dependent to tracker kwargs when the config asks for it. + + Done here rather than in _stability_kwargs so every + VariableDetector subclass is covered -- NewValueDetector + overrides neither construction hook and NewValueComboDetector + returns only a converter_function. + """ + if not self.config.time_dependent: + return kwargs + return {**(kwargs or {}), "time_dependent": True} + # ---- construction hooks ------------------------------------------------- def _event_data_class(self) -> type: @@ -75,6 +97,42 @@ def _stability_kwargs(self) -> Dict[str, Any]: "detector_config": self.config.to_dict(method_id=name), } + def _warn_time_fallback_once(self, reason: str) -> None: + """Log the first time-dependent misconfiguration, then stay quiet. + + A bad config would otherwise emit one warning per record, so the + flag latches after the first message. + """ + if self._warned_bad_timestamp: + return + self._warned_bad_timestamp = True + logger.warning( + "%s: %s; falling back to count-based stability segmentation.", + self.name, reason, + ) + + def _timestamp(self, input_: ParserSchema) -> float | None: + """Resolve the record's event time, or None to use count + segmentation.""" + if not self.config.time_dependent: + return None + if not self.config.timestamp_variable: + # Enabling the feature without naming the field is an operator + # error, not an opt-out -- say so rather than silently no-op. + self._warn_time_fallback_once( + "time_dependent is enabled but timestamp_variable is not set" + ) + return None + raw = input_["logFormatVariables"].get(self.config.timestamp_variable) + ts = self._time_handler.parse_timestamp(str(raw or ""), self.config.timestamp_format) + if ts == "0": + self._warn_time_fallback_once( + f"timestamp_variable {self.config.timestamp_variable!r} is missing or " + f"unparseable (got {raw!r})" + ) + return None + return float(ts) + # ---- per-detector hooks ------------------------------------------------- def _prepare_variables(self, variables: Dict[str, Any], stage: str) -> Dict[str, Any]: @@ -112,6 +170,7 @@ def _ingest(self, input_: ParserSchema, variables: Dict[str, Any], event_id: Any event_id=event_id, event_template=input_["template"], named_variables=variables, + timestamp=self._timestamp(input_), ) def detect(self, input_: ParserSchema, output_: DetectorSchema) -> bool: # type: ignore @@ -175,6 +234,7 @@ def configure(self, input_: ParserSchema) -> None: # type: ignore event_template=input_["template"], variables=input_["variables"], named_variables=input_["logFormatVariables"], + timestamp=self._timestamp(input_), ) @override @@ -200,6 +260,9 @@ def set_configuration(self) -> None: if selected: variables[event_id] = selected old_persist = self.config.persist + old_time_dependent = self.config.time_dependent + old_timestamp_variable = self.config.timestamp_variable + old_timestamp_format = self.config.timestamp_format config_dict = generate_detector_config( variable_selection=variables, detector_name=self.name, @@ -207,6 +270,9 @@ def set_configuration(self) -> None: ) self.config = type(self.config).from_dict(config_dict, self.name) self.config.persist = old_persist + self.config.time_dependent = old_time_dependent + self.config.timestamp_variable = old_timestamp_variable + self.config.timestamp_format = old_timestamp_format events = self.config.events if isinstance(events, EventsConfig) and not events.events: logger.warning( diff --git a/src/detectmatelibrary/detectors/new_value_combo_detector.py b/src/detectmatelibrary/detectors/new_value_combo_detector.py index e166250b..31f223b3 100644 --- a/src/detectmatelibrary/detectors/new_value_combo_detector.py +++ b/src/detectmatelibrary/detectors/new_value_combo_detector.py @@ -59,7 +59,9 @@ def __init__( # second-pass persistency to learn stability of variable combinations self.auto_conf_persistency_combos = persistency.EventPersistency( event_data_class=persistency.EventStabilityTracker, - event_data_kwargs={"converter_function": get_all_possible_combos}, + event_data_kwargs=self._with_time_flag( + {"converter_function": get_all_possible_combos} + ), ) self.inputs: list[ParserSchema] = [] @@ -101,6 +103,24 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: every possible combo up front would explode combinatorially). """ old_persist = self.config.persist + time_fields = { + "time_dependent": self.config.time_dependent, + "timestamp_variable": self.config.timestamp_variable, + "timestamp_format": self.config.timestamp_format, + } + + def restore_time_fields() -> None: + """Carry the time settings across a config reassignment. + + generate_detector_config only emits method_type / auto_config / + params / events, so every ``from_dict`` below resets these to their + defaults. The re-ingest loop calls ``_timestamp()`` under the pass-1 + config, so restoring only at the end would leave the combo trackers + timestamp-less. + """ + for field, value in time_fields.items(): + setattr(self.config, field, value) + # pass 1: stable individual variables -> combos variable_combos = {} for event_id, tracker in self.auto_conf_persistency.get_events_data().items(): @@ -114,6 +134,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: max_combo_size=max_combo_size or self.config.max_combo_size, ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) + restore_time_fields() # re-ingest all inputs to learn combos under the new configuration for input_ in self.inputs: @@ -122,6 +143,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: event_id=input_["EventID"], event_template=input_["template"], named_variables=configured_variables, + timestamp=self._timestamp(input_), ) # pass 2: stable/static combos -> final config @@ -148,6 +170,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) self.config.persist = old_persist + restore_time_fields() events = self.config.events if isinstance(events, EventsConfig) and not events.events: logger.warning( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py index 05b1c6d7..98067edd 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py @@ -72,7 +72,7 @@ def load(cls, data: bytes, **kwargs: Any) -> "EventTracker": ``multi_tracker_type`` recorded in the snapshot. For any subclass, ``cls(**kwargs)`` is called instead, which lets subclasses with closure-based factories (e.g. ``EventStabilityTracker``'s - ``expand_value``) rebuild their factory so it survives load. + ``time_dependent``) rebuild their factory so it survives load. Contract for subclasses: ``__init__`` must accept the kwargs forwarded to ``load()`` and must not require additional positional arguments. diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index bf907758..e4a0d153 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -22,17 +22,26 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N """Index boundaries of n_segments segments over total_len items. Equal-count by default. When timestamps are given (one per item, - chronological, non-zero span), boundaries are equal-DURATION + non-decreasing, non-zero span), boundaries are equal-DURATION cuts of the observed time span, mapped back to indices. Falls - back to equal-count on missing/mismatched/non-finite timestamps - or zero span. + back to equal-count on missing / mismatched / non-finite / out- + of-order timestamps, on zero span, and when the duration cut + would leave a segment empty. """ + segment_size = total_len / self.n_segments + count_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] + count_boundaries[-1] = total_len try: use_time = ( timestamps is not None and len(timestamps) == total_len and total_len > 0 and bool(np.all(np.isfinite(timestamps))) + # np.searchsorted below requires sorted input. Merged sources or + # concurrent writers can deliver stamps out of order, which would + # silently produce wrong boundaries rather than an error. O(N), + # same cost as the isfinite scan above. + and bool(np.all(np.diff(timestamps) >= 0)) and timestamps[-1] > timestamps[0] ) except TypeError: @@ -47,11 +56,18 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N boundaries = [int(np.searchsorted(timestamps, t, side="left")) for t in cuts] boundaries[0] = 0 boundaries[-1] = total_len - return boundaries - segment_size = total_len / self.n_segments - boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] - boundaries[-1] = total_len - return boundaries + # An empty segment gets a nan mean, and `not nan >= thresh` is True -- + # a free pass. Under equal-duration cuts a bursty series can leave + # several segments empty and be called STABLE while churning; a + # 1-sample segment's quantized 0/1 mean can misfire the other way. + # Equal-count segmentation keeps every segment populated, so fall + # back to it whenever the duration cut would not. + # ponytail: this drops such series back to count mode wholesale; a + # future upgrade could redistribute or merge the empty segments and + # keep a (coarser) time-aware verdict. + if all(boundaries[i + 1] > boundaries[i] for i in range(self.n_segments)): + return boundaries + return count_boundaries def is_stable( self, @@ -61,18 +77,18 @@ def is_stable( """Determine if a list of segment means is stable. Works efficiently with RLEList without expanding to a full list. - When timestamps are given (one per occurrence, chronological), + When timestamps are given (one per occurrence, non-decreasing), segments are equal-duration cuts of the time span instead of - equal-count cuts of the series. + equal-count cuts of the series. See ``_segment_boundaries`` for + the conditions under which time mode falls back to count mode. """ - # Handle both RLEList and regular list - if isinstance(change_series, RLEList): - total_len = len(change_series) - if total_len == 0: - return True + total_len = len(change_series) + if total_len == 0: + return True - segment_boundaries = self._segment_boundaries(total_len, timestamps) + segment_boundaries = self._segment_boundaries(total_len, timestamps) + if isinstance(change_series, RLEList): # Compute segment means directly from RLE runs segment_sums = [0.0] * self.n_segments segment_counts = [0] * self.n_segments @@ -98,28 +114,18 @@ def is_stable( position = run_end - # Calculate means self.segment_means = [ segment_sums[i] / segment_counts[i] if segment_counts[i] > 0 else np.nan for i in range(self.n_segments) ] else: - if timestamps is not None and len(timestamps) == len(change_series): - b = self._segment_boundaries(len(change_series), timestamps) - self.segment_means = [ - float(np.mean(change_series[b[i]:b[i + 1]])) if b[i + 1] > b[i] else np.nan - for i in range(self.n_segments) - ] - else: - # Original implementation for regular lists - self.segment_means = self._compute_segment_means(change_series) + self.segment_means = [ + float(np.mean(change_series[segment_boundaries[i]:segment_boundaries[i + 1]])) + if segment_boundaries[i + 1] > segment_boundaries[i] else np.nan + for i in range(self.n_segments) + ] return all([not q >= thresh for q, thresh in zip(self.segment_means, self.segment_threshs)]) - def _compute_segment_means(self, change_series: List[bool]) -> List[float]: - """Get means of each segment for a normal list.""" - segments = np.array_split(change_series, self.n_segments) - return list(map(lambda x: np.mean(x) if len(x) > 0 else np.nan, segments)) - def get_last_segment_means(self) -> List[float]: return self.segment_means diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index be6bcfd9..575f90ba 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -46,20 +46,17 @@ class SingleStabilityTracker(SingleTracker): def __init__( self, min_samples: int = 3, - expand_value: bool = False, time_dependent: bool = False, add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None, ) -> None: self.min_samples = min_samples - self.expand_value = expand_value self.time_dependent = time_dependent self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( segment_thresholds=[1.1, 0.3, 0.1, 0.01], ) - self._accum = set.update if expand_value else set.add # ponytail: O(N) timestamps; switch to fixed-width time buckets if # this ever runs unbounded/streaming. self.timestamps: List[float] = [] @@ -68,29 +65,37 @@ def __init__( self.extra_state: Dict[str, Any] = {} self.add_value_fn = add_value_fn self.detector_config = detector_config + self._value_fn: Callable[[Any], None] = self._default_add_value if add_value_fn != "default": detector_cls = getattr(importlib.import_module("detectmatelibrary.detectors"), add_value_fn) if detector_config is not None: detector = detector_cls(config=_strip_persist(detector_config, add_value_fn)) else: detector = detector_cls() - self.add_value = partial(detector.add_value, self) # type: ignore[method-assign] + self._value_fn = partial(detector.add_value, self) - def add_value(self, value: Any, timestamp: float | None = None) -> None: - """Add a new value to the tracker.""" + def _default_add_value(self, value: Any) -> None: + """Default value semantics: one set entry per whole value.""" before = len(self.unique_set) - self._accum(self.unique_set, value) + self.unique_set.add(value) self.change_series.append(len(self.unique_set) > before) - if self.time_dependent and timestamp is not None: + + def add_value(self, value: Any, timestamp: float | None = None) -> None: + """Add a new value to the tracker. + + Value semantics belong to ``_value_fn`` -- either the default above or a + detector's ``add_value``. Timestamp bookkeeping stays here so + ``timestamps`` cannot drift from ``change_series``: a detector may record + nothing for a value (ValueRangeDetector on non-numeric input), and a + length mismatch silently demotes the variable to count segmentation. + """ + before = len(self.change_series) + self._value_fn(value) + if self.time_dependent and timestamp is not None and len(self.change_series) > before: self.timestamps.append(float(timestamp)) def classify(self) -> Classification: """Classify the variable.""" - timestamps = ( - self.timestamps - if self.time_dependent and len(self.timestamps) == len(self.change_series) - else None - ) if len(self.change_series) < self.min_samples: return Classification( type="INSUFFICIENT_DATA", @@ -106,7 +111,9 @@ def classify(self) -> Classification: type="RANDOM", reason=f"Unique set size equals number of samples ({len(self.change_series)})" ) - elif self.stability_classifier.is_stable(self.change_series, timestamps=timestamps): + elif self.stability_classifier.is_stable( + self.change_series, timestamps=self._aligned_timestamps() + ): return Classification( type="STABLE", reason=( @@ -120,6 +127,13 @@ def classify(self) -> Classification: reason="No classification matched; variable is unstable" ) + def _aligned_timestamps(self) -> List[float] | None: + """Timestamps to classify with, or None to fall back to count + segments.""" + if self.time_dependent and len(self.timestamps) == len(self.change_series): + return self.timestamps + return None + def to_state(self) -> Dict[str, Any]: """Serialize tracker state to a plain dict (must be msgpack- compatible).""" @@ -127,7 +141,6 @@ def to_state(self) -> Dict[str, Any]: "type": self.__class__.__name__, "module": self.__class__.__module__, "min_samples": self.min_samples, - "expand_value": self.expand_value, "time_dependent": self.time_dependent, "timestamps": self.timestamps, "add_value_fn": self.add_value_fn, @@ -141,12 +154,14 @@ def to_state(self) -> Dict[str, Any]: @classmethod def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": """Restore tracker from a state dict produced by to_state().""" + # Every optional key is read with .get(): a snapshot old enough to still + # carry the removed `expand_value` predates `add_value_fn` too, so + # indexing here would KeyError on exactly the states this tolerance is for. tracker = cls( min_samples=state["min_samples"], - expand_value=state.get("expand_value", False), time_dependent=state.get("time_dependent", False), - add_value_fn=state["add_value_fn"], - detector_config=state["detector_config"] + add_value_fn=state.get("add_value_fn", "default"), + detector_config=state.get("detector_config"), ) runs = [(bool(r[0]), int(r[1])) for r in state["runs"]] tracker.change_series._runs = runs @@ -197,7 +212,6 @@ class EventStabilityTracker(EventTracker): def __init__( self, converter_function: Callable[[Any], Any] = lambda x: x, - expand_value: bool = False, time_dependent: bool = False, add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None @@ -207,7 +221,6 @@ def __init__( def make_tracker() -> SingleStabilityTracker: return SingleStabilityTracker( - expand_value=expand_value, time_dependent=time_dependent, add_value_fn=add_value_fn, detector_config=detector_config, diff --git a/tests/test_detectors/test_new_value_combo_detector.py b/tests/test_detectors/test_new_value_combo_detector.py index 5d5fcd79..439af094 100644 --- a/tests/test_detectors/test_new_value_combo_detector.py +++ b/tests/test_detectors/test_new_value_combo_detector.py @@ -1,5 +1,8 @@ -from detectmatelibrary.detectors.new_value_combo_detector import NewValueComboDetector +from detectmatelibrary.detectors.new_value_combo_detector import ( + NewValueComboDetector, + NewValueComboDetectorConfig, +) from detectmatelibrary.utils.data_buffer import BufferMode from detectmatelibrary.common._config import generate_detector_config from detectmatelibrary.parsers.template_matcher import MatcherParser @@ -577,3 +580,101 @@ def test_audit_log_anomalies(self): detected_ids.add(log["logID"]) assert detected_ids == {"1859", "1862", "1865", "1866"} + + +class TestNewValueComboDetectorTimeDependentConfigPreservation: + """set_configuration() reassigns self.config twice (pass 1: combo + candidates, pass 2: final selection), each from a freshly generated config + dict whose params only ever carry max_combo_size. + + time_dependent, timestamp_variable and timestamp_format must survive + both reassignments, the same way persist already does. + """ + + def test_time_dependent_fields_survive_set_configuration(self): + cfg = NewValueComboDetectorConfig( + time_dependent=True, + timestamp_variable="level", + timestamp_format="%y%m%d %H%M%S", + ) + detector = NewValueComboDetector(config=cfg, name="NewValueComboDetector") + assert detector.config.auto_config is True # the default (set_configuration-first) path + + for i in range(10): + parser_data = schemas.ParserSchema({ + "parserType": "test", + "EventID": 1, + "template": "Template 1", + "variables": ["constant", f"varying_{i}", "another_constant"], + "logID": str(i), + "parsedLogID": str(i), + "parserID": "test_parser", + "log": "test log", + "logFormatVariables": {"level": "081109 203615"}, + }) + detector.configure(parser_data) + + detector.set_configuration(max_combo_size=2) + + assert detector.config.time_dependent is True + assert detector.config.timestamp_variable == "level" + assert detector.config.timestamp_format == "%y%m%d %H%M%S" + + +class TestNewValueComboDetectorTimeDependentCombos: + """The combo-stability pass must honour time_dependent too. + + auto_conf_persistency_combos is built directly in __init__ rather than from + the _event_data_kwargs hook, and its re-ingest loop in set_configuration + calls ingest_event itself -- so both halves of the flag (the tracker kwarg + and the per-record timestamp) have to be wired up explicitly. A flag that + reaches the first-pass trackers but not the combo trackers is worse than no + flag at all: the generated config would be selected on a different rule + than the one the operator asked for. + """ + + @staticmethod + def _records(time_dependent=True): + detector = NewValueComboDetector( + config=NewValueComboDetectorConfig( + time_dependent=time_dependent, + timestamp_variable="ts", + ), + name="NewValueComboDetector", + ) + for i in range(12): + # var_0 and var_1 both cycle over 3 values -> both STABLE in pass 1, + # so pass 2 gets a ("var_0", "var_1") combo tracker to look at. + detector.configure(schemas.ParserSchema({ + "parserType": "test", + "EventID": 1, + "template": "Template 1", + "variables": [f"a{i % 3}", f"b{i % 3}", "constant"], + "logID": str(i), + "parsedLogID": str(i), + "parserID": "test_parser", + "log": "test log", + "logFormatVariables": {"ts": str(1700000000 + i * 60)}, + })) + return detector + + def test_combo_trackers_record_timestamps(self): + detector = self._records() + detector.set_configuration(max_combo_size=2) + + combo_trackers = detector.auto_conf_persistency_combos.get_events_data()[1].get_data() + assert ("var_0", "var_1") in combo_trackers + tracker = combo_trackers[("var_0", "var_1")] + assert tracker.time_dependent is True + assert len(tracker.timestamps) == len(tracker.change_series) == 12 + assert tracker.timestamps[1] - tracker.timestamps[0] == 60.0 + + def test_combo_trackers_stay_count_based_when_flag_is_off(self): + detector = self._records(time_dependent=False) + detector.set_configuration(max_combo_size=2) + + tracker = detector.auto_conf_persistency_combos.get_events_data()[1].get_data()[ + ("var_0", "var_1") + ] + assert tracker.time_dependent is False + assert tracker.timestamps == [] diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py index c497fa80..f7285c0f 100644 --- a/tests/test_persistency/test_time_dependent_stability.py +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -1,5 +1,9 @@ """Tests for the time_dependent option of the stability trackers.""" +import logging + +import detectmatelibrary.schemas as schemas +from detectmatelibrary.detectors.charset_detector import CharsetDetector, CharsetDetectorConfig from detectmatelibrary.utils.persistency.rle_list import RLEList from detectmatelibrary.utils.persistency import EventPersistency from detectmatelibrary.utils.persistency.event_data_structures.trackers import ( @@ -18,8 +22,21 @@ def make_classifier() -> StabilityClassifier: # Divergence fixture: 3 changes up front, then a quiet tail of 37. # In *time*, the changes span most of the observed window and the quiet # tail is a burst compressed into ~0.04s at the end. +# +# The three change stamps are spaced so that each equal-duration quarter of the +# window holds at least one occurrence (boundaries [0, 1, 2, 3, 40]). An empty +# quarter would trip the empty-segment fallback in _segment_boundaries and drop +# the fixture back to count mode -- see +# TestClassifierTimeBoundaries.test_empty_time_segment_falls_back_to_count_mode. DIVERGENT_SERIES = [True, True, True] + [False] * 37 -DIVERGENT_TIMES = [0.0, 40.0, 80.0] + [100.0 + 0.001 * i for i in range(37)] +DIVERGENT_TIMES = [0.0, 30.0, 60.0] + [90.0 + 0.001 * i for i in range(37)] + +# Burst fixture: 20 fresh values, each immediately repeated once, inside 40 ms, +# then a single further repeat an hour later. Equal-duration quarters put all 40 +# early occurrences in quarter 0 and leave quarters 1 and 2 with nothing in them. +BURSTY_VALUES = [f"v{i // 2}" for i in range(40)] + ["v19"] +BURSTY_SERIES = [True, False] * 20 + [False] +BURSTY_TIMES = [0.001 * i for i in range(40)] + [3600.0] class TestClassifierTimeBoundaries: @@ -81,6 +98,78 @@ def test_nan_timestamp_entry_falls_back_to_count_mode(self): assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + def test_empty_time_segment_falls_back_to_count_mode(self): + """A segment with no occurrences has a nan mean, and `not nan >= + thresh` is True -- an unconditional pass. + + Raw equal-duration cuts of BURSTY_TIMES give boundaries [0, 40, + 40, 40, 41] and means [0.5, nan, nan, 0.0]: two free passes plus + segment 0 (whose 1.1 threshold is unreachable by a mean of + booleans), which is enough to call this churning variable + stable. Equal-count cuts keep every segment populated, so time + mode must defer to them here. + """ + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(BURSTY_SERIES)) + assert expected is False # count mode sees the churn + + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(BURSTY_SERIES), timestamps=BURSTY_TIMES) is False + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_empty_time_segment_fallback_on_plain_list_path(self): + clf_count = make_classifier() + expected = clf_count.is_stable(list(BURSTY_SERIES)) + clf_time = make_classifier() + assert clf_time.is_stable(list(BURSTY_SERIES), timestamps=BURSTY_TIMES) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_out_of_order_timestamps_fall_back_to_count_mode(self): + """np.searchsorted requires sorted input. + + UNSORTED_TIMES is SORTED_TIMES with two entries transposed -- + what a multi-source merge or concurrent writers produce. + Unguarded, the cuts land at [0, 9, 20, 31, 40] instead of [0, + 10, 20, 30, 40], turning the exact means [0.5, 0.5, 0.5, 0.5] + into [0.556, 0.455, 0.545, 0.444] -- wrong, and silently so. + """ + series = [True, False] * 20 + sorted_times = [float(i) for i in range(40)] + unsorted_times = list(sorted_times) + unsorted_times[9], unsorted_times[30] = unsorted_times[30], unsorted_times[9] + + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(series), timestamps=unsorted_times) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_sorted_timestamps_still_use_time_mode(self): + """The monotonicity guard must not disable time mode for valid + input.""" + clf = make_classifier() + assert clf.is_stable(RLEList(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + count = make_classifier() + count.is_stable(RLEList(DIVERGENT_SERIES)) + assert clf.get_last_segment_means() != count.get_last_segment_means() + + def test_equal_timestamps_are_not_treated_as_out_of_order(self): + """Duplicate stamps are non-decreasing, so they stay in time mode.""" + series = [True, False] * 20 + times = [float(i // 2) for i in range(40)] # each stamp used twice + clf = make_classifier() + assert clf.is_stable(RLEList(series), timestamps=times) is False + assert not any(mean != mean for mean in clf.get_last_segment_means()) # no nan + + def test_list_and_rle_agree_on_ragged_length(self): + """13 items over 4 segments: both paths must cut identically.""" + series = [True, False, True] + [False] * 10 + clf_list = make_classifier() + clf_list.is_stable(list(series)) + clf_rle = make_classifier() + clf_rle.is_stable(RLEList(series)) + assert clf_list.get_last_segment_means() == clf_rle.get_last_segment_means() + def feed_divergent(tracker: SingleStabilityTracker) -> None: """3 new values spread over ~80s, then a repeated value bursting at @@ -138,6 +227,36 @@ def test_legacy_state_without_time_keys_defaults_off(self): assert restored.time_dependent is False assert restored.timestamps == [] + def test_legacy_state_without_add_value_keys_loads(self): + """A snapshot old enough to predate add_value_fn/detector_config must + still load; those keys were indexed, not .get()-ed, so it raised + KeyError.""" + tracker = SingleStabilityTracker() + tracker.add_value("hello") + state = tracker.to_state() + state.pop("add_value_fn", None) + state.pop("detector_config", None) + state["expand_value"] = True # the flag such a snapshot would carry + restored = SingleStabilityTracker.from_state(state) + assert restored.add_value_fn == "default" + assert restored.detector_config is None + assert restored.unique_set == {"hello"} + + def test_bursty_series_is_unstable_in_both_modes(self): + """Time mode must not turn a churning variable STABLE. + + Empty equal-duration segments would pass unconditionally (means + [0.5, nan, nan, 0.0]) and hand this variable to auto-config + variable selection as a monitoring candidate. + """ + count_mode = SingleStabilityTracker() + time_mode = SingleStabilityTracker(time_dependent=True) + for value, ts in zip(BURSTY_VALUES, BURSTY_TIMES): + count_mode.add_value(value) + time_mode.add_value(value, timestamp=ts) + assert count_mode.classify().type == "UNSTABLE" + assert time_mode.classify().type == "UNSTABLE" + class TestTimeDependentPlumbing: def test_event_tracker_propagates_flag_and_timestamp(self): @@ -173,3 +292,210 @@ def test_event_tracker_dump_load_preserves_timestamps(self): single = restored.get_data()["var1"] assert single.time_dependent is True assert single.timestamps == [1.0, 2.0] + + +class TestTimeDependentWithDetectorAddValueFn: + """time_dependent must work when a detector owns the value semantics.""" + + def test_detector_backed_tracker_records_timestamps(self): + tracker = SingleStabilityTracker( + add_value_fn="CharsetDetector", time_dependent=True + ) + tracker.add_value("ab", timestamp=1.0) + tracker.add_value("cd", timestamp=2.0) + assert tracker.unique_set == {"a", "b", "c", "d"} + assert tracker.timestamps == [1.0, 2.0] + assert len(tracker.timestamps) == len(tracker.change_series) + + def test_value_range_skipped_value_keeps_alignment(self): + """ValueRangeDetector returns early on non-numeric input without + appending to change_series; timestamps must not drift.""" + tracker = SingleStabilityTracker( + add_value_fn="ValueRangeDetector", time_dependent=True + ) + tracker.add_value("1", timestamp=1.0) + tracker.add_value("not-a-number", timestamp=2.0) # detector records nothing + tracker.add_value("5", timestamp=3.0) + assert len(tracker.change_series) == 2 + assert tracker.timestamps == [1.0, 3.0] + + def test_event_tracker_detector_backed_round_trip(self): + event_tracker = EventStabilityTracker( + add_value_fn="CharsetDetector", time_dependent=True + ) + event_tracker.add_data({"var1": "ab"}, timestamp=1.0) + event_tracker.add_data({"var1": "cd"}, timestamp=2.0) + restored = EventStabilityTracker.load( + event_tracker.dump(), add_value_fn="CharsetDetector", time_dependent=True + ) + single = restored.get_data()["var1"] + assert single.unique_set == {"a", "b", "c", "d"} + assert single.timestamps == [1.0, 2.0] + + +def _parser_record(ts_value, event_id=1): + return schemas.ParserSchema({ + "parserType": "test", + "EventID": event_id, + "template": "test template", + "variables": ["abc"], + "logID": "1", + "parsedLogID": "1", + "parserID": "test_parser", + "log": "test log message", + "logFormatVariables": {"ts": ts_value}, + }) + + +class TestTimestampResolution: + # Each test below constructs CharsetDetector(config=CharsetDetectorConfig()) + # explicitly rather than bare CharsetDetector(). CharsetDetector.__init__'s + # `config` default argument is a single shared CharsetDetectorConfig() + # instance (pre-existing mutable-default-arg pitfall, see + # TestTimeDependentConfigWiring.test_flag_reaches_per_variable_trackers), and + # several tests here mutate `detector.config.*` in place -- writing through + # to that shared instance and leaking state into any other bare-constructed + # CharsetDetector for the rest of the process. Passing a fresh config keeps + # every test isolated regardless of run order. + def test_returns_none_when_not_configured(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + + def test_parses_iso_timestamp(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True + detector.config.timestamp_variable = "ts" + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) == 1785837600.0 + + def test_parses_explicit_format(self): + """HDFS loghub style, absent from COMMON_TIME_FORMATS.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True + detector.config.timestamp_variable = "ts" + detector.config.timestamp_format = "%y%m%d %H%M%S" + first = detector._timestamp(_parser_record("081109 203615")) + second = detector._timestamp(_parser_record("081109 203645")) + assert second - first == 30.0 + + def test_unparseable_warns_once_and_falls_back(self, caplog): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True + detector.config.timestamp_variable = "ts" + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("not-a-time")) is None + assert detector._timestamp(_parser_record("also-not-a-time")) is None + warnings = [r for r in caplog.records if "timestamp_variable" in r.message] + assert len(warnings) == 1 + + def test_unset_timestamp_variable_warns_once_and_falls_back(self, caplog): + """time_dependent without timestamp_variable is an operator error, not + an opt-out: it must be distinguishable from a working time-dependent + run, and must not flood the log.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True # timestamp_variable left unset + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert detector._timestamp(_parser_record("2026-08-04 10:00:01")) is None + warnings = [r for r in caplog.records if "timestamp_variable" in r.message] + assert len(warnings) == 1 + assert "not set" in warnings[0].message + + def test_flag_off_stays_silent(self, caplog): + """No warning when the feature simply is not enabled.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert not [r for r in caplog.records if "timestamp_variable" in r.message] + + def test_missing_variable_warns_and_falls_back(self, caplog): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True + detector.config.timestamp_variable = "absent" + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert any("timestamp_variable" in r.message for r in caplog.records) + + +class TestTimeDependentConfigWiring: + def test_flag_reaches_per_variable_trackers(self): + # CharsetDetector's `config` parameter default is a single shared + # CharsetDetectorConfig() instance (pre-existing mutable-default-arg + # pitfall, unrelated to time_dependent). Other tests in this module + # mutate `detector.config.*` in place on a bare CharsetDetector(), so + # we pass explicit fresh configs here to stay isolated from that. + detector = CharsetDetector(config=CharsetDetectorConfig()) + assert detector.persistency.event_data_kwargs.get("time_dependent") is None + + configured = CharsetDetector(config=CharsetDetectorConfig()) + configured.config.time_dependent = True + rebuilt = CharsetDetector(config=configured.config.to_dict(method_id="CharsetDetector")) + assert rebuilt.persistency.event_data_kwargs["time_dependent"] is True + + def test_config_fields_round_trip(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.time_dependent = True + detector.config.timestamp_variable = "ts" + detector.config.timestamp_format = "%y%m%d %H%M%S" + restored = type(detector.config).from_dict( + detector.config.to_dict(method_id="CharsetDetector"), "CharsetDetector" + ) + assert restored.time_dependent is True + assert restored.timestamp_variable == "ts" + assert restored.timestamp_format == "%y%m%d %H%M%S" + + def test_train_populates_timestamps_end_to_end(self): + cfg = { + "detectors": { + "CharsetDetector": { + "method_type": "charset_detector", + "auto_config": False, + "params": { + "time_dependent": True, + "timestamp_variable": "ts", + "timestamp_format": "%y%m%d %H%M%S", + }, + "events": { + 1: { + "inst": { + "params": {}, + "variables": [{"pos": 0, "name": "v", "params": {}}], + } + } + }, + } + } + } + detector = CharsetDetector(config=cfg, name="CharsetDetector") + detector.train(_parser_record("081109 203615")) + detector.train(_parser_record("081109 203645")) + tracker = detector.persistency.get_events_data()[1].get_data()["v"] + assert tracker.time_dependent is True + assert len(tracker.timestamps) == len(tracker.change_series) == 2 + assert tracker.timestamps[1] - tracker.timestamps[0] == 30.0 + + def test_time_dependent_fields_survive_auto_config_set_configuration(self): + """set_configuration() reassigns self.config wholesale from a config + dict generated with empty params (generate_detector_config only emits + method_type/auto_config/params/events), so time_dependent, + timestamp_variable and timestamp_format must be carried across that + reassignment explicitly -- same as `persist` already is. + + auto_config defaults to True and core.py runs + set_configuration() before train(), so this is the path every + detector takes unless auto_config is explicitly disabled. + """ + cfg = CharsetDetectorConfig( + time_dependent=True, + timestamp_variable="ts", + timestamp_format="%y%m%d %H%M%S", + ) + detector = CharsetDetector(config=cfg, name="CharsetDetector") + assert detector.config.auto_config is True + + for _ in range(5): + detector.configure(_parser_record("081109 203615")) + detector.set_configuration() + + assert detector.config.time_dependent is True + assert detector.config.timestamp_variable == "ts" + assert detector.config.timestamp_format == "%y%m%d %H%M%S" From 5936b015a44ca4f17276eef2e45ddf3e6b584889 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 4 Aug 2026 18:17:55 +0200 Subject: [PATCH 6/7] feat: add "both" stability segmentation, rename time_dependent Replace the boolean VariableDetectorConfig.time_dependent with stability_segmentation: Literal["count", "time", "both"] (tracker kwarg, state key and event_data_kwargs key: "segmentation"). "count" is the historical behaviour and stays the default. "both" requires a variable to be stable under equal-count *and* equal-duration segmentation. Neither subsumes the other -- a variable that churns in a burst and then settles is count-UNSTABLE but time-STABLE, and one whose late churn is buried under a dense settled tail is the reverse -- so the conjunction is strictly stricter than either input. It is implemented entirely in SingleStabilityTracker._is_stable(), which calls StabilityClassifier twice over the same change series; the classifier stays mode-agnostic and is unmodified. Degradation is free: when timestamps are missing, misaligned or unusable, _aligned_timestamps() returns None, the time pass reuses the count boundaries and "both" collapses to plain "count" rather than to an unconditional pass. classify() now reports the segment means on the UNSTABLE branch too, not only on STABLE -- that is the verdict worth debugging, and in "both" mode the reason carries both mean vectors. Co-Authored-By: Claude Opus 5 --- docs/detectors.md | 25 +- .../common/variable_detector.py | 34 +-- .../detectors/new_value_combo_detector.py | 16 +- .../trackers/base/event_tracker.py | 2 +- .../trackers/stability/stability_tracker.py | 67 +++-- .../test_new_value_combo_detector.py | 26 +- .../test_time_dependent_stability.py | 235 ++++++++++++++---- 7 files changed, 297 insertions(+), 108 deletions(-) diff --git a/docs/detectors.md b/docs/detectors.md index b787342b..71708318 100644 --- a/docs/detectors.md +++ b/docs/detectors.md @@ -222,7 +222,7 @@ def set_configuration(self): When `auto_config` is `False`, steps 1 and 2 are skipped entirely. -### Time-dependent stability (optional) +### Stability segmentation (optional) Stability classification splits a variable's change history into four segments and compares each segment's rate of change against a threshold. By default the segments @@ -231,8 +231,8 @@ time they cover. For bursty log sources that is misleading — a variable that c constantly during a quiet night and then went silent under a flood of daytime traffic looks stable, because the flood supplies enough samples to dominate the later segments. -Setting `time_dependent: true` switches the segmentation to **equal-duration** cuts of -the observed time span, so each segment covers the same amount of wall-clock time. The +Setting `stability_segmentation: time` switches the segmentation to **equal-duration** cuts +of the observed time span, so each segment covers the same amount of wall-clock time. The detector then needs an event time per record, which it reads from the log's named variables (`logFormatVariables`, i.e. the fields declared in the parser's `log_format`) under the name given by `timestamp_variable`. @@ -247,17 +247,24 @@ detectors: method_type: new_value_detector auto_config: True params: - time_dependent: True + stability_segmentation: time timestamp_variable: Time # a field name from the parser's log_format timestamp_format: "%y%m%d %H%M%S" # optional; omit to auto-detect ``` +Setting `stability_segmentation: both` runs *both* segmentations and calls the variable +stable only when each one does. Neither segmentation subsumes the other — a variable that +churns in a burst and then settles is unstable by count but stable by time, and one whose +late churn is buried under a dense tail of repeats is the reverse — so `both` is strictly +stricter than either. Use it when a false "stable" is more costly than a missed one; use +`time` when the point is specifically to forgive early churn on a bursty source. + #### Fields | Field | Type | Default | Description | |---|---|---|---| -| `time_dependent` | `bool` | `false` | Segment the change history by equal time spans instead of equal sample counts. When `false` (the default) the other two fields are ignored and no timestamps are recorded. | -| `timestamp_variable` | `str \| null` | `null` | Name of the field in `logFormatVariables` holding the record's event time. Required for `time_dependent` to have any effect. Only named log-format fields are consulted — never the positional `variables` list. | +| `stability_segmentation` | `"count" \| "time" \| "both"` | `"count"` | How to cut the change history into segments. `count` uses equal sample counts; `time` uses equal time spans; `both` requires the variable to be stable under each. With `count` the other two fields are ignored and no timestamps are recorded. | +| `timestamp_variable` | `str \| null` | `null` | Name of the field in `logFormatVariables` holding the record's event time. Required for `time` and `both` to have any effect. Only named log-format fields are consulted — never the positional `variables` list. | | `timestamp_format` | `str \| null` | `null` | Explicit [`strftime`](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes) pattern for parsing that field. When unset, `TimeFormatHandler` auto-detects the format (ISO 8601, Apache, syslog, numeric epoch seconds/milliseconds, and other common layouts). | Set `timestamp_format` when the source uses a layout the auto-detection does not @@ -266,9 +273,9 @@ only parses with an explicit `"%y%m%d %H%M%S"`. #### Fallback behaviour -Time-dependent segmentation is best-effort and never fails a run: +Time-aware segmentation is best-effort and never fails a run: -* If `time_dependent` is `true` but `timestamp_variable` is unset, or the named field +* If `stability_segmentation` is not `count` but `timestamp_variable` is unset, or the named field is absent from a record, or its value cannot be parsed, the detector logs a **single** warning (once per detector, so a bad config cannot flood the log) and falls back to count-based segmentation. @@ -276,6 +283,8 @@ Time-dependent segmentation is best-effort and never fails a run: span is zero, or the equal-duration cut would leave a segment with no observations in it, the classifier silently falls back to count-based segmentation for that variable. +* Under `both`, any of the fallbacks above make the time pass reuse the count boundaries, + so the mode degrades to plain `count` rather than to an unconditional pass. In every fallback case classification still runs and produces a result — only the segmentation rule changes back to the default. diff --git a/src/detectmatelibrary/common/variable_detector.py b/src/detectmatelibrary/common/variable_detector.py index bc012f0b..928f1c45 100644 --- a/src/detectmatelibrary/common/variable_detector.py +++ b/src/detectmatelibrary/common/variable_detector.py @@ -18,7 +18,7 @@ from detectmatelibrary.constants import GLOBAL_EVENT_ID from detectmatelibrary.tools.logging import logger -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, Literal, Optional, cast from typing_extensions import override @@ -26,10 +26,12 @@ class VariableDetectorConfig(CoreDetectorConfig): use_stable_vars: bool = True use_static_vars: bool = True - # Time-dependent stability: cut the classifier's segments at equal-duration - # boundaries instead of equal-count ones. Needs a per-record event time, + # Stability segmentation. "count" cuts the classifier's segments at equal + # sample counts (the historical behaviour). "time" cuts them at equal + # durations instead. "both" requires the variable to pass under *both* + # segmentations. The two time-aware modes need a per-record event time, # named here and read from the record's logFormatVariables. - time_dependent: bool = False + stability_segmentation: Literal["count", "time", "both"] = "count" timestamp_variable: str | None = None timestamp_format: str | None = None # None -> TimeFormatHandler auto-detect @@ -56,26 +58,27 @@ def __init__(self, name: str, config: VariableDetectorConfig) -> None: self._warned_bad_timestamp = False self.persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._with_time_flag(self._event_data_kwargs()), + event_data_kwargs=self._with_segmentation(self._event_data_kwargs()), ) # auto config checks individual-variable stability to select features self.auto_conf_persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._with_time_flag(self._auto_conf_kwargs()), + event_data_kwargs=self._with_segmentation(self._auto_conf_kwargs()), ) self._register_persistency(self.persistency) - def _with_time_flag(self, kwargs: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: - """Add time_dependent to tracker kwargs when the config asks for it. + def _with_segmentation(self, kwargs: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Add the segmentation mode to tracker kwargs when it is not the + default. Done here rather than in _stability_kwargs so every VariableDetector subclass is covered -- NewValueDetector overrides neither construction hook and NewValueComboDetector returns only a converter_function. """ - if not self.config.time_dependent: + if self.config.stability_segmentation == "count": return kwargs - return {**(kwargs or {}), "time_dependent": True} + return {**(kwargs or {}), "segmentation": self.config.stability_segmentation} # ---- construction hooks ------------------------------------------------- @@ -114,13 +117,14 @@ def _warn_time_fallback_once(self, reason: str) -> None: def _timestamp(self, input_: ParserSchema) -> float | None: """Resolve the record's event time, or None to use count segmentation.""" - if not self.config.time_dependent: + if self.config.stability_segmentation == "count": return None if not self.config.timestamp_variable: - # Enabling the feature without naming the field is an operator + # Selecting a time-aware mode without naming the field is an operator # error, not an opt-out -- say so rather than silently no-op. self._warn_time_fallback_once( - "time_dependent is enabled but timestamp_variable is not set" + f"stability_segmentation is {self.config.stability_segmentation!r} " + "but timestamp_variable is not set" ) return None raw = input_["logFormatVariables"].get(self.config.timestamp_variable) @@ -260,7 +264,7 @@ def set_configuration(self) -> None: if selected: variables[event_id] = selected old_persist = self.config.persist - old_time_dependent = self.config.time_dependent + old_segmentation = self.config.stability_segmentation old_timestamp_variable = self.config.timestamp_variable old_timestamp_format = self.config.timestamp_format config_dict = generate_detector_config( @@ -270,7 +274,7 @@ def set_configuration(self) -> None: ) self.config = type(self.config).from_dict(config_dict, self.name) self.config.persist = old_persist - self.config.time_dependent = old_time_dependent + self.config.stability_segmentation = old_segmentation self.config.timestamp_variable = old_timestamp_variable self.config.timestamp_format = old_timestamp_format events = self.config.events diff --git a/src/detectmatelibrary/detectors/new_value_combo_detector.py b/src/detectmatelibrary/detectors/new_value_combo_detector.py index 31f223b3..84dc5f26 100644 --- a/src/detectmatelibrary/detectors/new_value_combo_detector.py +++ b/src/detectmatelibrary/detectors/new_value_combo_detector.py @@ -59,7 +59,7 @@ def __init__( # second-pass persistency to learn stability of variable combinations self.auto_conf_persistency_combos = persistency.EventPersistency( event_data_class=persistency.EventStabilityTracker, - event_data_kwargs=self._with_time_flag( + event_data_kwargs=self._with_segmentation( {"converter_function": get_all_possible_combos} ), ) @@ -103,14 +103,14 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: every possible combo up front would explode combinatorially). """ old_persist = self.config.persist - time_fields = { - "time_dependent": self.config.time_dependent, + segmentation_fields = { + "stability_segmentation": self.config.stability_segmentation, "timestamp_variable": self.config.timestamp_variable, "timestamp_format": self.config.timestamp_format, } - def restore_time_fields() -> None: - """Carry the time settings across a config reassignment. + def restore_segmentation_fields() -> None: + """Carry the segmentation settings across a config reassignment. generate_detector_config only emits method_type / auto_config / params / events, so every ``from_dict`` below resets these to their @@ -118,7 +118,7 @@ def restore_time_fields() -> None: config, so restoring only at the end would leave the combo trackers timestamp-less. """ - for field, value in time_fields.items(): + for field, value in segmentation_fields.items(): setattr(self.config, field, value) # pass 1: stable individual variables -> combos @@ -134,7 +134,7 @@ def restore_time_fields() -> None: max_combo_size=max_combo_size or self.config.max_combo_size, ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) - restore_time_fields() + restore_segmentation_fields() # re-ingest all inputs to learn combos under the new configuration for input_ in self.inputs: @@ -170,7 +170,7 @@ def restore_time_fields() -> None: ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) self.config.persist = old_persist - restore_time_fields() + restore_segmentation_fields() events = self.config.events if isinstance(events, EventsConfig) and not events.events: logger.warning( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py index 98067edd..7e12103a 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py @@ -72,7 +72,7 @@ def load(cls, data: bytes, **kwargs: Any) -> "EventTracker": ``multi_tracker_type`` recorded in the snapshot. For any subclass, ``cls(**kwargs)`` is called instead, which lets subclasses with closure-based factories (e.g. ``EventStabilityTracker``'s - ``time_dependent``) rebuild their factory so it survives load. + ``segmentation``) rebuild their factory so it survives load. Contract for subclasses: ``__init__`` must accept the kwargs forwarded to ``load()`` and must not require additional positional arguments. diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index 575f90ba..a66ea4b3 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -46,12 +46,12 @@ class SingleStabilityTracker(SingleTracker): def __init__( self, min_samples: int = 3, - time_dependent: bool = False, + segmentation: Literal["count", "time", "both"] = "count", add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None, ) -> None: self.min_samples = min_samples - self.time_dependent = time_dependent + self.segmentation = segmentation self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( @@ -65,6 +65,9 @@ def __init__( self.extra_state: Dict[str, Any] = {} self.add_value_fn = add_value_fn self.detector_config = detector_config + # Transient: set by _is_stable() for classify()'s reason string. Not + # persisted -- it is derived from change_series on every classify(). + self._stability_note: str = "" self._value_fn: Callable[[Any], None] = self._default_add_value if add_value_fn != "default": detector_cls = getattr(importlib.import_module("detectmatelibrary.detectors"), add_value_fn) @@ -91,7 +94,7 @@ def add_value(self, value: Any, timestamp: float | None = None) -> None: """ before = len(self.change_series) self._value_fn(value) - if self.time_dependent and timestamp is not None and len(self.change_series) > before: + if self.segmentation != "count" and timestamp is not None and len(self.change_series) > before: self.timestamps.append(float(timestamp)) def classify(self) -> Classification: @@ -111,26 +114,62 @@ def classify(self) -> Classification: type="RANDOM", reason=f"Unique set size equals number of samples ({len(self.change_series)})" ) - elif self.stability_classifier.is_stable( - self.change_series, timestamps=self._aligned_timestamps() - ): + elif self._is_stable(): return Classification( type="STABLE", reason=( - f"Segment means of change series {self.stability_classifier.get_last_segment_means()} " - f"are below segment thresholds: {self.stability_classifier.get_segment_thresholds()}" + f"{self._stability_note} are below segment thresholds: " + f"{self.stability_classifier.get_segment_thresholds()}" ) ) else: return Classification( type="UNSTABLE", - reason="No classification matched; variable is unstable" + reason=( + f"{self._stability_note} exceed segment thresholds: " + f"{self.stability_classifier.get_segment_thresholds()}" + ) ) + def _is_stable(self) -> bool: + """Stability verdict under the configured segmentation. + + Sets ``_stability_note`` for ``classify()``'s reason string. + + ``both`` runs the count pass and the time pass over the same + change series and requires both. Neither segmentation subsumes + the other -- a variable that churns in a burst and then settles is + count-UNSTABLE but time-STABLE, and one whose late churn is buried + under a dense settled tail is the reverse -- so the conjunction is + strictly stricter than either input. + + Deliberately not short-circuited: both passes always run so the + note carries both mean vectors, which is what anyone debugging a + ``both`` verdict needs. Costs one extra O(runs x n_segments) scan. + """ + clf, ts = self.stability_classifier, self._aligned_timestamps() + if self.segmentation != "both": + stable = clf.is_stable(self.change_series, timestamps=ts) + self._stability_note = f"Segment means of change series {clf.get_last_segment_means()}" + return stable + count_stable = clf.is_stable(self.change_series) + # Snapshot now, not after the time pass: is_stable() rebinds + # clf.segment_means to a fresh list on every call, so calling + # get_last_segment_means() after the time pass below would return + # the time means for both halves of the note instead of the count + # means it is meant to capture here. + count_means = clf.get_last_segment_means() + time_stable = clf.is_stable(self.change_series, timestamps=ts) + self._stability_note = ( + f"Segment means of change series: count {count_means}, " + f"time {clf.get_last_segment_means()}" + ) + return count_stable and time_stable + def _aligned_timestamps(self) -> List[float] | None: """Timestamps to classify with, or None to fall back to count segments.""" - if self.time_dependent and len(self.timestamps) == len(self.change_series): + if self.segmentation != "count" and len(self.timestamps) == len(self.change_series): return self.timestamps return None @@ -141,7 +180,7 @@ def to_state(self) -> Dict[str, Any]: "type": self.__class__.__name__, "module": self.__class__.__module__, "min_samples": self.min_samples, - "time_dependent": self.time_dependent, + "segmentation": self.segmentation, "timestamps": self.timestamps, "add_value_fn": self.add_value_fn, "detector_config": self.detector_config, @@ -159,7 +198,7 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": # indexing here would KeyError on exactly the states this tolerance is for. tracker = cls( min_samples=state["min_samples"], - time_dependent=state.get("time_dependent", False), + segmentation=state.get("segmentation", "count"), add_value_fn=state.get("add_value_fn", "default"), detector_config=state.get("detector_config"), ) @@ -212,7 +251,7 @@ class EventStabilityTracker(EventTracker): def __init__( self, converter_function: Callable[[Any], Any] = lambda x: x, - time_dependent: bool = False, + segmentation: Literal["count", "time", "both"] = "count", add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None @@ -221,7 +260,7 @@ def __init__( def make_tracker() -> SingleStabilityTracker: return SingleStabilityTracker( - time_dependent=time_dependent, + segmentation=segmentation, add_value_fn=add_value_fn, detector_config=detector_config, ) diff --git a/tests/test_detectors/test_new_value_combo_detector.py b/tests/test_detectors/test_new_value_combo_detector.py index 439af094..cc1f319e 100644 --- a/tests/test_detectors/test_new_value_combo_detector.py +++ b/tests/test_detectors/test_new_value_combo_detector.py @@ -582,18 +582,18 @@ def test_audit_log_anomalies(self): assert detected_ids == {"1859", "1862", "1865", "1866"} -class TestNewValueComboDetectorTimeDependentConfigPreservation: +class TestNewValueComboDetectorSegmentationConfigPreservation: """set_configuration() reassigns self.config twice (pass 1: combo candidates, pass 2: final selection), each from a freshly generated config dict whose params only ever carry max_combo_size. - time_dependent, timestamp_variable and timestamp_format must survive - both reassignments, the same way persist already does. + stability_segmentation, timestamp_variable and timestamp_format must + survive both reassignments, the same way persist already does. """ - def test_time_dependent_fields_survive_set_configuration(self): + def test_segmentation_fields_survive_set_configuration(self): cfg = NewValueComboDetectorConfig( - time_dependent=True, + stability_segmentation="time", timestamp_variable="level", timestamp_format="%y%m%d %H%M%S", ) @@ -616,13 +616,13 @@ def test_time_dependent_fields_survive_set_configuration(self): detector.set_configuration(max_combo_size=2) - assert detector.config.time_dependent is True + assert detector.config.stability_segmentation == "time" assert detector.config.timestamp_variable == "level" assert detector.config.timestamp_format == "%y%m%d %H%M%S" -class TestNewValueComboDetectorTimeDependentCombos: - """The combo-stability pass must honour time_dependent too. +class TestNewValueComboDetectorSegmentationCombos: + """The combo-stability pass must honour stability_segmentation too. auto_conf_persistency_combos is built directly in __init__ rather than from the _event_data_kwargs hook, and its re-ingest loop in set_configuration @@ -634,10 +634,10 @@ class TestNewValueComboDetectorTimeDependentCombos: """ @staticmethod - def _records(time_dependent=True): + def _records(segmentation="time"): detector = NewValueComboDetector( config=NewValueComboDetectorConfig( - time_dependent=time_dependent, + stability_segmentation=segmentation, timestamp_variable="ts", ), name="NewValueComboDetector", @@ -665,16 +665,16 @@ def test_combo_trackers_record_timestamps(self): combo_trackers = detector.auto_conf_persistency_combos.get_events_data()[1].get_data() assert ("var_0", "var_1") in combo_trackers tracker = combo_trackers[("var_0", "var_1")] - assert tracker.time_dependent is True + assert tracker.segmentation == "time" assert len(tracker.timestamps) == len(tracker.change_series) == 12 assert tracker.timestamps[1] - tracker.timestamps[0] == 60.0 def test_combo_trackers_stay_count_based_when_flag_is_off(self): - detector = self._records(time_dependent=False) + detector = self._records(segmentation="count") detector.set_configuration(max_combo_size=2) tracker = detector.auto_conf_persistency_combos.get_events_data()[1].get_data()[ ("var_0", "var_1") ] - assert tracker.time_dependent is False + assert tracker.segmentation == "count" assert tracker.timestamps == [] diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py index f7285c0f..328d2ba5 100644 --- a/tests/test_persistency/test_time_dependent_stability.py +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -1,4 +1,4 @@ -"""Tests for the time_dependent option of the stability trackers.""" +"""Tests for the stability_segmentation option of the stability trackers.""" import logging @@ -38,6 +38,24 @@ def make_classifier() -> StabilityClassifier: BURSTY_SERIES = [True, False] * 20 + [False] BURSTY_TIMES = [0.001 * i for i in range(40)] + [3600.0] +# Opposite-direction divergence fixture. DIVERGENT_* above is count-STABLE and +# time-UNSTABLE; this one is count-UNSTABLE and time-STABLE. 30 fresh values one +# second apart, then the same value repeated 10 times spread over ~17 minutes. +# +# count quarters -> means [1.0, 1.0, 1.0, 0.0] -> UNSTABLE (segments 2 and 3) +# time quarters -> means [0.938, 0.0, 0.0, 0.0] -> STABLE (0.938 < 1.1) +# +# The pair of fixtures is what makes "both" testable in each direction: neither +# segmentation subsumes the other. +CHURN_VALUES = [f"v{i}" for i in range(30)] + ["v29"] * 10 +CHURN_TIMES = [float(i) for i in range(30)] + [100.0 * (i + 1) for i in range(10)] + +# Both segmentations agree on STABLE: two values, then one of them repeated, +# evenly spaced so the duration cuts coincide with the count cuts. +# Both give means [0.2, 0.0, 0.0, 0.0]. +AGREE_VALUES = ["a", "b"] + ["b"] * 38 +AGREE_TIMES = [float(i) for i in range(40)] + class TestClassifierTimeBoundaries: def test_count_mode_is_stable_on_divergent_fixture(self): @@ -179,14 +197,26 @@ def feed_divergent(tracker: SingleStabilityTracker) -> None: tracker.add_value(value, timestamp=ts) -class TestSingleStabilityTrackerTimeDependent: +def feed_churn(tracker: SingleStabilityTracker) -> None: + """30 new values one second apart, then a repeated value over ~17 min.""" + for value, ts in zip(CHURN_VALUES, CHURN_TIMES): + tracker.add_value(value, timestamp=ts) + + +def feed_agreeing(tracker: SingleStabilityTracker) -> None: + """One change up front, then a settled value, evenly spaced.""" + for value, ts in zip(AGREE_VALUES, AGREE_TIMES): + tracker.add_value(value, timestamp=ts) + + +class TestSingleStabilityTrackerSegmentation: def test_timestamps_stored_only_when_enabled(self): - on = SingleStabilityTracker(time_dependent=True) + on = SingleStabilityTracker(segmentation="time") on.add_value("a", timestamp=1.0) on.add_value("b", timestamp=2.0) assert on.timestamps == [1.0, 2.0] - off = SingleStabilityTracker() # default time_dependent=False + off = SingleStabilityTracker() # default segmentation="count" off.add_value("a", timestamp=1.0) assert off.timestamps == [] @@ -195,13 +225,13 @@ def test_classification_diverges_between_modes(self): feed_divergent(count_mode) assert count_mode.classify().type == "STABLE" - time_mode = SingleStabilityTracker(time_dependent=True) + time_mode = SingleStabilityTracker(segmentation="time") feed_divergent(time_mode) assert time_mode.classify().type == "UNSTABLE" def test_missing_timestamps_fall_back_to_count_mode(self): - # time_dependent on, but values arrive without timestamps - tracker = SingleStabilityTracker(time_dependent=True) + # time segmentation on, but values arrive without timestamps + tracker = SingleStabilityTracker(segmentation="time") for value in ["a", "b", "c"] + ["c"] * 37: tracker.add_value(value) reference = SingleStabilityTracker() @@ -210,10 +240,10 @@ def test_missing_timestamps_fall_back_to_count_mode(self): assert tracker.classify().type == reference.classify().type def test_round_trip_preserves_time_state(self): - tracker = SingleStabilityTracker(time_dependent=True) + tracker = SingleStabilityTracker(segmentation="time") feed_divergent(tracker) restored = SingleStabilityTracker.from_state(tracker.to_state()) - assert restored.time_dependent is True + assert restored.segmentation == "time" assert restored.timestamps == tracker.timestamps assert restored.classify().type == "UNSTABLE" @@ -221,10 +251,10 @@ def test_legacy_state_without_time_keys_defaults_off(self): tracker = SingleStabilityTracker() tracker.add_value("hello") state = tracker.to_state() - state.pop("time_dependent", None) # simulate pre-flag snapshot + state.pop("segmentation", None) # simulate pre-flag snapshot state.pop("timestamps", None) restored = SingleStabilityTracker.from_state(state) - assert restored.time_dependent is False + assert restored.segmentation == "count" assert restored.timestamps == [] def test_legacy_state_without_add_value_keys_loads(self): @@ -250,7 +280,7 @@ def test_bursty_series_is_unstable_in_both_modes(self): variable selection as a monitoring candidate. """ count_mode = SingleStabilityTracker() - time_mode = SingleStabilityTracker(time_dependent=True) + time_mode = SingleStabilityTracker(segmentation="time") for value, ts in zip(BURSTY_VALUES, BURSTY_TIMES): count_mode.add_value(value) time_mode.add_value(value, timestamp=ts) @@ -258,19 +288,19 @@ def test_bursty_series_is_unstable_in_both_modes(self): assert time_mode.classify().type == "UNSTABLE" -class TestTimeDependentPlumbing: +class TestSegmentationPlumbing: def test_event_tracker_propagates_flag_and_timestamp(self): - event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker = EventStabilityTracker(segmentation="time") event_tracker.add_data({"var1": "a"}, timestamp=1.0) event_tracker.add_data({"var1": "b"}, timestamp=2.0) single = event_tracker.get_data()["var1"] - assert single.time_dependent is True + assert single.segmentation == "time" assert single.timestamps == [1.0, 2.0] def test_ingest_event_forwards_timestamp(self): storage = EventPersistency( EventStabilityTracker, - event_data_kwargs={"time_dependent": True}, + event_data_kwargs={"segmentation": "time"}, ) storage.ingest_event(1, "tpl <*>", variables=["a"], timestamp=10.0) storage.ingest_event(1, "tpl <*>", variables=["b"], timestamp=20.0) @@ -285,21 +315,22 @@ def test_ingest_event_without_timestamp_still_works(self): assert single.timestamps == [] def test_event_tracker_dump_load_preserves_timestamps(self): - event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker = EventStabilityTracker(segmentation="time") event_tracker.add_data({"var1": "a"}, timestamp=1.0) event_tracker.add_data({"var1": "b"}, timestamp=2.0) - restored = EventStabilityTracker.load(event_tracker.dump(), time_dependent=True) + restored = EventStabilityTracker.load(event_tracker.dump(), segmentation="time") single = restored.get_data()["var1"] - assert single.time_dependent is True + assert single.segmentation == "time" assert single.timestamps == [1.0, 2.0] -class TestTimeDependentWithDetectorAddValueFn: - """time_dependent must work when a detector owns the value semantics.""" +class TestSegmentationWithDetectorAddValueFn: + """Segmentation="time" must work when a detector owns the value + semantics.""" def test_detector_backed_tracker_records_timestamps(self): tracker = SingleStabilityTracker( - add_value_fn="CharsetDetector", time_dependent=True + add_value_fn="CharsetDetector", segmentation="time" ) tracker.add_value("ab", timestamp=1.0) tracker.add_value("cd", timestamp=2.0) @@ -311,7 +342,7 @@ def test_value_range_skipped_value_keeps_alignment(self): """ValueRangeDetector returns early on non-numeric input without appending to change_series; timestamps must not drift.""" tracker = SingleStabilityTracker( - add_value_fn="ValueRangeDetector", time_dependent=True + add_value_fn="ValueRangeDetector", segmentation="time" ) tracker.add_value("1", timestamp=1.0) tracker.add_value("not-a-number", timestamp=2.0) # detector records nothing @@ -321,12 +352,12 @@ def test_value_range_skipped_value_keeps_alignment(self): def test_event_tracker_detector_backed_round_trip(self): event_tracker = EventStabilityTracker( - add_value_fn="CharsetDetector", time_dependent=True + add_value_fn="CharsetDetector", segmentation="time" ) event_tracker.add_data({"var1": "ab"}, timestamp=1.0) event_tracker.add_data({"var1": "cd"}, timestamp=2.0) restored = EventStabilityTracker.load( - event_tracker.dump(), add_value_fn="CharsetDetector", time_dependent=True + event_tracker.dump(), add_value_fn="CharsetDetector", segmentation="time" ) single = restored.get_data()["var1"] assert single.unique_set == {"a", "b", "c", "d"} @@ -352,7 +383,7 @@ class TestTimestampResolution: # explicitly rather than bare CharsetDetector(). CharsetDetector.__init__'s # `config` default argument is a single shared CharsetDetectorConfig() # instance (pre-existing mutable-default-arg pitfall, see - # TestTimeDependentConfigWiring.test_flag_reaches_per_variable_trackers), and + # TestSegmentationConfigWiring.test_flag_reaches_per_variable_trackers), and # several tests here mutate `detector.config.*` in place -- writing through # to that shared instance and leaking state into any other bare-constructed # CharsetDetector for the rest of the process. Passing a fresh config keeps @@ -363,14 +394,14 @@ def test_returns_none_when_not_configured(self): def test_parses_iso_timestamp(self): detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True + detector.config.stability_segmentation = "time" detector.config.timestamp_variable = "ts" assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) == 1785837600.0 def test_parses_explicit_format(self): """HDFS loghub style, absent from COMMON_TIME_FORMATS.""" detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True + detector.config.stability_segmentation = "time" detector.config.timestamp_variable = "ts" detector.config.timestamp_format = "%y%m%d %H%M%S" first = detector._timestamp(_parser_record("081109 203615")) @@ -379,7 +410,7 @@ def test_parses_explicit_format(self): def test_unparseable_warns_once_and_falls_back(self, caplog): detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True + detector.config.stability_segmentation = "time" detector.config.timestamp_variable = "ts" with caplog.at_level(logging.WARNING): assert detector._timestamp(_parser_record("not-a-time")) is None @@ -388,11 +419,11 @@ def test_unparseable_warns_once_and_falls_back(self, caplog): assert len(warnings) == 1 def test_unset_timestamp_variable_warns_once_and_falls_back(self, caplog): - """time_dependent without timestamp_variable is an operator error, not - an opt-out: it must be distinguishable from a working time-dependent - run, and must not flood the log.""" + """stability_segmentation="time" without timestamp_variable is an + operator error, not an opt-out: it must be distinguishable from a + working time-dependent run, and must not flood the log.""" detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True # timestamp_variable left unset + detector.config.stability_segmentation = "time" # timestamp_variable left unset with caplog.at_level(logging.WARNING): assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None assert detector._timestamp(_parser_record("2026-08-04 10:00:01")) is None @@ -409,37 +440,38 @@ def test_flag_off_stays_silent(self, caplog): def test_missing_variable_warns_and_falls_back(self, caplog): detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True + detector.config.stability_segmentation = "time" detector.config.timestamp_variable = "absent" with caplog.at_level(logging.WARNING): assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None assert any("timestamp_variable" in r.message for r in caplog.records) -class TestTimeDependentConfigWiring: +class TestSegmentationConfigWiring: def test_flag_reaches_per_variable_trackers(self): # CharsetDetector's `config` parameter default is a single shared # CharsetDetectorConfig() instance (pre-existing mutable-default-arg - # pitfall, unrelated to time_dependent). Other tests in this module - # mutate `detector.config.*` in place on a bare CharsetDetector(), so - # we pass explicit fresh configs here to stay isolated from that. + # pitfall, unrelated to stability_segmentation). Other tests in this + # module mutate `detector.config.*` in place on a bare + # CharsetDetector(), so we pass explicit fresh configs here to stay + # isolated from that. detector = CharsetDetector(config=CharsetDetectorConfig()) - assert detector.persistency.event_data_kwargs.get("time_dependent") is None + assert detector.persistency.event_data_kwargs.get("segmentation") is None configured = CharsetDetector(config=CharsetDetectorConfig()) - configured.config.time_dependent = True + configured.config.stability_segmentation = "time" rebuilt = CharsetDetector(config=configured.config.to_dict(method_id="CharsetDetector")) - assert rebuilt.persistency.event_data_kwargs["time_dependent"] is True + assert rebuilt.persistency.event_data_kwargs["segmentation"] == "time" def test_config_fields_round_trip(self): detector = CharsetDetector(config=CharsetDetectorConfig()) - detector.config.time_dependent = True + detector.config.stability_segmentation = "time" detector.config.timestamp_variable = "ts" detector.config.timestamp_format = "%y%m%d %H%M%S" restored = type(detector.config).from_dict( detector.config.to_dict(method_id="CharsetDetector"), "CharsetDetector" ) - assert restored.time_dependent is True + assert restored.stability_segmentation == "time" assert restored.timestamp_variable == "ts" assert restored.timestamp_format == "%y%m%d %H%M%S" @@ -450,7 +482,7 @@ def test_train_populates_timestamps_end_to_end(self): "method_type": "charset_detector", "auto_config": False, "params": { - "time_dependent": True, + "stability_segmentation": "time", "timestamp_variable": "ts", "timestamp_format": "%y%m%d %H%M%S", }, @@ -469,14 +501,14 @@ def test_train_populates_timestamps_end_to_end(self): detector.train(_parser_record("081109 203615")) detector.train(_parser_record("081109 203645")) tracker = detector.persistency.get_events_data()[1].get_data()["v"] - assert tracker.time_dependent is True + assert tracker.segmentation == "time" assert len(tracker.timestamps) == len(tracker.change_series) == 2 assert tracker.timestamps[1] - tracker.timestamps[0] == 30.0 - def test_time_dependent_fields_survive_auto_config_set_configuration(self): + def test_segmentation_fields_survive_auto_config_set_configuration(self): """set_configuration() reassigns self.config wholesale from a config dict generated with empty params (generate_detector_config only emits - method_type/auto_config/params/events), so time_dependent, + method_type/auto_config/params/events), so stability_segmentation, timestamp_variable and timestamp_format must be carried across that reassignment explicitly -- same as `persist` already is. @@ -485,7 +517,7 @@ def test_time_dependent_fields_survive_auto_config_set_configuration(self): detector takes unless auto_config is explicitly disabled. """ cfg = CharsetDetectorConfig( - time_dependent=True, + stability_segmentation="time", timestamp_variable="ts", timestamp_format="%y%m%d %H%M%S", ) @@ -496,6 +528,111 @@ def test_time_dependent_fields_survive_auto_config_set_configuration(self): detector.configure(_parser_record("081109 203615")) detector.set_configuration() - assert detector.config.time_dependent is True + assert detector.config.stability_segmentation == "time" assert detector.config.timestamp_variable == "ts" assert detector.config.timestamp_format == "%y%m%d %H%M%S" + + +class TestBothSegmentation: + """`both` is STABLE only when count and time segmentation agree.""" + + def test_rejects_when_only_time_is_unstable(self): + count_mode = SingleStabilityTracker() + feed_divergent(count_mode) + assert count_mode.classify().type == "STABLE" + + time_mode = SingleStabilityTracker(segmentation="time") + feed_divergent(time_mode) + assert time_mode.classify().type == "UNSTABLE" + + both_mode = SingleStabilityTracker(segmentation="both") + feed_divergent(both_mode) + assert both_mode.classify().type == "UNSTABLE" + + def test_rejects_when_only_count_is_unstable(self): + """The opposite direction: proves neither pass is dead code.""" + count_mode = SingleStabilityTracker() + feed_churn(count_mode) + assert count_mode.classify().type == "UNSTABLE" + + time_mode = SingleStabilityTracker(segmentation="time") + feed_churn(time_mode) + assert time_mode.classify().type == "STABLE" + + both_mode = SingleStabilityTracker(segmentation="both") + feed_churn(both_mode) + assert both_mode.classify().type == "UNSTABLE" + + def test_accepts_when_both_agree(self): + """`both` must not be vacuously strict.""" + for mode in ("count", "time", "both"): + tracker = SingleStabilityTracker(segmentation=mode) + feed_agreeing(tracker) + assert tracker.classify().type == "STABLE", mode + + def test_without_timestamps_matches_count_mode(self): + """No usable timestamps -> the time pass runs on count boundaries, so + `both` degrades to plain `count` rather than to a free pass. + + Uses CHURN_VALUES (count-UNSTABLE) precisely because an + implementation that degrades to an unconditional free pass would + also call an all-STABLE fixture STABLE here; only a fixture that + is UNSTABLE when fed without timestamps can tell the two apart. + """ + both_mode = SingleStabilityTracker(segmentation="both") + reference = SingleStabilityTracker() + for value in CHURN_VALUES: + both_mode.add_value(value) # no timestamp argument + reference.add_value(value) + assert reference.classify().type == "UNSTABLE" + assert both_mode.classify().type == "UNSTABLE" + + def test_reason_reports_both_mean_vectors(self): + """The note must carry the *actual* count and time mean vectors, not + just the words "count"/"time" -- and they must be distinct, which + catches a snapshot-ordering bug where the time pass's overwrite of + StabilityClassifier.segment_means leaks into the count half of the note + (see the comment at the count_means snapshot in _is_stable()). + + feed_churn is used because its count and time means genuinely + differ ([1.0, 1.0, 1.0, 0.0] vs [0.9375, 0.0, 0.0, 0.0]); it + also yields UNSTABLE, so this exercises the note on the branch + finding 1 wires it into. + """ + tracker = SingleStabilityTracker(segmentation="both") + feed_churn(tracker) + classification = tracker.classify() + assert classification.type == "UNSTABLE" + reason = classification.reason + assert "count [1.0, 1.0, 1.0, 0.0]" in reason + assert "time [0.9375, 0.0, 0.0, 0.0]" in reason + + def test_round_trip_preserves_both_mode(self): + tracker = SingleStabilityTracker(segmentation="both") + feed_churn(tracker) + restored = SingleStabilityTracker.from_state(tracker.to_state()) + assert restored.segmentation == "both" + assert restored.timestamps == tracker.timestamps + assert restored.classify().type == "UNSTABLE" + + def test_stability_note_is_not_persisted(self): + tracker = SingleStabilityTracker(segmentation="both") + feed_agreeing(tracker) + tracker.classify() + assert "_stability_note" not in tracker.to_state() + + def test_config_accepts_both_and_reaches_trackers(self): + configured = CharsetDetector(config=CharsetDetectorConfig()) + configured.config.stability_segmentation = "both" + rebuilt = CharsetDetector( + config=configured.config.to_dict(method_id="CharsetDetector") + ) + assert rebuilt.persistency.event_data_kwargs["segmentation"] == "both" + + def test_event_tracker_propagates_both(self): + event_tracker = EventStabilityTracker(segmentation="both") + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + single = event_tracker.get_data()["var1"] + assert single.segmentation == "both" + assert single.timestamps == [1.0, 2.0] From eeb7a1296fc2af9930b138b3142e74687a18e746 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 4 Aug 2026 18:31:15 +0200 Subject: [PATCH 7/7] fix: score empty duration segments as 0.0 instead of falling back to count An equal-duration cut that left a segment with no observations used to discard the time boundaries wholesale and re-cut the series by count. That guard existed because an empty segment produced a nan mean and `not nan >= thresh` is an unconditional pass -- a bursty variable could be called STABLE while churning. Score the empty segment 0.0 instead: nothing observed in a window means nothing changed in it. segment_means is now always finite, so the verdict no longer leans on nan comparison semantics, and time mode keeps its own boundaries in exactly the case the fallback used to swallow. The leniency the guard prevented is real and now reachable: a burst of churn followed by silence is time-STABLE with means [0.5, 0, 0, 0]. "both" is the answer to it -- the count pass keeps every segment populated and still calls that series UNSTABLE. Co-Authored-By: Claude Opus 5 --- docs/detectors.md | 11 ++- .../stability/stability_classifier.py | 27 ++++---- .../test_time_dependent_stability.py | 67 +++++++++---------- 3 files changed, 51 insertions(+), 54 deletions(-) diff --git a/docs/detectors.md b/docs/detectors.md index 71708318..c3615e5d 100644 --- a/docs/detectors.md +++ b/docs/detectors.md @@ -280,15 +280,20 @@ Time-aware segmentation is best-effort and never fails a run: **single** warning (once per detector, so a bad config cannot flood the log) and falls back to count-based segmentation. * If timestamps stop lining up with the recorded observations, or the observed time - span is zero, or the equal-duration cut would leave a segment with no observations - in it, the classifier silently falls back to count-based segmentation for that - variable. + span is zero, or they arrive out of order, the classifier silently falls back to + count-based segmentation for that variable. * Under `both`, any of the fallbacks above make the time pass reuse the count boundaries, so the mode degrades to plain `count` rather than to an unconditional pass. In every fallback case classification still runs and produces a result — only the segmentation rule changes back to the default. +A segment with no observations in it is *not* a fallback: it scores a mean of 0.0, +because nothing observed means nothing changed. Equal-duration cuts of a bursty +variable leave such segments routinely, so `time` on its own is lenient towards a +burst of churn followed by silence. Use `both` when that leniency matters — the +count pass keeps every segment populated. + ### Saving state (persist) diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index e4a0d153..c629547c 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -25,8 +25,10 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N non-decreasing, non-zero span), boundaries are equal-DURATION cuts of the observed time span, mapped back to indices. Falls back to equal-count on missing / mismatched / non-finite / out- - of-order timestamps, on zero span, and when the duration cut - would leave a segment empty. + of-order timestamps and on zero span. A duration cut that leaves + a segment empty is kept as-is: nothing observed in that window + means no changes in it, which ``is_stable`` scores as a mean of + 0.0. """ segment_size = total_len / self.n_segments count_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] @@ -56,17 +58,7 @@ def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = N boundaries = [int(np.searchsorted(timestamps, t, side="left")) for t in cuts] boundaries[0] = 0 boundaries[-1] = total_len - # An empty segment gets a nan mean, and `not nan >= thresh` is True -- - # a free pass. Under equal-duration cuts a bursty series can leave - # several segments empty and be called STABLE while churning; a - # 1-sample segment's quantized 0/1 mean can misfire the other way. - # Equal-count segmentation keeps every segment populated, so fall - # back to it whenever the duration cut would not. - # ponytail: this drops such series back to count mode wholesale; a - # future upgrade could redistribute or merge the empty segments and - # keep a (coarser) time-aware verdict. - if all(boundaries[i + 1] > boundaries[i] for i in range(self.n_segments)): - return boundaries + return boundaries return count_boundaries def is_stable( @@ -81,6 +73,11 @@ def is_stable( segments are equal-duration cuts of the time span instead of equal-count cuts of the series. See ``_segment_boundaries`` for the conditions under which time mode falls back to count mode. + + A segment with no observations in it scores a mean of 0.0 -- no + occurrences means no changes. Equal-duration cuts of a bursty + series leave such segments routinely; pair ``time`` with + ``count`` (segmentation ``both``) if that leniency matters. """ total_len = len(change_series) if total_len == 0: @@ -115,13 +112,13 @@ def is_stable( position = run_end self.segment_means = [ - segment_sums[i] / segment_counts[i] if segment_counts[i] > 0 else np.nan + segment_sums[i] / segment_counts[i] if segment_counts[i] > 0 else 0.0 for i in range(self.n_segments) ] else: self.segment_means = [ float(np.mean(change_series[segment_boundaries[i]:segment_boundaries[i + 1]])) - if segment_boundaries[i + 1] > segment_boundaries[i] else np.nan + if segment_boundaries[i + 1] > segment_boundaries[i] else 0.0 for i in range(self.n_segments) ] return all([not q >= thresh for q, thresh in zip(self.segment_means, self.segment_threshs)]) diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py index 328d2ba5..182e784f 100644 --- a/tests/test_persistency/test_time_dependent_stability.py +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -24,16 +24,16 @@ def make_classifier() -> StabilityClassifier: # tail is a burst compressed into ~0.04s at the end. # # The three change stamps are spaced so that each equal-duration quarter of the -# window holds at least one occurrence (boundaries [0, 1, 2, 3, 40]). An empty -# quarter would trip the empty-segment fallback in _segment_boundaries and drop -# the fixture back to count mode -- see -# TestClassifierTimeBoundaries.test_empty_time_segment_falls_back_to_count_mode. +# window holds exactly one occurrence (boundaries [0, 1, 2, 3, 40]), which is +# what puts a lone mean of 1.0 into a quarter whose threshold is 0.3. DIVERGENT_SERIES = [True, True, True] + [False] * 37 DIVERGENT_TIMES = [0.0, 30.0, 60.0] + [90.0 + 0.001 * i for i in range(37)] # Burst fixture: 20 fresh values, each immediately repeated once, inside 40 ms, # then a single further repeat an hour later. Equal-duration quarters put all 40 -# early occurrences in quarter 0 and leave quarters 1 and 2 with nothing in them. +# early occurrences in quarter 0 and leave quarters 1 and 2 with nothing in them, +# which scores them 0.0 -- so time mode calls this churning variable STABLE and +# only count mode (or `both`) catches it. BURSTY_VALUES = [f"v{i // 2}" for i in range(40)] + ["v19"] BURSTY_SERIES = [True, False] * 20 + [False] BURSTY_TIMES = [0.001 * i for i in range(40)] + [3600.0] @@ -116,31 +116,26 @@ def test_nan_timestamp_entry_falls_back_to_count_mode(self): assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() - def test_empty_time_segment_falls_back_to_count_mode(self): - """A segment with no occurrences has a nan mean, and `not nan >= - thresh` is True -- an unconditional pass. + def test_empty_time_segment_scores_zero(self): + """An empty segment means nothing was observed in that window, so its + mean is 0.0 -- the boundaries are kept, not discarded. - Raw equal-duration cuts of BURSTY_TIMES give boundaries [0, 40, - 40, 40, 41] and means [0.5, nan, nan, 0.0]: two free passes plus - segment 0 (whose 1.1 threshold is unreachable by a mean of - booleans), which is enough to call this churning variable - stable. Equal-count cuts keep every segment populated, so time - mode must defer to them here. + Equal-duration cuts of BURSTY_TIMES give boundaries [0, 40, 40, + 40, 41] and means [0.5, 0.0, 0.0, 0.0], which passes: time mode + alone is lenient on a burst followed by silence. Count mode + still sees the churn, so `both` catches it. """ clf_count = make_classifier() - expected = clf_count.is_stable(RLEList(BURSTY_SERIES)) - assert expected is False # count mode sees the churn + assert clf_count.is_stable(RLEList(BURSTY_SERIES)) is False clf_time = make_classifier() - assert clf_time.is_stable(RLEList(BURSTY_SERIES), timestamps=BURSTY_TIMES) is False - assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + assert clf_time.is_stable(RLEList(BURSTY_SERIES), timestamps=BURSTY_TIMES) is True + assert clf_time.get_last_segment_means() == [0.5, 0.0, 0.0, 0.0] - def test_empty_time_segment_fallback_on_plain_list_path(self): - clf_count = make_classifier() - expected = clf_count.is_stable(list(BURSTY_SERIES)) + def test_empty_time_segment_scores_zero_on_plain_list_path(self): clf_time = make_classifier() - assert clf_time.is_stable(list(BURSTY_SERIES), timestamps=BURSTY_TIMES) == expected - assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + assert clf_time.is_stable(list(BURSTY_SERIES), timestamps=BURSTY_TIMES) is True + assert clf_time.get_last_segment_means() == [0.5, 0.0, 0.0, 0.0] def test_out_of_order_timestamps_fall_back_to_count_mode(self): """np.searchsorted requires sorted input. @@ -272,20 +267,20 @@ def test_legacy_state_without_add_value_keys_loads(self): assert restored.detector_config is None assert restored.unique_set == {"hello"} - def test_bursty_series_is_unstable_in_both_modes(self): - """Time mode must not turn a churning variable STABLE. - - Empty equal-duration segments would pass unconditionally (means - [0.5, nan, nan, 0.0]) and hand this variable to auto-config - variable selection as a monitoring candidate. - """ - count_mode = SingleStabilityTracker() - time_mode = SingleStabilityTracker(segmentation="time") + def test_bursty_series_needs_the_count_pass(self): + """A burst followed by silence is time-STABLE (empty quarters score + 0.0) but count-UNSTABLE, so only `both` refuses to hand it to auto- + config variable selection as a monitoring candidate.""" + trackers = { + mode: SingleStabilityTracker(segmentation=mode) + for mode in ("count", "time", "both") + } for value, ts in zip(BURSTY_VALUES, BURSTY_TIMES): - count_mode.add_value(value) - time_mode.add_value(value, timestamp=ts) - assert count_mode.classify().type == "UNSTABLE" - assert time_mode.classify().type == "UNSTABLE" + for tracker in trackers.values(): + tracker.add_value(value, timestamp=ts) + assert trackers["count"].classify().type == "UNSTABLE" + assert trackers["time"].classify().type == "STABLE" + assert trackers["both"].classify().type == "UNSTABLE" class TestSegmentationPlumbing: