forked from wolph/python-progressbar
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathalgorithms.py
More file actions
52 lines (40 loc) · 1.65 KB
/
Copy pathalgorithms.py
File metadata and controls
52 lines (40 loc) · 1.65 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from __future__ import annotations
import abc
import typing
from datetime import timedelta
class SmoothingAlgorithm(abc.ABC):
@abc.abstractmethod
def __init__(self, **kwargs: typing.Any):
raise NotImplementedError
@abc.abstractmethod
def update(self, new_value: float, elapsed: timedelta) -> float:
"""Updates the algorithm with a new value and returns the smoothed
value.
"""
raise NotImplementedError
class ExponentialMovingAverage(SmoothingAlgorithm):
"""
The Exponential Moving Average (EMA) is an exponentially weighted moving
average that reduces the lag that's typically associated with a simple
moving average. It's more responsive to recent changes in data.
"""
def __init__(self, alpha: float = 0.5) -> None:
self.alpha = alpha
self.value = 0
def update(self, new_value: float, elapsed: timedelta) -> float:
self.value = self.alpha * new_value + (1 - self.alpha) * self.value
return self.value
class DoubleExponentialMovingAverage(SmoothingAlgorithm):
"""
The Double Exponential Moving Average (DEMA) is essentially an EMA of an
EMA, which reduces the lag that's typically associated with a simple EMA.
It's more responsive to recent changes in data.
"""
def __init__(self, alpha: float = 0.5) -> None:
self.alpha = alpha
self.ema1 = 0
self.ema2 = 0
def update(self, new_value: float, elapsed: timedelta) -> float:
self.ema1 = self.alpha * new_value + (1 - self.alpha) * self.ema1
self.ema2 = self.alpha * self.ema1 + (1 - self.alpha) * self.ema2
return 2 * self.ema1 - self.ema2