impl/python.py

import math
from typing import List, Sequence

from .stats_moving_average_types import MovingAverageKind


POW10 = [1.0, 10.0, 100.0, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8, 1e9, 1e10, 1e11, 1e12]


def _round_to(x: float, decimals: int) -> float:
    # Half away from zero on the binary64 value, then -0 becomes 0. Python's
    # round() is half-even and would give round(0.5) == 0; this does not.
    scale = POW10[decimals]
    y = abs(x) * scale
    r = float(math.floor(y))
    if y - r >= 0.5:
        r += 1.0
    out = r / scale
    return (-out if x < 0 else out) + 0.0


def _window_mean(values: List[float], end: int, window: int) -> float:
    # Summed afresh, left to right; math.fsum or sum() with a start value
    # would round differently from the other languages.
    total = 0.0
    for i in range(end - window + 1, end + 1):
        total = total + values[i]
    return total / window


def moving_average(values: Sequence[float], window: int, kind: MovingAverageKind, decimals: int) -> List[float]:
    """Simple or exponential moving average, one value per full window.

    The exponential average is seeded with the simple average of the first
    window and carried unrounded; only the returned values are rounded.
    """
    if isinstance(values, (str, bytes)) or not isinstance(values, (list, tuple)):
        raise TypeError("values must be a list of numbers")
    for v in values:
        if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v):
            raise TypeError("values must be finite numbers, received %r" % (v,))
    if isinstance(window, bool) or not isinstance(window, int) or window < 1:
        raise ValueError("window must be a whole number of 1 or greater, received %r" % (window,))
    if kind not in ("simple", "exponential"):
        raise ValueError('unknown moving average kind "%s"' % (kind,))
    if isinstance(decimals, bool) or not isinstance(decimals, int) or decimals < 0 or decimals > 12:
        raise ValueError("decimals must be a whole number from 0 to 12, received %r" % (decimals,))

    series = [float(v) for v in values]
    out: List[float] = []
    if len(series) < window:
        return out

    if kind == "simple":
        for end in range(window - 1, len(series)):
            out.append(_round_to(_window_mean(series, end, window), decimals))
        return out

    alpha = 2.0 / (window + 1)
    ema = _window_mean(series, window - 1, window)
    out.append(_round_to(ema, decimals))
    for i in range(window, len(series)):
        ema = ema + alpha * (series[i] - ema)
        out.append(_round_to(ema, decimals))
    return out