feat: Add more powerful drift windowing strategies, warmup and dynamic thresholds

benchmark/huffpost_kaggle/README.md

@@ -7,7 +7,6 @@ In this directory, you can find the files necessary to run experiments using the
 The goal is to predict the tag of news given headlines.
 The dataset contains more than 60k samples collected from 2012 to 2018.
 Titles belonging to the same year are grouped into the same CSV file and stored together.
-Each year is mapped to a year starting from 1/1/1970.
 There is a total of 42 categories/classes.
 
 > Note: The wild-time variant of the huffpost dataset has only 11 classes. This is due to the fact that

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/arxiv_datadrift.yaml

@@ -45,7 +45,9 @@ trigger:
   metrics:
     ev_mmd:
       id: EvidentlyModelDriftMetric
-      threshold: 0.7
+      decision_criterion:
+        id: ThresholdDecisionCriterion
+        threshold: 0.7
   aggregation_strategy:
     id: MajorityVote
 selection_strategy:

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/huffpost_datadrift.yaml

@@ -47,6 +47,8 @@ trigger:
     ev_mmd:
       id: AlibiDetectMmdDriftMetric
       num_permutations: 1000
+      decision_criterion:
+        id: HypothesisTestDecisionCriterion
   aggregation_strategy:
     id: MajorityVote
 selection_strategy:

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/yearbook_datadrift.yaml

@@ -48,7 +48,9 @@ trigger:
   metrics:
     ev_mmd:
       id: EvidentlyModelDriftMetric
-      threshold: 0.7
+      decision_criterion:
+        id: ThresholdDecisionCriterion
+        threshold: 0.7
   aggregation_strategy:
     id: MajorityVote
 selection_strategy:

modyn/config/schema/pipeline/trigger/drift/__init__.py

@@ -1,5 +1,6 @@
 from .aggregation import *  # noqa
 from .alibi_detect import *  # noqa
 from .config import *  # noqa
+from .detection_window import *  # noqa
 from .evidently import *  # noqa
 from .result import *  # noqa

modyn/config/schema/pipeline/trigger/drift/alibi_detect.py

@@ -1,10 +1,11 @@
 from typing import Annotated, Literal, Union
 
 from modyn.config.schema.base_model import ModynBaseModel
+from modyn.config.schema.pipeline.trigger.drift.metric import BaseMetric
 from pydantic import Field, model_validator
 
 
-class _AlibiDetectBaseDriftMetric(ModynBaseModel):
+class _AlibiDetectBaseDriftMetric(BaseMetric):
     p_val: float = Field(0.05, description="The p-value threshold for the drift detection.")
     x_ref_preprocessed: bool = Field(False)
 

modyn/config/schema/pipeline/trigger/drift/config.py

@@ -1,11 +1,9 @@
 from __future__ import annotations
 
-from functools import cached_property
 from typing import Annotated, ForwardRef, Literal, Optional, Union
 
 from modyn.config.schema.base_model import ModynBaseModel
-from modyn.const.regex import REGEX_TIME_UNIT
-from modyn.utils.utils import SECONDS_PER_UNIT
+from modyn.config.schema.pipeline.trigger.drift.detection_window import AmountWindowingStrategy, DriftWindowingStrategy
 from pydantic import Field
 
 from .aggregation import DriftAggregationStrategy, MajorityVoteDriftAggregationStrategy
@@ -23,50 +21,26 @@
 ]
 
 
-class AmountWindowingStrategy(ModynBaseModel):
-    id: Literal["AmountWindowingStrategy"] = Field("AmountWindowingStrategy")
-    amount: int = Field(1000, description="How many data points should fit in the window")
-
-
-class TimeWindowingStrategy(ModynBaseModel):
-    id: Literal["TimeWindowingStrategy"] = Field("TimeWindowingStrategy")
-    limit: str = Field(
-        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
-        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
-    )
-
-    @cached_property
-    def limit_seconds(self) -> int:
-        unit = str(self.limit)[-1:]
-        num = int(str(self.limit)[:-1])
-        return num * SECONDS_PER_UNIT[unit]
-
-
-DriftWindowingStrategy = Annotated[
-    Union[
-        AmountWindowingStrategy,
-        TimeWindowingStrategy,
-    ],
-    Field(discriminator="id"),
-]
-
-
 class DataDriftTriggerConfig(ModynBaseModel):
     id: Literal["DataDriftTrigger"] = Field("DataDriftTrigger")
 
     detection_interval: Optional[__TriggerConfig] = Field(  # type: ignore[valid-type]
         None, description="The Trigger policy to determine the interval at which drift detection is performed."
     )  # currently not used
-
     detection_interval_data_points: int = Field(
         1000, description="The number of samples in the interval after which drift detection is performed.", ge=1
     )
+
     windowing_strategy: DriftWindowingStrategy = Field(
         AmountWindowingStrategy(), description="Which windowing strategy to use for current and reference data"
     )
-
-    reset_current_window_on_trigger: bool = Field(
-        False, description="Whether the current window should be reset on trigger or rather be extended."
+    warmup_intervals: int | None = Field(
+        None,
+        description=(
+            "The number of intervals before starting to use the drift detection. Some "
+            "`DecisionCriteria` use this to calibrate the threshold. During the warmup, every interval will cause "
+            "a trigger."
+        ),
     )
 
     metrics: dict[str, DriftMetric] = Field(

modyn/config/schema/pipeline/trigger/drift/detection_window/__init__.py

@@ -0,0 +1,17 @@
+from typing import Annotated, Union
+
+from modyn.config.schema.pipeline.trigger.drift.detection_window.window import (
+    AmountWindowingStrategy,
+    TimeWindowingStrategy,
+)
+from pydantic import Field
+
+from .window import *  # noqa
+
+DriftWindowingStrategy = Annotated[
+    Union[
+        AmountWindowingStrategy,
+        TimeWindowingStrategy,
+    ],
+    Field(discriminator="id"),
+]

modyn/config/schema/pipeline/trigger/drift/detection_window/amount.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Annotated, Literal, Union
+
+from modyn.config.schema.base_model import ModynBaseModel
+from modyn.const.regex import REGEX_TIME_UNIT
+from modyn.utils.utils import SECONDS_PER_UNIT
+from pydantic import Field
+
+
+class _BaseWindowingStrategy(ModynBaseModel):
+    allow_overlap: bool = Field(
+        False,
+        description=(
+            "Whether the windows are allowed to overlap. This is useful for time-based windows."
+            "If set to False, the current window will be reset after each trigger."
+        ),
+    )
+
+
+class AmountWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["AmountWindowingStrategy"] = Field("AmountWindowingStrategy")
+    amount_ref: int = Field(1000, description="How many data points should fit in the reference window", ge=1)
+    amount_cur: int = Field(1000, description="How many data points should fit in the current window", ge=1)
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return self.amount_cur
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return self.amount_ref
+
+
+class TimeWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["TimeWindowingStrategy"] = Field("TimeWindowingStrategy")
+    limit_ref: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+    limit_cur: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+
+    @cached_property
+    def limit_seconds_ref(self) -> int:
+        unit = str(self.limit_ref)[-1:]
+        num = int(str(self.limit_ref)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @cached_property
+    def limit_seconds_cur(self) -> int:
+        unit = str(self.limit_cur)[-1:]
+        num = int(str(self.limit_cur)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return None
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return None
+
+
+DriftWindowingStrategy = Annotated[
+    Union[
+        AmountWindowingStrategy,
+        TimeWindowingStrategy,
+    ],
+    Field(discriminator="id"),
+]

modyn/config/schema/pipeline/trigger/drift/detection_window/timestamp.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Literal
+
+from modyn.const.regex import REGEX_TIME_UNIT
+from modyn.utils.utils import SECONDS_PER_UNIT
+from pydantic import Field
+
+from .window import _BaseWindowingStrategy
+
+
+class TimeWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["TimeWindowingStrategy"] = Field("TimeWindowingStrategy")
+    limit_ref: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+    limit_cur: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+
+    @cached_property
+    def limit_seconds_ref(self) -> int:
+        unit = str(self.limit_ref)[-1:]
+        num = int(str(self.limit_ref)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @cached_property
+    def limit_seconds_cur(self) -> int:
+        unit = str(self.limit_cur)[-1:]
+        num = int(str(self.limit_cur)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return None
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return None

modyn/config/schema/pipeline/trigger/drift/detection_window/window.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Literal
+
+from modyn.config.schema.base_model import ModynBaseModel
+from modyn.const.regex import REGEX_TIME_UNIT
+from modyn.utils.utils import SECONDS_PER_UNIT
+from pydantic import Field
+
+
+class _BaseWindowingStrategy(ModynBaseModel):
+    allow_overlap: bool = Field(
+        False,
+        description=(
+            "Whether the windows are allowed to overlap. This is useful for time-based windows."
+            "If set to False, the current window will be reset after each trigger."
+        ),
+    )
+
+
+class AmountWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["AmountWindowingStrategy"] = Field("AmountWindowingStrategy")
+    amount_ref: int = Field(1000, description="How many data points should fit in the reference window", ge=1)
+    amount_cur: int = Field(1000, description="How many data points should fit in the current window", ge=1)
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return self.amount_cur
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return self.amount_ref
+
+
+class TimeWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["TimeWindowingStrategy"] = Field("TimeWindowingStrategy")
+    limit_ref: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+    limit_cur: str = Field(
+        description="Window size as an integer followed by a time unit: s, m, h, d, w, y",
+        pattern=rf"^\d+{REGEX_TIME_UNIT}$",
+    )
+
+    @cached_property
+    def limit_seconds_ref(self) -> int:
+        unit = str(self.limit_ref)[-1:]
+        num = int(str(self.limit_ref)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @cached_property
+    def limit_seconds_cur(self) -> int:
+        unit = str(self.limit_cur)[-1:]
+        num = int(str(self.limit_cur)[:-1])
+        return num * SECONDS_PER_UNIT[unit]
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return None
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return None

modyn/config/schema/pipeline/trigger/drift/evidently.py

@@ -1,10 +1,10 @@
 from typing import Annotated, Literal, Union
 
-from modyn.config.schema.base_model import ModynBaseModel
+from modyn.config.schema.pipeline.trigger.drift.metric import BaseMetric
 from pydantic import Field
 
 
-class _EvidentlyBaseDriftMetric(ModynBaseModel):
+class _EvidentlyBaseDriftMetric(BaseMetric):
     num_pca_component: int | None = Field(None)
 
 

modyn/config/schema/pipeline/trigger/drift/metric.py

@@ -0,0 +1,47 @@
+from typing import Annotated, Literal, Union
+
+from modyn.config.schema.base_model import ModynBaseModel
+from pydantic import Field
+
+
+class HypothesisTestDecisionCriterion(ModynBaseModel):
+    id: Literal["HypothesisTestDecisionCriterion"] = "HypothesisTestDecisionCriterion"
+
+    @property
+    def needs_calibration(self) -> bool:
+        return False
+
+
+class ThresholdDecisionCriterion(ModynBaseModel):
+    id: Literal["ThresholdDecisionCriterion"] = "ThresholdDecisionCriterion"
+    threshold: float
+
+    @property
+    def needs_calibration(self) -> bool:
+        return False
+
+
+class DynamicThresholdCriterion(ModynBaseModel):
+    id: Literal["DynamicThresholdCriterion"] = "DynamicThresholdCriterion"
+    window_size: int
+    percentile_threshold: float = Field(
+        0.05, description="The percentile that a threshold has to be in to trigger a drift event."
+    )
+
+    @property
+    def needs_calibration(self) -> bool:
+        return True
+
+
+DecisionCriterion = Annotated[
+    Union[
+        HypothesisTestDecisionCriterion,
+        ThresholdDecisionCriterion,
+        DynamicThresholdCriterion,
+    ],
+    Field(discriminator="id"),
+]
+
+
+class BaseMetric(ModynBaseModel):
+    decision_criterion: DecisionCriterion