Module `indicators`

Sub-modules

backtest.indicators.TA
backtest.indicators.indicator
backtest.indicators.indicatorSet

Classes

class Indicator (out_feat: List[str], numba: bool = False, name: Optional[str] = None, **expected_params: type)

This class is a decorator designed to make an Indicator out of a function.

Example

>>> @Indicator(out_feat=["SMA"], period=int)
... def SMA(data: np.ndarray, index: List[str], features: List[str],
...     previous_data: Optional[npt.NDArray[np.float32]], period: int = 10) -> np.ndarray:
...
...     out = np.zeros(len(data), dtype=np.float32)
...     for i in range(len(data) - period + 1):
...         out[i] = data[i: i+period, 0].sum() / period
...     return out[:, np.newaxis]
>>> # Usage
>>> series = pd.DataFrame(np.random.rand(1000), index=pd.date_range(start="2021-01-01", periods=1000, freq="min"), columns=["Close"])
>>> indicator_10d = SMA()
>>> indicator_20d = SMA(period=20)
>>> out_10d = indicator_10d.get(series)
>>> out_20d = indicator_20d.get(series)

The indicator function receives the data as a 2D numpy array, the index as a list of datetimes (or string if jit compiled), the features as a list of strings, and any parameters set by the user. The function must return a 2D numpy array with the same length as the input data.

Args:

out_feat: The features name that this indicator will return. (Will be the column name in resulting df)
numba: Whether the function uses numba JIT compilation or not. If True, the index passed to the indicator logic will be a list of strings, otherwise a list of datetime objects.
name: The name of the indicator. If not set, the name of the function will be used.
expected_params: The expected parameters of the indicator. Used to validate the parameters passed to the indicator.

Static methods

def run(data: numpy.ndarray[typing.Any, numpy.dtype[numpy.float32]], index: Union[List[datetime.datetime], List[str]], features: List[str], previous_data: Optional[numpy.ndarray[typing.Any, numpy.dtype[numpy.float32]]] = None, **params) ‑> numpy.ndarray[typing.Any, numpy.dtype[numpy.float32]]

This method can be overridden to implement the indicator logic. This method is called when the callback method is not set.

Args:

data: The input data as a 2d numpy array.
index: The index of the input data. (datetime if not compiled with numba or string if compiled
features: The features of the input data
previous_data: An array of the previous indicator results. The recent points in the array will be nan. These are the values to fill. This parameter is useful when streaming the indicator (more compute efficient, but not all indicators support it.)
params: Any parameters set by the user

Returns:

A 2D array with the indicator results. Must be the same length as the input data.

Instance variables

var name : str

Returns:

A unique name for the indicator. (The name of the function and its identifier if different from 0)

var out : List[str]

Returns:

The output features of the indicator

var type_name : str

Returns:

The name of the indicator type

Methods

def get(self, data: pandas.core.frame.DataFrame, previous_values: Optional[pandas.core.frame.DataFrame]) ‑> pandas.core.frame.DataFrame

Run the indicator logic on the data and returns the result as a DataFrame.

Args:

data: The input dataframe with a datetime index. (Usually the chart)
previous_values: The previous values of the indicator. (Used when streaming the indicator)

Returns:

The indicator results as a dataframe

def get_args(self) ‑> str

Returns:

The arguments of the indicator as a string

def get_name_no_id(self) ‑> str

Returns:

The name of the indicator without the identifier

def set_callback(self, cb: Callable[[numpy.ndarray[Any, numpy.dtype[numpy.float32]], Union[List[datetime.datetime], List[str]], List[str], Optional[numpy.ndarray[Any, numpy.dtype[numpy.float32]]], ...], numpy.ndarray[Any, numpy.dtype[numpy.float32]]]) ‑> None

Set the callback method to be used when the indicator is called.

Args:

cb: The indicator logic

Returns:

None

def set_id(self, new: int)

Set the identifier of the indicator. Used to identify the indicator in the indicator set when there are multiple indicators with the same name.

Args:

new: The new id

Returns:

None

def set_params(self, **params: Any) ‑> None

Set the parameters of the indicator.

Args:

params: The parameters set by the user

Returns:

None

class IndicatorSet (*indicators: Indicator, streaming: bool = False)

Args:

indicators: Indicators passed as arguments
streaming: Whether to use the streaming capabilities of the indicators or not. (Note that if true, indicators will have the choice to use it or not.) Streaming can improve backtest run time. This parameter is handled in the Backtest object.

Instance variables

var indicators : Dict[str, Indicator]

Get the indicators in the IndicatorSet as a dictionary with the indicator name as the key. IF there are multiple indicators with the same name, the name of the indicator will be appended with a number. Example, if I have two SMA, the first one will be SMA_1, and the second one SMA_2

Returns:

The dictionary of indicators

var out : list[str]

Returns:

The name of the columns added to the chart

var streaming : bool

Returns:

Whether the IndicatorSet is set to use streaming capabilities or not.

Methods

def add(self, indicators: Union[Indicator, List[Indicator]])

Add an indicator or list of indicators to the IndicatorSet.

Args:

indicators: A single indicator or a list of indicators

Returns:

None

def run_all(self, data: pandas.core.frame.DataFrame, previous_data: Optional[pandas.core.frame.DataFrame] = None) ‑> pandas.core.frame.DataFrame

Run all the indicators in the IndicatorSet on the data and return the results as a DataFrame. It concatenates the input data with the results of the indicators. (The index should be a datetime index)

Args:

data: A dataframe containing the previously calculated values of the indicators. (Used when streaming) If streaming is False, this parameter is ignored. This should be a dataframe with the same index as the data. This means that the indicators points that needs to be calculated should be nan, and one already calculated should be the previously calculated values. The columns should be the output columns names of the indicators.

Returns:

The data concatenated with the indicator results

def toList(self) ‑> List[str]

Convert the IndicatorSet to a list of string representations of the indicators.

Returns:

List of string representations of the indicators