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

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

@@ -3,8 +3,10 @@
 from modyn.config.schema.base_model import ModynBaseModel
 from pydantic import Field, model_validator
 
+from modyn.config.schema.pipeline.trigger.drift.metric import BaseMetric
 
-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)
 
@@ -51,7 +53,7 @@ def validate_threshold_permutations(self) -> "AlibiDetectMmdDriftMetric":
         if self.threshold is not None and self.num_permutations is not None:
             raise ValueError(
                 "threshold and num_permutations are mutually exclusive."
-                + "Please specify whether you want to use hypothosis testing "
+                + "Please specify whether you want to use hypothesis testing "
                 + "or threshold comparison for making drift decisions."
             )
 

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

@@ -1,13 +1,12 @@
 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 pydantic import Field
 
+from modyn.config.schema.base_model import ModynBaseModel
+from modyn.config.schema.pipeline.trigger.drift.detection_window import AmountWindowingStrategy, DriftWindowingStrategy
+
 from .aggregation import DriftAggregationStrategy, MajorityVoteDriftAggregationStrategy
 from .alibi_detect import AlibiDetectDriftMetric
 from .evidently import EvidentlyDriftMetric
@@ -23,52 +22,31 @@
 ]
 
 
-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"
     )
+    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."
+        ),
+    )
 
     reset_current_window_on_trigger: bool = Field(
         False, description="Whether the current window should be reset on trigger or rather be extended."
     )
-
     metrics: dict[str, DriftMetric] = Field(
         min_length=1, description="The metrics used for drift detection keyed by a reference."
     )

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

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Annotated, Literal, Union
+
+from pydantic import Field
+
+from modyn.config.schema.base_model import ModynBaseModel
+from modyn.const.regex import REGEX_TIME_UNIT
+from modyn.utils.utils import SECONDS_PER_UNIT
+
+
+class _BaseWindowingStrategy(ModynBaseModel):
+    allow_overlap: bool = Field(
+        False, description="Whether the windows are allowed to overlap. This is useful for time-based windows."
+    )
+
+
+class AmountWindowingStrategy(_BaseWindowingStrategy):
+    id: Literal["AmountWindowingStrategy"] = Field("AmountWindowingStrategy")
+    amount_ref: int = Field(1000, description="How many data points should fit in the reference window")
+    amount_cur: int = Field(1000, description="How many data points should fit in the current window")
+
+    @property
+    def current_buffer_size(self) -> int | None:
+        return self.amount_ref
+
+    @property
+    def reference_buffer_size(self) -> int | None:
+        return self.amount_cur
+
+
+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/evidently.py

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

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

@@ -0,0 +1,48 @@
+from typing import Annotated, Literal, Union
+
+from pydantic import Field
+
+from modyn.config.schema.base_model import ModynBaseModel
+
+
+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