Exponential Moving Average

exponential_moving_average.py

from time import time


class ExponentialMovingAverage:

    def __init__(self, timeout, tau) -> None:
        self.timeout = timeout
        self.historic = {}
        assert 0 < tau < 1
        self.tau = tau

    def __call__(self, **kwargs):

        def calculate(key, value):
            now = time()
            if (
                key not in self.historic or
                now - self.historic[key][0] > self.timeout
            ):
                self.historic[key] = {'value': value, 'time': now}
                return value

            else:
                historic_value = self.historic[key]['value']
                new_value = self.tau*historic_value + (1-self.tau)*value
                self.historic[key] = {'value': new_value, 'time': now}
                return new_value

        if len(kwargs) == 1:
            key, value = kwargs.popitem()
            return calculate(key, value)
        else:
            return [calculate(key, value) for key, value in kwargs.items()]


if __name__ == '__main__':
    from random import random
    ema = ExponentialMovingAverage(timeout=0.5, tau=0.9)
    while True:
        print(ema(a=10*random(), b=10*random()))

var_backup.py

from functools import wraps, partial


class VarBackup:
    """ Class to backup variables """

    def __init__(self, num_error: int = 0) -> None:
        self.num_error = num_error
        self.historic = {}

    def __call__(self, **kwargs):

        def restore(key, value):

            if value is not None:
                self.historic[key] = {'value': value, 'num_error': 0}
                return value
            elif key in self.historic:
                if self.historic[key]['num_error'] < self.num_error:
                    self.historic[key]['num_error'] += 1
                    return self.historic[key]['value']

        if len(kwargs) == 1:
            key, value = kwargs.popitem()
            return restore(key, value)
        else:
            return (restore(key, value) for key, value in kwargs.items())


def backup(func=None, max_error: int = 3):
    """ Decorator to backup functions results """

    if func is None:
        return partial(backup, max_error=max_error)

    num_error = 0
    last_result = None

    @wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if result is None:
            if num_error < max_error:
                nonlocal num_error, last_result
                num_error += 1
                return last_result
        else:
            nonlocal num_error, last_result
            num_error = 0
            last_result = result
            return result
    return wrapper