impl/python.py

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)