CreateIndicatorsProtocol#

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

class CreateIndicatorsProtocol[source]#

Bases: Protocol

Call signature for create_indicators function.

This Protocol class defines create_indicators() function call signature. Strategy modules and backtests can provide on create_indicators function to define what indicators a strategy needs. Used with :py:class`IndicatorSet` to define the indicators the strategy can use.

These indicators are precalculated and cached for fast performance.

Example for a grid search:

class MyParameters:
    stop_loss_pct = [0.9, 0.95]
    cycle_duration = CycleDuration.cycle_1d
    initial_cash = 10_000

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


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 can be custom, and do not need to be calculated per trading pair. Here is an example of creating indicators “ETH/BTC price” and “ETC/BTC price RSI with length of 20 bars”:

def calculate_eth_btc(strategy_universe: TradingStrategyUniverse):
    weth_usdc = strategy_universe.get_pair_by_human_description((ChainId.ethereum, "test-dex", "WETH", "USDC"))
    wbtc_usdc = strategy_universe.get_pair_by_human_description((ChainId.ethereum, "test-dex", "WBTC", "USDC"))
    btc_price = strategy_universe.data_universe.candles.get_candles_by_pair(wbtc_usdc.internal_id)
    eth_price = strategy_universe.data_universe.candles.get_candles_by_pair(weth_usdc.internal_id)
    series = eth_price["close"] / btc_price["close"]  # Divide two series
    return series

def calculate_eth_btc_rsi(strategy_universe: TradingStrategyUniverse, length: int):
    weth_usdc = strategy_universe.get_pair_by_human_description((ChainId.ethereum, "test-dex", "WETH", "USDC"))
    wbtc_usdc = strategy_universe.get_pair_by_human_description((ChainId.ethereum, "test-dex", "WBTC", "USDC"))
    btc_price = strategy_universe.data_universe.candles.get_candles_by_pair(wbtc_usdc.internal_id)
    eth_price = strategy_universe.data_universe.candles.get_candles_by_pair(weth_usdc.internal_id)
    eth_btc = eth_price["close"] / btc_price["close"]
    return pandas_ta.rsi(eth_btc, length=length)

def create_indicators(parameters: StrategyParameters, indicators: IndicatorSet, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext):
    indicators.add("eth_btc", calculate_eth_btc, source=IndicatorSource.strategy_universe)
    indicators.add("eth_btc_rsi", calculate_eth_btc_rsi, parameters={"length": parameters.eth_btc_rsi_length}, source=IndicatorSource.strategy_universe)
__init__(*args, **kwargs)#

Methods

__init__(*args, **kwargs)

__call__(parameters, indicators, strategy_universe, execution_context)[source]#

Build technical indicators for the strategy.

Parameters:

  • parameters (StrategyParameters) –

    Passed from the backtest / live strategy parametrs.

    If doing a grid search, each paramter is simplified.

  • indicators (IndicatorSet) –

    Indicator builder helper class.

    Call IndicatorBuilder.create() to add new indicators to the strategy.

  • strategy_universe (TradingStrategyUniverse) –

    The loaded strategy universe.

    Use to resolve symbolic pair information if needed

  • execution_context (ExecutionContext) – Information about if this is a live or backtest run.

Returns:

This function does not return anything.

Instead indicators.add is used to attach new indicators to the strategy.

__init__(*args, **kwargs)#