IndicatorDependencyResolver#

API documentation for tradeexecutor.strategy.pandas_trader.indicator.IndicatorDependencyResolver Python class in Trading Strategy framework.

class IndicatorDependencyResolver[source]#

Bases: object

A helper class allowing access to the indicators we depend on.

  • Allows you to define indicators that use data from other indicators.

  • Indicators are calculated in the order defined by IndicatorDefinition.dependency_order, higher dependency order can read data from lower one. You usually specify this with indicator.add(order) argument.

  • If you add a parameter dependency_resolver to your indicator functions, the instance of this class is passed and you can use dependency_resolver to laod and read data from the past indicator calculations.

  • The indicator function still can take its usual values like close (for close price series), strategy_universe, etc. based on IndicatorSource, even if these values

    are not used in the calculations

An example where a single-pair indicator uses data from two other indicators:

def ma_crossover(
    close: pd.Series,
    pair: TradingPairIdentifier,
    dependency_resolver: IndicatorDependencyResolver,
) -> pd.Series:
    # Do cross-over calculation based on other two earlier moving average indicators.
    # Return pd.Series with True/False valeus and DatetimeIndex
    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

indicators = IndicatorSet()
# Slow moving average
indicators.add(
    "fast_sma",
    pandas_ta.sma,
    parameters={"length": 7},
    order=1,
)
# Fast moving average
indicators.add(
    "slow_sma",
    pandas_ta.sma,
    parameters={"length": 21},
    order=1,
)
# An indicator that depends on both fast MA and slow MA above
indicators.add(
    "ma_crossover",
    ma_crossover,
    source=IndicatorSource.ohlcv,
    order=2,  # 2nd order indicators can depend on the data of 1st order indicators
)

When you use indicator dependency resolution with the grid search, you need to specify indicator parameters you want read, as for each named indicator there might be multiple copies with different grid search combinations:

class Parameters:
    cycle_duration = CycleDuration.cycle_1d
    initial_cash = 10_000

    # Indicator values that are searched in the grid search
    slow_ema_candle_count = [21, 30]
    fast_ema_candle_count = [7, 12]
    combined_indicator_modes = [1, 2]

def combined_indicator(close: pd.Series, mode: int, pair: TradingPairIdentifier, dependency_resolver: IndicatorDependencyResolver):
    # An indicator that peeks the earlier grid search indicator calculations
    match mode:
        case 1:
            # When we look up data in grid search we need to give the parameter of which data we want,
            # and the trading pair if needed
            fast_ema = dependency_resolver.get_indicator_data("slow_ema", pair=pair, parameters={"length": 21})
            slow_ema = dependency_resolver.get_indicator_data("fast_ema", pair=pair, parameters={"length": 7})
        case 2:
            # Look up one set of parameters
            fast_ema = dependency_resolver.get_indicator_data("slow_ema", pair=pair, parameters={"length": 30})
            slow_ema = dependency_resolver.get_indicator_data("fast_ema", pair=pair, parameters={"length": 12})
        case _:
            raise NotImplementedError()

    return fast_ema * slow_ema * close # Calculate something based on two indicators and price

def create_indicators(parameters: StrategyParameters, indicators: IndicatorSet, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext):
    indicators.add("slow_ema", pandas_ta.ema, {"length": parameters.slow_ema_candle_count})
    indicators.add("fast_ema", pandas_ta.ema, {"length": parameters.fast_ema_candle_count})
    indicators.add(
        "combined_indicator",
        combined_indicator,
        {"mode": parameters.combined_indicator_modes},
        source=IndicatorSource.ohlcv,
        order=2,
    )

combinations = prepare_grid_combinations(
    Parameters,
    tmp_path,
    strategy_universe=strategy_universe,
    create_indicators=create_indicators,
    execution_context=ExecutionContext(mode=ExecutionMode.unit_testing, grid_search=True),
)
__init__(strategy_universe, all_indicators, indicator_storage, current_dependency_order=0)#
Parameters:
Return type:

None

Methods

__init__(strategy_universe, all_indicators, ...)

get_indicator_data(name[, column, pair, ...])

Read data from another indicator.

get_indicator_data_pairs_combined(name)

Get a DataFrame that contains indicator data for all pairs combined.

match_indicator(name[, pair, parameters])

Find an indicator key for an indicator.

Attributes

current_dependency_order

The current resolved dependency order level

strategy_universe

Trading universe

all_indicators

Available indicators as defined in create_indicators()

indicator_storage

Raw cached indicator results or ones calculated in the memory
strategy_universe: TradingStrategyUniverse#

Trading universe

  • Perform additional pair lookups if needed

all_indicators: set[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey]#

Available indicators as defined in create_indicators()

indicator_storage: IndicatorStorage#

Raw cached indicator results or ones calculated in the memory

current_dependency_order: int = 0#

The current resolved dependency order level

match_indicator(name, pair=None, parameters=None)[source]#

Find an indicator key for an indicator.

  • Get the IndicatorKey instance that is the look up for loading the indicator data

  • Check by name, pair and parameter

  • Make sure that the indicator defined is on a lower level than the current dependency order level

Parameters:
Return type:

IndicatorKey

get_indicator_data_pairs_combined(name)[source]#

Get a DataFrame that contains indicator data for all pairs combined.

  • Allows to access the indicator data for all pairs as a combined dataframe.

Example:

def regime(
    close: pd.Series,
    pair: TradingPairIdentifier,
    length: int,
    dependency_resolver: IndicatorDependencyResolver,
) -> pd.Series:
    fast_sma: pd.Series = dependency_resolver.get_indicator_data("fast_sma", pair=pair, parameters={"length": length})
    return close > fast_sma

def multipair(universe: TradingStrategyUniverse, dependency_resolver: IndicatorDependencyResolver) -> pd.DataFrame:
    # Test multipair data resolution
    series = dependency_resolver.get_indicator_data_pairs_combined("regime")
    assert isinstance(series.index, pd.MultiIndex)
    assert isinstance(series, pd.Series)
    return series
    # Change from pd.Series to pd.DataFrame with column "value"
    # df = series.to_frame(name='value')
    # assert df.columns == ["value"]
    # return df

def create_indicators(parameters: StrategyParameters, indicators: IndicatorSet, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext):
    indicators.add("regime", regime, {"length": parameters.fast_sma}, order=2)
    indicators.add("multipair", multipair, {}, IndicatorSource.strategy_universe, order=3)

Output:

pair_id  timestamp
1        2021-06-01    False
         2021-06-02    False
         2021-06-03    False
         2021-06-04    False
         2021-06-05    False
                       ...
2        2021-12-27     True
         2021-12-28     True
         2021-12-29    False
         2021-12-30    False
         2021-12-31    False
Parameters:

name (str) – An indicator that was previously calculated by its order.

Returns:

DataFrame with MultiIndex (pair_id, timestamp)

Return type:

Series

get_indicator_data(name, column=None, pair=None, parameters=None)[source]#

Read data from another indicator.

-The indicator must be prepared in create_indicators() earlier,

and then calculated

Parameters:

  • name (str) – Indicator name, as given to IndicatorSet.add().

  • parameters (dict | None) – If the dependent indicator has multiple versions with different parameters, we need to get the specify parameters.

  • column (str | None) –

    Column name for multi-column indicators.

    Use when the indicator is pandas.DataFrame instead of pandas.Series.

    Set to string all to get the whole DataFrame.

  • pair (Optional[Union[TradingPairIdentifier, Tuple[ChainId, str | None, str, str, float], Tuple[ChainId, str | None, str, str]]]) –

    Needed when universe contains multiple trading pairs.

    Can be omitted from non-pair indicators.

Returns:

The indicator data as is was saved on the disk

Return type:

pandas.core.series.Series | pandas.core.frame.DataFrame

__init__(strategy_universe, all_indicators, indicator_storage, current_dependency_order=0)#
Parameters:
Return type:

None