You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2025/03/10 21:11:34 UTC
(beam) branch master updated: [AnomalyDetection] Add detectors and some necessary univariate trackers. (#34232)
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 9015b2b180b [AnomalyDetection] Add detectors and some necessary univariate trackers. (#34232)
9015b2b180b is described below
commit 9015b2b180b25c816878889111f9f2c63e80853f
Author: Shunping Huang <sh...@google.com>
AuthorDate: Mon Mar 10 17:11:26 2025 -0400
[AnomalyDetection] Add detectors and some necessary univariate trackers. (#34232)
* Add detectors and some necessary univariate trackers.
* Make score_one return float or None.
* Raise exception when more than one variable used in univariate detectors.
* Fix lints.
---
sdks/python/apache_beam/ml/anomaly/base.py | 5 +-
.../apache_beam/ml/anomaly/detectors/__init__.py | 20 ++++
.../python/apache_beam/ml/anomaly/detectors/iqr.py | 129 +++++++++++++++++++++
.../apache_beam/ml/anomaly/detectors/iqr_test.py | 52 +++++++++
.../ml/anomaly/detectors/robust_zscore.py | 116 ++++++++++++++++++
.../ml/anomaly/detectors/robust_zscore_test.py | 51 ++++++++
.../apache_beam/ml/anomaly/detectors/zscore.py | 126 ++++++++++++++++++++
.../ml/anomaly/detectors/zscore_test.py | 83 +++++++++++++
.../apache_beam/ml/anomaly/univariate/base.py | 2 +
.../apache_beam/ml/anomaly/univariate/mad.py | 87 ++++++++++++++
.../apache_beam/ml/anomaly/univariate/mad_test.py | 51 ++++++++
.../apache_beam/ml/anomaly/univariate/median.py | 67 +++++++++++
.../ml/anomaly/univariate/median_test.py | 72 ++++++++++++
.../apache_beam/ml/anomaly/univariate/quantile.py | 58 +++++++--
14 files changed, 907 insertions(+), 12 deletions(-)
diff --git a/sdks/python/apache_beam/ml/anomaly/base.py b/sdks/python/apache_beam/ml/anomaly/base.py
index 4242decf97e..5886c727811 100644
--- a/sdks/python/apache_beam/ml/anomaly/base.py
+++ b/sdks/python/apache_beam/ml/anomaly/base.py
@@ -165,14 +165,15 @@ class AnomalyDetector(abc.ABC):
raise NotImplementedError
@abc.abstractmethod
- def score_one(self, x: beam.Row) -> float:
+ def score_one(self, x: beam.Row) -> Optional[float]:
"""Scores a single data instance for anomalies.
Args:
x: A `beam.Row` representing the data instance.
Returns:
- The outlier score as a float.
+ The outlier score as a float. None if an exception occurs during scoring,
+ and NaN if the model is not ready.
"""
raise NotImplementedError
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/__init__.py b/sdks/python/apache_beam/ml/anomaly/detectors/__init__.py
new file mode 100644
index 00000000000..f3268755cf9
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/__init__.py
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from apache_beam.ml.anomaly.detectors.zscore import ZScore
+from apache_beam.ml.anomaly.detectors.robust_zscore import RobustZScore
+from apache_beam.ml.anomaly.detectors.iqr import IQR
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/iqr.py b/sdks/python/apache_beam/ml/anomaly/detectors/iqr.py
new file mode 100644
index 00000000000..d8d92c176f8
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/iqr.py
@@ -0,0 +1,129 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import math
+from typing import Optional
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.base import AnomalyDetector
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.thresholds import FixedThreshold
+from apache_beam.ml.anomaly.univariate.base import EPSILON
+from apache_beam.ml.anomaly.univariate.quantile import BufferedSlidingQuantileTracker # pylint: disable=line-too-long
+from apache_beam.ml.anomaly.univariate.quantile import QuantileTracker
+from apache_beam.ml.anomaly.univariate.quantile import SecondaryBufferedQuantileTracker # pylint: disable=line-too-long
+
+DEFAULT_WINDOW_SIZE = 1000
+
+
+@specifiable
+class IQR(AnomalyDetector):
+ """Interquartile Range (IQR) anomaly detector.
+
+ This class implements an anomaly detection algorithm based on the
+ Interquartile Range (IQR) [#]_ . It calculates the IQR using quantile trackers
+ for Q1 (25th percentile) and Q3 (75th percentile) and scores data points based
+ on their deviation from these quartiles.
+
+ The score is calculated as follows:
+
+ * If a data point is above Q3, the score is (value - Q3) / IQR.
+ * If a data point is below Q1, the score is (Q1 - value) / IQR.
+ * If a data point is within the IQR (Q1 <= value <= Q3), the score is 0.
+ Initializes the IQR anomaly detector.
+
+ Args:
+ q1_tracker: Optional QuantileTracker for Q1 (25th percentile). If None, a
+ BufferedSlidingQuantileTracker with a default window size is used.
+ q3_tracker: Optional QuantileTracker for Q3 (75th percentile). If None, a
+ SecondaryBufferedQuantileTracker based on q1_tracker is used.
+ threshold_criterion: Optional ThresholdFn to apply on the score. Defaults
+ to `FixedThreshold(1.5)` since outliers are commonly defined as data
+ points that fall below Q1 - 1.5 IQR or above Q3 + 1.5 IQR.
+ **kwargs: Additional keyword arguments.
+
+ .. [#] https://en.wikipedia.org/wiki/Interquartile_range
+ """
+ def __init__(
+ self,
+ q1_tracker: Optional[QuantileTracker] = None,
+ q3_tracker: Optional[QuantileTracker] = None,
+ **kwargs):
+ if "threshold_criterion" not in kwargs:
+ kwargs["threshold_criterion"] = FixedThreshold(1.5)
+ super().__init__(**kwargs)
+
+ self._q1_tracker = q1_tracker or \
+ BufferedSlidingQuantileTracker(DEFAULT_WINDOW_SIZE, 0.25)
+ assert self._q1_tracker._q == 0.25, \
+ "q1_tracker must be initialized with q = 0.25"
+
+ self._q3_tracker = q3_tracker or \
+ SecondaryBufferedQuantileTracker(self._q1_tracker, 0.75)
+ assert self._q3_tracker._q == 0.75, \
+ "q3_tracker must be initialized with q = 0.75"
+
+ def learn_one(self, x: beam.Row) -> None:
+ """Updates the quantile trackers with a new data point.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "IQR.learn_one expected univariate input, but got %s", str(x))
+
+ v = next(iter(x))
+ self._q1_tracker.push(v)
+ self._q3_tracker.push(v)
+
+ def score_one(self, x: beam.Row) -> Optional[float]:
+ """Scores a data point based on its deviation from the IQR.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+
+ Returns:
+ float | None: The anomaly score.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "IQR.score_one expected univariate input, but got %s", str(x))
+
+ v = next(iter(x))
+ if v is None or math.isnan(v):
+ return None
+
+ q1 = self._q1_tracker.get()
+ q3 = self._q3_tracker.get()
+
+ # not enough data points to compute median or median absolute deviation
+ if math.isnan(q1) or math.isnan(q3):
+ return float('NaN')
+
+ iqr = q3 - q1
+ if abs(iqr) < EPSILON:
+ return 0.0
+
+ if v > q3:
+ return (v - q3) / iqr
+
+ if v < q1:
+ return (q1 - v) / iqr
+
+ # q1 <= v <= q3, normal points
+ return 0
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/iqr_test.py b/sdks/python/apache_beam/ml/anomaly/detectors/iqr_test.py
new file mode 100644
index 00000000000..c2d175fdd02
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/iqr_test.py
@@ -0,0 +1,52 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import math
+import unittest
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.detectors.iqr import IQR
+
+
+class IQRTest(unittest.TestCase):
+ input = [
+ beam.Row(x=1),
+ beam.Row(x=1),
+ beam.Row(x=5),
+ beam.Row(x=9),
+ beam.Row(x=20),
+ beam.Row(x=10),
+ beam.Row(x=1)
+ ]
+
+ def test_with_default_trackers(self):
+ iqr = IQR()
+
+ scores = []
+ for row in IQRTest.input:
+ scores.append(iqr.score_one(row))
+ iqr.learn_one(row)
+
+ self.assertTrue(math.isnan(scores[0]))
+ self.assertEqual(
+ scores[1:], [0.0, 0.0, 3.0, 2.8, 0.125, 0.12903225806451613])
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore.py b/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore.py
new file mode 100644
index 00000000000..864aa9f8fde
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore.py
@@ -0,0 +1,116 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import math
+from typing import Optional
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.base import AnomalyDetector
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.thresholds import FixedThreshold
+from apache_beam.ml.anomaly.univariate.base import EPSILON
+from apache_beam.ml.anomaly.univariate.mad import MadTracker
+
+
+# pylint: disable=line-too-long
+@specifiable
+class RobustZScore(AnomalyDetector):
+ """Robust Z-Score anomaly detector.
+
+ This class implements an detection algorithm based on Robust Z-Score (also
+ known as Modified Z-Score), which is a robust alternative to the traditional
+ Z-score [#]_. It uses the median and Median Absolute Deviation (MAD) to
+ compute a score that is less sensitive to outliers.
+
+ The score is calculated as: `|0.6745 * (value - median) / MAD|`
+
+ Important:
+ In the streaming setting, we use the online version of median and MAD in the
+ calculation. Therefore, the score computed here does not exactly match its
+ batch counterpart.
+
+ This implementation is adapted from the implementation within PySAD [#]_:
+ https://github.com/selimfirat/pysad/blob/master/pysad/models/median_absolute_deviation.py
+
+ The batch version can be seen at PyOD [#]_:
+ https://github.com/yzhao062/pyod/blob/master/pyod/models/mad.py
+
+
+ Args:
+ mad_tracker: Optional `MadTracker` instance. If None, a default `MadTracker`
+ is created.
+ threshold_criterion: threshold_criterion: Optional `ThresholdFn` to apply on
+ the score. Defaults to `FixedThreshold(3)` due to the commonly used
+ 3-sigma rule.
+ **kwargs: Additional keyword arguments.
+
+ .. [#] Hoaglin, David C.. (2013). Volume 16: How to Detect and Handle Outliers.
+ .. [#] Yilmaz, Selim & Kozat, Suleyman. (2020). PySAD: A Streaming Anomaly Detection Framework in Python. 10.48550/arXiv.2009.02572.
+ .. [#] Zhao, Y., Nasrullah, Z. and Li, Z.. (2019). PyOD: A Python Toolbox for Scalable Outlier Detection. Journal of machine learning research (JMLR), 20(96), pp.1-7.
+ """
+ # pylint: enable=line-too-long
+ SCALE_FACTOR = 0.6745
+
+ def __init__(self, mad_tracker: Optional[MadTracker] = None, **kwargs):
+ if "threshold_criterion" not in kwargs:
+ kwargs["threshold_criterion"] = FixedThreshold(3)
+ super().__init__(**kwargs)
+ self._mad_tracker = mad_tracker or MadTracker()
+
+ def learn_one(self, x: beam.Row) -> None:
+ """Updates the `MadTracker` with a new data point.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "RobustZScore.learn_one expected univariate input, but got %s",
+ str(x))
+
+ v = next(iter(x))
+ self._mad_tracker.push(v)
+
+ def score_one(self, x: beam.Row) -> Optional[float]:
+ """Scores a data point using the Robust Z-Score.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+
+ Returns:
+ float | None: The Robust Z-Score.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "RobustZScore.score_one expected univariate input, but got %s",
+ str(x))
+
+ v = next(iter(x))
+ if v is None or math.isnan(v):
+ return None
+
+ median = self._mad_tracker.get_median()
+ mad = self._mad_tracker.get()
+
+ # not enough data points to compute median or median absolute deviation
+ if math.isnan(mad) or math.isnan(median):
+ return float('NaN')
+
+ if abs(mad) < EPSILON:
+ return 0.0
+
+ return abs(RobustZScore.SCALE_FACTOR * (v - median) / mad)
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore_test.py b/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore_test.py
new file mode 100644
index 00000000000..be217260290
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/robust_zscore_test.py
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import math
+import unittest
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.detectors.robust_zscore import RobustZScore
+
+
+class RobustZScoreTest(unittest.TestCase):
+ input = [
+ beam.Row(x=1),
+ beam.Row(x=1),
+ beam.Row(x=5),
+ beam.Row(x=7),
+ beam.Row(x=20),
+ beam.Row(x=6),
+ beam.Row(x=1)
+ ]
+
+ def test_with_default_trackers(self):
+ zscore = RobustZScore()
+
+ scores = []
+ for row in RobustZScoreTest.input:
+ scores.append(zscore.score_one(row))
+ zscore.learn_one(row)
+
+ self.assertTrue(math.isnan(scores[0]))
+ self.assertEqual(scores[1:], [0.0, 0.0, 0.0, 5.73325, 0.168625, 1.349])
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/zscore.py b/sdks/python/apache_beam/ml/anomaly/detectors/zscore.py
new file mode 100644
index 00000000000..3f141f81ec4
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/zscore.py
@@ -0,0 +1,126 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import math
+from typing import Optional
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.base import AnomalyDetector
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.thresholds import FixedThreshold
+from apache_beam.ml.anomaly.univariate.base import EPSILON
+from apache_beam.ml.anomaly.univariate.mean import IncSlidingMeanTracker
+from apache_beam.ml.anomaly.univariate.mean import MeanTracker
+from apache_beam.ml.anomaly.univariate.stdev import IncSlidingStdevTracker
+from apache_beam.ml.anomaly.univariate.stdev import StdevTracker
+
+DEFAULT_WINDOW_SIZE = 1000
+
+
+# pylint: disable=line-too-long
+@specifiable
+class ZScore(AnomalyDetector):
+ """Z-Score anomaly detector.
+
+ This class implements an anomaly detection algorithm based on Z-Score (also
+ known as Standard Score [#]_ ), which measures how many standard deviations a
+ data point is from the mean.
+
+ The score is calculated as: `| (value - mean) / stdev |`
+
+ Important:
+ In the streaming setting, we use the online version of mean and standard
+ deviation in the calculation.
+
+ This implementation is adapted from the implementations within PySAD [#]_ and
+ River [#]_:
+
+ * https://github.com/selimfirat/pysad/blob/master/pysad/models/standard_absolute_deviation.py
+ * https://github.com/online-ml/river/blob/main/river/anomaly/sad.py
+
+ Args:
+ sub_stat_tracker: Optional `MeanTracker` instance. If None, an
+ `IncSlidingMeanTracker` with a default window size 1000 is created.
+ stdev_tracker: Optional `StdevTracker` instance. If None, an
+ `IncSlidingStdevTracker` with a default window size 1000 is created.
+ threshold_criterion: threshold_criterion: Optional `ThresholdFn` to apply on
+ the score. Defaults to `FixedThreshold(3)` due to the commonly used
+ 3-sigma rule.
+ **kwargs: Additional keyword arguments.
+
+ .. [#] https://en.wikipedia.org/wiki/Standard_score
+ .. [#] Yilmaz, Selim & Kozat, Suleyman. (2020). PySAD: A Streaming Anomaly Detection Framework in Python. 10.48550/arXiv.2009.02572.
+ .. [#] Jacob Montiel, Max Halford, Saulo Martiello Mastelini, Geoffrey Bolmier, Raphaël Sourty, et al.. (2021). River: machine learning for streaming data in Python. Journal of Machine Learning Research, 2021, 22, pp.1-8.
+ """
+
+ # pylint: enable=line-too-long
+ def __init__(
+ self,
+ sub_stat_tracker: Optional[MeanTracker] = None,
+ stdev_tracker: Optional[StdevTracker] = None,
+ **kwargs):
+ if "threshold_criterion" not in kwargs:
+ kwargs["threshold_criterion"] = FixedThreshold(3)
+ super().__init__(**kwargs)
+
+ self._sub_stat_tracker = sub_stat_tracker or IncSlidingMeanTracker(
+ DEFAULT_WINDOW_SIZE)
+ self._stdev_tracker = stdev_tracker or IncSlidingStdevTracker(
+ DEFAULT_WINDOW_SIZE)
+
+ def learn_one(self, x: beam.Row) -> None:
+ """Updates the mean and standard deviation trackers with a new data point.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "ZScore.learn_one expected univariate input, but got %s", str(x))
+
+ v = next(iter(x))
+ self._stdev_tracker.push(v)
+ self._sub_stat_tracker.push(v)
+
+ def score_one(self, x: beam.Row) -> Optional[float]:
+ """Scores a data point using the Z-Score.
+
+ Args:
+ x: A `beam.Row` containing a single numerical value.
+
+ Returns:
+ float | None: The Z-Score.
+ """
+ if len(x.__dict__) != 1:
+ raise ValueError(
+ "ZScore.score_one expected univariate input, but got %s", str(x))
+
+ v = next(iter(x))
+ if v is None or math.isnan(v):
+ return None
+
+ sub_stat = self._sub_stat_tracker.get()
+ stdev = self._stdev_tracker.get()
+
+ # not enough data points to compute sub_stat or standard deviation
+ if math.isnan(stdev) or math.isnan(sub_stat):
+ return float('NaN')
+
+ if abs(stdev) < EPSILON:
+ return 0.0
+
+ return abs((v - sub_stat) / stdev)
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/zscore_test.py b/sdks/python/apache_beam/ml/anomaly/detectors/zscore_test.py
new file mode 100644
index 00000000000..830d57b25c4
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/zscore_test.py
@@ -0,0 +1,83 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import math
+import unittest
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.detectors.zscore import ZScore
+from apache_beam.ml.anomaly.univariate.mean import IncSlidingMeanTracker
+from apache_beam.ml.anomaly.univariate.stdev import IncSlidingStdevTracker
+
+
+class ZScoreTest(unittest.TestCase):
+ input = [
+ beam.Row(x=1),
+ beam.Row(x=1),
+ beam.Row(x=5),
+ beam.Row(x=9),
+ beam.Row(x=20),
+ beam.Row(x=10),
+ beam.Row(x=1)
+ ]
+
+ def test_with_default_trackers(self):
+ zscore = ZScore()
+
+ scores = []
+ for row in ZScoreTest.input:
+ scores.append(zscore.score_one(row))
+ zscore.learn_one(row)
+
+ self.assertTrue(math.isnan(scores[0]))
+ self.assertTrue(math.isnan(scores[1]))
+ self.assertEqual(
+ scores[2:],
+ [
+ 0.0,
+ 2.8867513459481287,
+ 4.177863742936748,
+ 0.35502819053868157,
+ 0.932910509720565
+ ])
+
+ def test_with_custom_mean_tracker(self):
+ zscore = ZScore(
+ sub_stat_tracker=IncSlidingMeanTracker(3),
+ stdev_tracker=IncSlidingStdevTracker(3))
+
+ scores = []
+ for row in ZScoreTest.input:
+ scores.append(zscore.score_one(row))
+ zscore.learn_one(row)
+
+ self.assertTrue(math.isnan(scores[0]))
+ self.assertTrue(math.isnan(scores[1]))
+ self.assertEqual(
+ scores[2:], [
+ 0.0,
+ 2.8867513459481287,
+ 3.75,
+ 0.17165643016914964,
+ 1.9727878476642864
+ ])
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/base.py b/sdks/python/apache_beam/ml/anomaly/univariate/base.py
index b0eb2aba1e6..d44abaa82a2 100644
--- a/sdks/python/apache_beam/ml/anomaly/univariate/base.py
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/base.py
@@ -19,6 +19,8 @@ import abc
from collections import deque
from enum import Enum
+EPSILON = 1e-9
+
class BaseTracker(abc.ABC):
"""Abstract base class for all univariate trackers."""
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/mad.py b/sdks/python/apache_beam/ml/anomaly/univariate/mad.py
new file mode 100644
index 00000000000..62e97b3f941
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/mad.py
@@ -0,0 +1,87 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Trackers for calculating median absolute deviation in windowed fashion."""
+
+from typing import Optional
+
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.univariate.base import BaseTracker
+from apache_beam.ml.anomaly.univariate.median import MedianTracker
+
+
+@specifiable
+class MadTracker(BaseTracker):
+ """Tracks the Median Absolute Deviation (MAD) of a stream of values.
+
+ This class calculates the MAD, a robust measure of statistical dispersion, in
+ an online setting.
+
+ Similar functionality is available in the River library:
+ https://github.com/online-ml/river/blob/main/river/stats/mad.py
+
+ Important:
+ This online version of MAD that does not exactly match its batch
+ counterpart. In a streaming data context, where the true median is initially
+ unknown, we employ an iterative estimation process. For each incoming data
+ point, we first update the estimated median, and then calculate the absolute
+ difference between the data point and this updated median. To maintain
+ computational efficiency, previously calculated absolute differences are not
+ recalculated with each subsequent median update.
+
+ Args:
+ median_tracker: An optional `MedianTracker` instance for tracking the
+ median of the input values. If None, a default `MedianTracker` is
+ created.
+ diff_median_tracker: An optional `MedianTracker` instance for tracking
+ the median of the absolute deviations from the median. If None, a
+ default `MedianTracker` is created.
+ """
+ def __init__(
+ self,
+ median_tracker: Optional[MedianTracker] = None,
+ diff_median_tracker: Optional[MedianTracker] = None):
+ self._median_tracker = median_tracker or MedianTracker()
+ self._diff_median_tracker = diff_median_tracker or MedianTracker()
+
+ def push(self, x):
+ """Adds a new value to the tracker and updates the MAD.
+
+ Args:
+ x: The value to be added to the tracked stream.
+ """
+ self._median_tracker.push(x)
+ median = self._median_tracker.get()
+ self._diff_median_tracker.push(abs(x - median))
+
+ def get(self):
+ """Retrieves the current MAD value.
+
+ Returns:
+ float: The MAD of the values within the defined window. Returns `NaN` if
+ the window is empty.
+ """
+ return self._diff_median_tracker.get()
+
+ def get_median(self):
+ """Retrieves the current median value.
+
+ Returns:
+ float: The median of the values within the defined window. Returns `NaN`
+ if the window is empty.
+ """
+ return self._median_tracker.get()
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/mad_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/mad_test.py
new file mode 100644
index 00000000000..2d0114af360
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/mad_test.py
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import math
+import unittest
+
+from apache_beam.ml.anomaly.univariate.mad import MadTracker
+
+
+class MadTest(unittest.TestCase):
+ def test_default_tracker(self):
+ t = MadTracker()
+ self.assertTrue(math.isnan(t.get()))
+
+ t.push(4.0) # median=4, abs_dev=[0]
+ self.assertEqual(t.get(), 0)
+
+ t.push(2.0) # median=3, abs_dev=[0, 1]
+ self.assertEqual(t.get(), 0.5)
+
+ t.push(5.0) # median=4, abs_dev=[0, 1, 1]
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(3.0) # median=3.5 abs_dev=[0, 0.5, 1, 1]
+ self.assertEqual(t.get(), 0.75)
+
+ t.push(0.0) # median=3 abs_dev=[0, 0.5, 1, 1, 3]
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(4.0) # median=3.5 abs_dev[0, 0.5, 0.5, 1, 1, 3]
+ self.assertEqual(t.get(), 0.75)
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/median.py b/sdks/python/apache_beam/ml/anomaly/univariate/median.py
new file mode 100644
index 00000000000..1d190b70e5b
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/median.py
@@ -0,0 +1,67 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Trackers for calculating median in windowed fashion."""
+
+from typing import Optional
+
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.univariate.base import BaseTracker
+from apache_beam.ml.anomaly.univariate.quantile import BufferedSlidingQuantileTracker # pylint: disable=line-too-long
+from apache_beam.ml.anomaly.univariate.quantile import QuantileTracker
+
+DEFAULT_WINDOW_SIZE = 1000
+
+
+@specifiable
+class MedianTracker(BaseTracker):
+ """Tracks the median of a stream of values using a quantile tracker.
+
+ This wrapper class encapsulates a `QuantileTracker` configured specifically
+ for the 0.5 quantile (median).
+
+ Args:
+ quantile_tracker: An optional `QuantileTracker` instance. If not provided,
+ a `BufferedSlidingQuantileTracker` with a default window size 1000 and
+ q=0.5 is created.
+
+ Raises:
+ AssertionError: If the provided quantile_tracker is not initialized with
+ q=0.5.
+ """
+ def __init__(self, quantile_tracker: Optional[QuantileTracker] = None):
+ self._quantile_tracker = quantile_tracker or BufferedSlidingQuantileTracker(
+ DEFAULT_WINDOW_SIZE, 0.5)
+ assert self._quantile_tracker._q == 0.5, \
+ "quantile_tracker must be initialized with q = 0.5"
+
+ def push(self, x):
+ """Pushes a new value and updates the internal quantile tracker.
+
+ Args:
+ x: The new value to be pushed.
+ """
+ self._quantile_tracker.push(x)
+
+ def get(self):
+ """Calculates and returns the median (q = 0.5).
+
+ Returns:
+ float: The median of the values in the window setting specified in the
+ internal quantile tracker. Returns NaN if the window is empty.
+ """
+ return self._quantile_tracker.get()
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/median_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/median_test.py
new file mode 100644
index 00000000000..d1ce660d4d2
--- /dev/null
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/median_test.py
@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import math
+import unittest
+
+from apache_beam.ml.anomaly.univariate.median import MedianTracker
+from apache_beam.ml.anomaly.univariate.quantile import SimpleSlidingQuantileTracker # pylint: disable=line-too-long
+
+
+class MedianTest(unittest.TestCase):
+ def test_default_tracker(self):
+ t = MedianTracker()
+ self.assertTrue(math.isnan(t.get()))
+
+ t.push(1.0)
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(2.0)
+ self.assertEqual(t.get(), 1.5)
+
+ t.push(1.0)
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(20.0)
+ self.assertEqual(t.get(), 1.5)
+
+ t.push(10.0)
+ self.assertEqual(t.get(), 2.0)
+
+ def test_custom_tracker(self):
+ t = MedianTracker(SimpleSlidingQuantileTracker(3, 0.5))
+ self.assertTrue(math.isnan(t.get()))
+
+ t.push(1.0)
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(2.0)
+ self.assertEqual(t.get(), 1.5)
+
+ t.push(1.0)
+ self.assertEqual(t.get(), 1.0)
+
+ t.push(20.0)
+ self.assertEqual(t.get(), 2.0)
+
+ t.push(10.0)
+ self.assertEqual(t.get(), 10.0)
+
+ def test_wrong_tracker(self):
+ t = MedianTracker(SimpleSlidingQuantileTracker(50, 0.1))
+ self.assertRaises(AssertionError, lambda: t.run_original_init())
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/quantile.py b/sdks/python/apache_beam/ml/anomaly/univariate/quantile.py
index 44d2b1ab244..ff3ee6061ee 100644
--- a/sdks/python/apache_beam/ml/anomaly/univariate/quantile.py
+++ b/sdks/python/apache_beam/ml/anomaly/univariate/quantile.py
@@ -116,6 +116,21 @@ class BufferedQuantileTracker(WindowedTracker, QuantileTracker):
super().push(x)
+ @staticmethod
+ def _get_helper(sorted_items, q):
+ n = len(sorted_items)
+ if n < 1:
+ return float("nan")
+
+ pos = q * (n - 1)
+ lo = math.floor(pos)
+ lo_value = typing.cast(float, sorted_items[lo])
+
+ # Use linear interpolation to yield the requested quantile
+ hi = min(lo + 1, n - 1)
+ hi_value: float = typing.cast(float, sorted_items[hi])
+ return lo_value + (hi_value - lo_value) * (pos - lo)
+
def get(self):
"""Returns the current quantile value using the sorted list.
@@ -124,18 +139,41 @@ class BufferedQuantileTracker(WindowedTracker, QuantileTracker):
Returns:
float: The calculated quantile value. Returns NaN if the window is empty.
"""
- n = len(self._sorted_items)
- if n < 1:
- return float("nan")
+ return self._get_helper(self._sorted_items, self._q)
- pos = self._q * (n - 1)
- lo = math.floor(pos)
- lo_value = typing.cast(float, self._sorted_items[lo])
- # Use linear interpolation to yield the requested quantile
- hi = min(lo + 1, n - 1)
- hi_value: float = typing.cast(float, self._sorted_items[hi])
- return lo_value + (hi_value - lo_value) * (pos - lo)
+@specifiable
+class SecondaryBufferedQuantileTracker(WindowedTracker, QuantileTracker):
+ """A secondary quantile tracker that shares its data with a master tracker.
+
+ This tracker acts as a read-only view of the master tracker's data, providing
+ quantile calculations without maintaining its own independent buffer. It
+ relies on the master's sorted items for quantile estimations.
+
+ Args:
+ master: The BufferedQuantileTracker instance to share data with.
+ q: A list of quantiles to track.
+ """
+ def __init__(self, master: QuantileTracker, q):
+ assert isinstance(master, BufferedQuantileTracker), \
+ "Cannot create secondary tracker from non-BufferedQuantileTracker"
+ self._master = master
+ super().__init__(self._master._window_mode)
+ QuantileTracker.__init__(self, q)
+ self._sorted_items = self._master._sorted_items
+
+ def push(self, x):
+ """Does nothing, as this is a secondary tracker.
+ """
+ pass
+
+ def get(self):
+ """Returns the calculated quantiles based on the master tracker's buffer.
+
+ Returns:
+ A list of calculated quantiles.
+ """
+ return self._master._get_helper(self._master._sorted_items, self._q)
@specifiable