import math from typing import List, Optional, Sequence from .monitor_anomaly_types import Anomaly, AnomalyMethod def _half_up(n: int, d: int) -> int: # Half-up with halves away from zero, as math.round-div rounds. q = (2 * abs(n) + d) // (2 * d) return -q if n < 0 else q def _median_twice(ordered: List[int]) -> int: m = len(ordered) // 2 return 2 * ordered[m] if len(ordered) % 2 == 1 else ordered[m - 1] + ordered[m] def detect_anomaly(history: Sequence[int], value: int, method: AnomalyMethod, threshold_hundredths: int) -> Anomaly: """Whether `value` is an outlier against `history`, by z-score (`stddev`) or by Iglewicz and Hoaglin's modified z-score (`mad`). Exact integer arithmetic: the threshold comparison never depends on float rounding.""" if len(history) < 2: raise ValueError("history must have at least 2 values, received %d" % len(history)) if threshold_hundredths < 1: raise ValueError("thresholdHundredths must be at least 1, received %d" % threshold_hundredths) n = len(history) t = threshold_hundredths score: Optional[int] if method == "stddev": s = sum(history) ss = sum(x * x for x in history) # n^2 * variance and n * (value - mean), both whole numbers. q = n * ss - s * s signed = n * value - s center = _half_up(s * 1000, n) spread = math.isqrt(1000000 * q) // n if q == 0: anomaly = signed != 0 score = None else: anomaly = 10000 * signed * signed > t * t * q score = math.isqrt((10000 * signed * signed) // q) elif method == "mad": # Twice the median and four times the MAD are whole numbers. med2 = _median_twice(sorted(history)) mad4 = _median_twice(sorted(abs(2 * x - med2) for x in history)) signed = 2 * value - med2 center = med2 * 500 spread = mad4 * 250 if mad4 == 0: anomaly = signed != 0 score = None else: # |M| = 0.6745 * |2v - med2| / 2 / (mad4 / 4) = 1349 * |signed| / (1000 * mad4) anomaly = 1349 * abs(signed) * 100 > t * 1000 * mad4 score = (1349 * abs(signed) * 100) // (1000 * mad4) else: raise ValueError("unknown anomaly method: %s" % method) direction = "normal" if not anomaly else ("high" if signed > 0 else "low") return Anomaly(anomaly=anomaly, direction=direction, center_milli=center, spread_milli=spread, score_hundredths=score)