Skip to content

Moving Averages

SimpleMovingAverage

Bases: IndicatorBase

Simple Moving Average (SMA) indicator.

This indicator computes the arithmetic mean of a selected bar field over a fixed rolling window. One scalar value is produced per incoming bar and stored per symbol.

The rolling window is maintained independently for each symbol. Until the window is fully populated, the indicator yields numpy.nan.

name property

Canonical indicator name.

Returns:

Type Description
str

Identifier encoding the indicator type, period, and bar field.
Source code in src/onesecondtrader/indicators/moving_averages.py 
45
46
47
48
49
50
51
52
def name(self) -> str:
    """
    Canonical indicator name.

    Returns:
        Identifier encoding the indicator type, period, and bar field.
    """
    return f"SMA_{self.period}_{self.bar_field.name}"

__init__(period=200, max_history=100, bar_field=models.BarField.CLOSE, plot_at=0)

Parameters:

Name Type Description Default
period int

Window size used to compute the moving average.

 200
max_history int

Maximum number of computed indicator values retained per symbol.

 100
bar_field BarField

Bar field used as the input series.

 CLOSE
plot_at int

Opaque plotting identifier forwarded to the charting backend.

 0
Source code in src/onesecondtrader/indicators/moving_averages.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    period: int = 200,
    max_history: int = 100,
    bar_field: models.BarField = models.BarField.CLOSE,
    plot_at: int = 0,
) -> None:
    """
    Parameters:
        period:
            Window size used to compute the moving average.
        max_history:
            Maximum number of computed indicator values retained per symbol.
        bar_field:
            Bar field used as the input series.
        plot_at:
            Opaque plotting identifier forwarded to the charting backend.
    """
    super().__init__(max_history=max_history, plot_at=plot_at)

    self.period: int = max(1, int(period))
    self.bar_field: models.BarField = bar_field
    self._window: dict[str, collections.deque[float]] = {}

_compute_indicator(incoming_bar)

Compute the simple moving average for a single received bar.

Parameters:

Name Type Description Default
incoming_bar BarReceived

Market bar used as input for the computation.

 required

Returns:

Type Description
float

Simple moving average value, or numpy.nan if the rolling window is not yet fully populated.
Source code in src/onesecondtrader/indicators/moving_averages.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def _compute_indicator(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Compute the simple moving average for a single received bar.

    Parameters:
        incoming_bar:
            Market bar used as input for the computation.

    Returns:
        Simple moving average value, or `numpy.nan` if the rolling window is not yet fully populated.
    """
    symbol = incoming_bar.symbol
    if symbol not in self._window:
        self._window[symbol] = collections.deque(maxlen=self.period)

    window = self._window[symbol]
    value = self._extract_field(incoming_bar)
    window.append(value)

    if len(window) < self.period:
        return np.nan
    return sum(window) / self.period

_extract_field(incoming_bar)

Extract the configured bar field from an incoming bar.

Parameters:

Name Type Description Default
incoming_bar BarReceived

Market bar providing the input data.

 required

Returns:

Type Description
float

Extracted field value, or numpy.nan if unavailable.
Source code in src/onesecondtrader/indicators/moving_averages.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def _extract_field(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Extract the configured bar field from an incoming bar.

    Parameters:
        incoming_bar:
            Market bar providing the input data.

    Returns:
        Extracted field value, or `numpy.nan` if unavailable.
    """
    match self.bar_field:
        case models.BarField.OPEN:
            return incoming_bar.open
        case models.BarField.HIGH:
            return incoming_bar.high
        case models.BarField.LOW:
            return incoming_bar.low
        case models.BarField.CLOSE:
            return incoming_bar.close
        case models.BarField.VOLUME:
            return (
                float(incoming_bar.volume)
                if incoming_bar.volume is not None
                else np.nan
            )
        case _:
            return incoming_bar.close