contango.stream.indicators.calculations.sma

contango.stream.indicators.calculations.sma

SMA

SMA(pd: days)

Bases: Indicator[float | None]

Simple Moving Average.

Maintains a window of prices and tracks their sum incrementally. Returns None until the window is full.

Initializes SMA.

Parameters:

Name	Type	Description	Default
pd	days	The number of days (period) for the rolling window.	required

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

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

    Args:
        pd: The number of days (period) for the rolling window.
    """
    self._period = pd
    self._window: deque[float] = deque(maxlen=pd)
    self._sum = 0.0
    self._value: float | None = None

value property

value: float | None

Returns the latest value for the SMA.

update

update(event: MarketDataEvent) -> float | None

Takes in a new price and returns the current SMA, or None if the window is not yet full.

Parameters:

Name	Type	Description	Default
event	MarketDataEvent	The underlying event to update teh SMA with.	required

Returns:

Type	Description
float | None	The SMA over the current window, or None if period amount of days have not passed.

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

def update(self, event: MarketDataEvent) -> float | None:
    """
    Takes in a new price and returns the current SMA, or None if the window is not yet full.

    Args:
        event: The underlying event to update teh SMA with.

    Returns:
        The SMA over the current window, or `None` if `period` amount of days have not passed.
    """
    if len(self._window) == self._period:
        self._sum -= self._window[0]

    self._window.append(event.close)
    self._sum += event.close

    if len(self._window) < self._period:
        return None

    self._value = self._sum / self._period

    return self._value