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