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 valuesare 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:
strategy_universe (TradingStrategyUniverse) –
all_indicators (set[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey]) –
indicator_storage (IndicatorStorage) –
current_dependency_order (int) –
- Return type:
None
Methods
__init__
(strategy_universe, all_indicators, ...)get_indicator_data
(name[, column, pair, ...])Read data from another indicator.
Get a DataFrame that contains indicator data for all pairs combined.
match_indicator
(name[, pair, parameters])Find an indicator key for an indicator.
Attributes
The current resolved dependency order level
Trading universe
Available indicators as defined in create_indicators()
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
- 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
- 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
- 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 ofpandas.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:
strategy_universe (TradingStrategyUniverse) –
all_indicators (set[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey]) –
indicator_storage (IndicatorStorage) –
current_dependency_order (int) –
- Return type:
None