CreateIndicatorsProtocolV2#

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

class CreateIndicatorsProtocolV2[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. These indicators are precalculated and cached for fast performance.

  • There are multiple indicator types, depending on if they are calculated on pair close price, pair OHLCV data or the whole strategy universe. See IndicatorSource.

  • Uses :py:class`IndicatorSet` class to construct the indicators the strategy can use.

  • To read indicator values in decide_trades() function, see StrategyInputIndicators.

  • For most pandas_ta functions. like pandas_ta.ma, pandas_ta.rsi, pandas_ta.mfi, you can pass them directly to indicators.add() - as those functions have standard argument names like close, high, low that are data series provided.

  • If you wish to use data from earlier indicators calculations in later indicator calculations, see IndicatorDependencyResolver for how to do it

Note

For new examples of defining indicators, please see tradeexecutor.strategy.pandas_trader.indicator_decorator. The examples below work too, but with Python decorator-based syntax you can make the code more reaadable and shorter.

Example for creating an Exponential Moving Average (EMA) indicator based on the close price. This example is for a grid search. Unless specified, indicators are assumed to be IndicatorSource.close_price type and they only use trading pair close price as input.

class Parameters:
    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(
    timestamp: datetime.datetime | None,
    parameters: StrategyParameters,
    strategy_universe: TradingStrategyUniverse,
    execution_context: ExecutionContext
):
    indicators = IndicatorSet()
    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})
    return indicators

Some indicators may use multiple OHLCV datapoints. In this case, you need to tell the indicator to be IndicatorSource.ohlcv type. Here is an example for Money Flow Index (MFI) indicator:

import pandas_ta

from tradeexecutor.strategy.parameters import StrategyParameters
from tradeexecutor.strategy.pandas_trader.indicator import IndicatorSet, IndicatorSource

class Parameters:
    my_mfi_length = 20

def create_indicators(
    timestamp: datetime.datetime | None,
    parameters: StrategyParameters,
    strategy_universe: TradingStrategyUniverse,
    execution_context: ExecutionContext
):
    indicators = IndicatorSet()
    indicators.add(
        "mfi",
        pandas_ta.mfi,
        parameters={"length": parameters.my_mfi_length},
        source=IndicatorSource.ohlcv,
    )

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, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext) -> IndicatorSet:
    indicators = IndicatorSet()
    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)
    return indicators

This protocol class is second (v2) iteration of the function signature.

__init__(*args, **kwargs)#

Methods

__init__(*args, **kwargs)

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

Build technical indicators for the strategy.

Parameters:

  • timestamp (datetime.datetime | None) –

    The current live execution timestamp.

    Set None for backtesting, as create_indicators() is called only once during the backtest setup.

  • parameters (StrategyParameters) –

    Passed from the backtest / live strategy parametrs.

    If doing a grid search, each paramter is simplified.

  • 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:

Indicators the strategy is going to need.

Return type:

IndicatorSet

__init__(*args, **kwargs)#