IndicatorRegistry

API documentation for tradeexecutor.strategy.pandas_trader.indicator_decorator.IndicatorRegistry Python class in Trading Strategy framework.

class IndicatorRegistry

Bases: object

Decorator-based helper class for defining strategy indicators.

Use @indicatores.define() decorator to create strategy technical indicators in your notebooks.

• Less typing than with manuaal indicator.add() syntax

• Parameters class members are automatically mapped to the function argument with the same name

• Indicators list other indicator functions they depend on using syntax @indicators.define(dependencies=(slow_sma, fast_sma)), this is automatically resolved to the current dependency resolution order when the indicator calulation order is determined.

• Indicator data can be looked up by indicator function or its string name

Example:

import pandas_ta
from tradeexecutor.strategy.pandas_trader.indicator_decorator import IndicatorRegistry

class Parameters:
    rsi_length = 20

indicators = IndicatorRegistry()

@indicators.define()
def rsi(close, rsi_length):
    return pandas_ta.rsi(close, rsi_length)

# Calculate indicators - will spawn multiple worker processed,
# or load cached results from the disk
parameters = StrategyParameters.from_class(Parameters)
indicators = calculate_and_load_indicators_inline(
    strategy_universe=strategy_universe,
    parameters=parameters,
    create_indicators=indicators.create_indicators,
)

With indicators that depend on each other:

import pandas_ta
from tradeexecutor.strategy.pandas_trader.indicator_decorator import IndicatorRegistry


class Parameters:
    fast_ma_length = 7
    slow_ma_length = 21

indicators = IndicatorRegistry()

@indicators.define()
def slow_sma(close, slow_ma_length):
    return pandas_ta.sma(close, slow_ma_length)

@indicators.define()
def fast_sma(close, fast_ma_length):
    return pandas_ta.sma(close, fast_ma_length)

@indicators.define(dependencies=(slow_sma, fast_sma))
def ma_crossover(
    close: pd.Series,
    pair: TradingPairIdentifier,
    dependency_resolver: IndicatorDependencyResolver,
) -> pd.Series:
    # You can use name or function to look up previously calcualted indicators data
    slow_sma: pd.Series = dependency_resolver.get_indicator_data(slow_sma)
    fast_sma: pd.Series = dependency_resolver.get_indicator_data("fast_sma")
    return fast_sma > slow_sma

__init__()

Methods

__init__()
create_indicators(timestamp, parameters, ...)	Hook function to be passed to calculate_and_load_indicators_inline().
define([source, order, dependencies])	Function decorator to define indicator functions in your notebook.
detect_order(func, dependencies)	Automatically resolve the order of indicators.
get_diagnostics()	Return a table that explains registered indicators.

__init__()

define(source=None, order=None, dependencies=None)

Function decorator to define indicator functions in your notebook.

For the usage see IndicatorRegistry.

Short example:

class Parameters:
    slow_ma_length = 21

indicators = IndicatorRegistry()

@indicators.define()
def slow_sma(close, slow_ma_length):
    return pandas_ta.sma(close, slow_ma_length)

Parameters:

• source (IndicatorSource) –

  What kind of trading data this indicator needs as input.

  • Per-pair

  • World

  • Raw data

  • Data from earlier calculated indicators

• order (None | int) –

  Manually define indicator order.

  Not usually needed.

• dependencies (Sequence[Callable]) –

  List of functions which result value this indicator consumes.

  Needed with dependencies_only_per_pair and dependencies_only_universe.

detect_order(func, dependencies)

Automatically resolve the order of indicators.

Returns:

Linear order number for running the indicator.

Can be used to parallerise the process.

Parameters:

• func (Callable) –

• dependencies (list[Callable]) –

Return type:

int

create_indicators(timestamp, parameters, strategy_universe, execution_context)

Hook function to be passed to calculate_and_load_indicators_inline().

Example:

indicators = IndicatorRegistry()

@indicators.define()
def rsi(close, rsi_length):
    return pandas_ta.rsi(close, rsi_length)

# Calculate indicators - will spawn multiple worker processed,
# or load cached results from the disk
parameters = StrategyParameters.from_class(Parameters)
indicators = calculate_and_load_indicators_inline(
    strategy_universe=strategy_universe,
    parameters=parameters,
    create_indicators=indicators.create_indicators,
)

Parameters:

• timestamp (datetime) –

• parameters (StrategyParameters) –

• strategy_universe (TradingStrategyUniverse) –

• execution_context (ExecutionContext) –

Return type:

IndicatorSet

get_diagnostics()

Return a table that explains registered indicators.

Returns:

Human-readable table

Return type:

DataFrame