contango.stream.indicators.calculations.ema

contango.stream.indicators.calculations.ema

EMA

EMA(pd: days)

Bases: Indicator[float | None]

Exponential Moving Average.

Uses alpha = 2 / (period + 1) and seeds the first value directly from the first price to avoid bias (towards 0).

Initializes Ema.

Parameters:

Name	Type	Description	Default
pd	days	Lookback period to derive the smoothing factor. Larger values -> slower to react to price changes.	required

Source code in src\contango\stream\indicators\calculations\ema.py

def __init__(self, pd: days) -> None:
    """
    Initializes `Ema`.

    Args:
        pd: Lookback period to derive the smoothing factor. Larger values -> slower to react to price changes.
    """
    self._alpha = 2 / (pd + 1)
    self._value: float | None = None
    self._period = pd

value property

value: float | None

Returns the latest value for the EMA.

update

update(event: MarketDataEvent) -> float

Takes in a new price and returns the current EMA.

Parameters:

Name	Type	Description	Default
event	MarketDataEvent	The current market data event.	required

Returns:

Type	Description
float	The current EMA. The first call returns price unsmoothed.

Source code in src\contango\stream\indicators\calculations\ema.py

def update(self, event: MarketDataEvent) -> float:
    """
    Takes in a new price and returns the current EMA.

    Args:
        event: The current market data event.

    Returns:
        The current EMA. The first call returns `price` unsmoothed.
    """
    if self._value is None:
        self._value = event.close
    else:
        self._value = event.close * self._alpha + self._value * (1 - self._alpha)
    return self._value