Skip to content

Base

IndicatorBase

Bases: ABC

Base class for scalar technical indicators with per-symbol history.

The class provides a thread-safe mechanism for storing and retrieving indicator values computed from incoming market bars, keyed by symbol. It does not manage input windows or rolling computation state.

Subclasses define a stable indicator identifier via the name property and implement _compute_indicator, which computes a single scalar value per incoming bar. Indicators with multiple conceptual outputs must be implemented as multiple single-output indicators (e.g. Bollinger Bands must be implemented via three separate indicators BBUpper, BBMiddle, and BBLower).

The update mechanism is thread-safe. Indicator computation is performed outside the internal lock. Subclasses that maintain internal state are responsible for ensuring its thread safety and must not access _history_data.

Indicator values are stored per symbol in bounded FIFO buffers. Missing data and out-of-bounds access yield numpy.nan.

The plot_at attribute is an opaque identifier forwarded to the charting backend and has no intrinsic meaning within the indicator subsystem.

name abstractmethod property

Canonical indicator name.

Returns:

Type Description
str

Stable identifier used for charting and downstream integration.
Source code in src/onesecondtrader/indicators/base.py 
48
49
50
51
52
53
54
55
def name(self) -> str:
    """
    Canonical indicator name.

    Returns:
        Stable identifier used for charting and downstream integration.
    """
    pass

plot_at property

Plotting identifier.

Returns:

Type Description
int

Opaque identifier consumed by the charting backend.
Source code in src/onesecondtrader/indicators/base.py 
135
136
137
138
139
140
141
142
def plot_at(self) -> int:
    """
    Plotting identifier.

    Returns:
        Opaque identifier consumed by the charting backend.
    """
    return self._plot_at

__init__(max_history=100, plot_at=99)

Parameters:

Name Type Description Default
max_history int

Maximum number of indicator values retained per symbol. Cannot be less than 1.

 100
plot_at int

Opaque plotting identifier forwarded to the charting backend.

 99
Source code in src/onesecondtrader/indicators/base.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, max_history: int = 100, plot_at: int = 99) -> None:
    """
    Parameters:
        max_history:
            Maximum number of indicator values retained per symbol.
            Cannot be less than 1.
        plot_at:
            Opaque plotting identifier forwarded to the charting backend.
    """
    self._lock = threading.Lock()
    self._max_history = max(1, int(max_history))
    self._history_data: dict[str, collections.deque[float]] = {}
    self._plot_at = plot_at

_compute_indicator(incoming_bar) abstractmethod

Compute the indicator value for a single market bar.

This method is executed outside the internal lock. Implementations must not access _history_data and must ensure thread safety of any internal computation state.

Parameters:

Name Type Description Default
incoming_bar BarReceived

Market bar used as input for indicator computation.

 required

Returns:

Type Description
float

Computed indicator value.
Source code in src/onesecondtrader/indicators/base.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abc.abstractmethod
def _compute_indicator(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Compute the indicator value for a single market bar.

    This method is executed outside the internal lock.
    Implementations must not access `_history_data` and must ensure thread safety of any internal computation state.

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

    Returns:
        Computed indicator value.
    """
    pass

update(incoming_bar)

Update the indicator with a new market bar.

The computed value is appended to the per-symbol history buffer.

Parameters:

Name Type Description Default
incoming_bar BarReceived

Market bar triggering the update.

 required
Source code in src/onesecondtrader/indicators/base.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def update(self, incoming_bar: events.market.BarReceived) -> None:
    """
    Update the indicator with a new market bar.

    The computed value is appended to the per-symbol history buffer.

    Parameters:
        incoming_bar:
            Market bar triggering the update.
    """
    symbol = incoming_bar.symbol

    value = self._compute_indicator(incoming_bar)

    with self._lock:
        if symbol not in self._history_data:
            self._history_data[symbol] = collections.deque(maxlen=self._max_history)

        self._history_data[symbol].append(value)

latest(symbol)

Return the most recent indicator value for a symbol.

Parameters:

Name Type Description Default
symbol str

Symbol identifier.

 required

Returns:

Type Description
float

Most recent value, or numpy.nan if unavailable.
Source code in src/onesecondtrader/indicators/base.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def latest(self, symbol: str) -> float:
    """
    Return the most recent indicator value for a symbol.

    Parameters:
        symbol:
            Symbol identifier.

    Returns:
        Most recent value, or `numpy.nan` if unavailable.
    """
    return self[symbol, -1]

__getitem__(key)

Retrieve an indicator value by symbol and index.

Indexing follows standard Python sequence semantics. Negative indices refer to positions relative to the most recent value.

Parameters:

Name Type Description Default
key tuple[str, int]

(symbol, index) pair specifying the symbol and history offset.

 required

Returns:

Type Description
float

Indicator value at the specified position, or numpy.nan if unavailable.
Source code in src/onesecondtrader/indicators/base.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def __getitem__(self, key: tuple[str, int]) -> float:
    """
    Retrieve an indicator value by symbol and index.

    Indexing follows standard Python sequence semantics.
    Negative indices refer to positions relative to the most recent value.

    Parameters:
        key:
            `(symbol, index)` pair specifying the symbol and history offset.

    Returns:
        Indicator value at the specified position, or `numpy.nan` if unavailable.
    """
    symbol, index = key

    with self._lock:
        history = self._history_data.get(symbol)

        if history is None:
            return np.nan

        try:
            return history[index]
        except IndexError:
            return np.nan