Feat/time dependent config engine

docs/detectors.md

@@ -222,6 +222,79 @@ def set_configuration(self):
 When `auto_config` is `False`, steps 1 and 2 are skipped entirely.
 
 
+### 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
+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 `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`.
+
+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:
+      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 |
+|---|---|---|---|
+| `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
+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-aware segmentation is best-effort and never fails a run:
+
+* 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.
+* If timestamps stop lining up with the recorded observations, or the observed time
+  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)
 
 Detectors can persist their training state to disk (or cloud storage) so it

src/detectmatelibrary/common/variable_detector.py

@@ -13,18 +13,28 @@
 )
 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
 
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Literal, Optional, cast
 from typing_extensions import override
 
 
 class VariableDetectorConfig(CoreDetectorConfig):
     use_stable_vars: bool = True
     use_static_vars: bool = True
 
+    # 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.
+    stability_segmentation: Literal["count", "time", "both"] = "count"
+    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 +54,32 @@ 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_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._auto_conf_kwargs(),
+            event_data_kwargs=self._with_segmentation(self._auto_conf_kwargs()),
         )
         self._register_persistency(self.persistency)
 
+    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 self.config.stability_segmentation == "count":
+            return kwargs
+        return {**(kwargs or {}), "segmentation": self.config.stability_segmentation}
+
     # ---- construction hooks -------------------------------------------------
 
     def _event_data_class(self) -> type:
@@ -75,6 +100,43 @@ 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 self.config.stability_segmentation == "count":
+            return None
+        if not self.config.timestamp_variable:
+            # 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(
+                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)
+        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 +174,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 +238,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,13 +264,19 @@ def set_configuration(self) -> None:
             if selected:
                 variables[event_id] = selected
         old_persist = self.config.persist
+        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(
             variable_selection=variables,
             detector_name=self.name,
             method_type=self.config.method_type,
         )
         self.config = type(self.config).from_dict(config_dict, self.name)
         self.config.persist = old_persist
+        self.config.stability_segmentation = old_segmentation
+        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(

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_segmentation(
+                {"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
+        segmentation_fields = {
+            "stability_segmentation": self.config.stability_segmentation,
+            "timestamp_variable": self.config.timestamp_variable,
+            "timestamp_format": self.config.timestamp_format,
+        }
+
+        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
+            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 segmentation_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_segmentation_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_segmentation_fields()
         events = self.config.events
         if isinstance(events, EventsConfig) and not events.events:
             logger.warning(

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: ...

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)

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:

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."""
@@ -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.
+        ``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.

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."""

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