|
| 1 | +""" |
| 2 | + Calculate the exponential moving average (EMA) on the series of stock prices. |
| 3 | + Wikipedia Reference: https://en.wikipedia.org/wiki/Exponential_smoothing |
| 4 | + https://www.investopedia.com/terms/e/ema.asp#toc-what-is-an-exponential |
| 5 | + -moving-average-ema |
| 6 | +
|
| 7 | + Exponential moving average is used in finance to analyze changes stock prices. |
| 8 | + EMA is used in conjunction with Simple moving average (SMA), EMA reacts to the |
| 9 | + changes in the value quicker than SMA, which is one of the advantages of using EMA. |
| 10 | +""" |
| 11 | + |
| 12 | +from collections.abc import Iterator |
| 13 | + |
| 14 | + |
| 15 | +def exponential_moving_average( |
| 16 | + stock_prices: Iterator[float], window_size: int |
| 17 | +) -> Iterator[float]: |
| 18 | + """ |
| 19 | + Yields exponential moving averages of the given stock prices. |
| 20 | + >>> tuple(exponential_moving_average(iter([2, 5, 3, 8.2, 6, 9, 10]), 3)) |
| 21 | + (2, 3.5, 3.25, 5.725, 5.8625, 7.43125, 8.715625) |
| 22 | +
|
| 23 | + :param stock_prices: A stream of stock prices |
| 24 | + :param window_size: The number of stock prices that will trigger a new calculation |
| 25 | + of the exponential average (window_size > 0) |
| 26 | + :return: Yields a sequence of exponential moving averages |
| 27 | +
|
| 28 | + Formula: |
| 29 | +
|
| 30 | + st = alpha * xt + (1 - alpha) * st_prev |
| 31 | +
|
| 32 | + Where, |
| 33 | + st : Exponential moving average at timestamp t |
| 34 | + xt : stock price in from the stock prices at timestamp t |
| 35 | + st_prev : Exponential moving average at timestamp t-1 |
| 36 | + alpha : 2/(1 + window_size) - smoothing factor |
| 37 | +
|
| 38 | + Exponential moving average (EMA) is a rule of thumb technique for |
| 39 | + smoothing time series data using an exponential window function. |
| 40 | + """ |
| 41 | + |
| 42 | + if window_size <= 0: |
| 43 | + raise ValueError("window_size must be > 0") |
| 44 | + |
| 45 | + # Calculating smoothing factor |
| 46 | + alpha = 2 / (1 + window_size) |
| 47 | + |
| 48 | + # Exponential average at timestamp t |
| 49 | + moving_average = 0.0 |
| 50 | + |
| 51 | + for i, stock_price in enumerate(stock_prices): |
| 52 | + if i <= window_size: |
| 53 | + # Assigning simple moving average till the window_size for the first time |
| 54 | + # is reached |
| 55 | + moving_average = (moving_average + stock_price) * 0.5 if i else stock_price |
| 56 | + else: |
| 57 | + # Calculating exponential moving average based on current timestamp data |
| 58 | + # point and previous exponential average value |
| 59 | + moving_average = (alpha * stock_price) + ((1 - alpha) * moving_average) |
| 60 | + yield moving_average |
| 61 | + |
| 62 | + |
| 63 | +if __name__ == "__main__": |
| 64 | + import doctest |
| 65 | + |
| 66 | + doctest.testmod() |
| 67 | + |
| 68 | + stock_prices = [2.0, 5, 3, 8.2, 6, 9, 10] |
| 69 | + window_size = 3 |
| 70 | + result = tuple(exponential_moving_average(iter(stock_prices), window_size)) |
| 71 | + print(f"{stock_prices = }") |
| 72 | + print(f"{window_size = }") |
| 73 | + print(f"{result = }") |
0 commit comments