StrategyInputIndicators#

API documentation for tradeexecutor.strategy.pandas_trader.strategy_input.StrategyInputIndicators Python class in Trading Strategy framework.

class StrategyInputIndicators[source]#

Bases: object

Indicator results for the strategy decision.

A helper class to read and manipulate indicator and price values. Thi class wraps the indicator results, both cached and real-time, to a format that has good developer experience when accessed from decide_trades().

  • Indicators are prepared in create_indicators function

  • The framework takes care of recalculating indicators when needed, for backtest and live access

  • For backtests, this class is instiated only once

  • We assume all indicator data is forward-filled and no gaps

How to use

__init__(strategy_universe, available_indicators, indicator_results, timestamp=None)#
Parameters:
Return type:

None

Methods

__init__(strategy_universe, ...[, timestamp])

get_indicator_data_pairs_combined(name[, ...])

Get indicator data for all pairs in a single blob.

get_indicator_dataframe(name[, pair])

Get the whole raw indicator data for DataFrame-like indicator with multiple columns.

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

Get the whole indicator data series.

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

Read the available value of an indicator.

get_ohlcv([pair])

Get full OHLCV price feed for a trading pair.

get_price([pair, data_lag_tolerance, index, ...])

Read the available close price of a trading pair.

get_price_series([column, pair])

Get the whole price series.

get_recent_price_series([column, pair])

Similar to get_price_series, but only returns the past data.

get_tvl([pair, data_lag_tolerance, index, ...])

Read the available TVL of a trading pair.

prepare_decision_cycle(cycle, timestamp)

Called for each decision cycle by the framework..

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

Get access to indicator data series/frame.

Attributes

strategy_universe

Trading universe

available_indicators

Available indicators as defined in create_indicators()

indicator_results

Raw cached indicator results or ones calculated in the memory

timestamp

The current decision_cycle() timestamp.
strategy_universe: TradingStrategyUniverse#

Trading universe

  • Perform additional pair lookups if needed

available_indicators: IndicatorSet#

Available indicators as defined in create_indicators()

indicator_results: dict[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey, tradeexecutor.strategy.pandas_trader.indicator.IndicatorResult]#

Raw cached indicator results or ones calculated in the memory

timestamp: pandas._libs.tslibs.timestamps.Timestamp | None#

The current decision_cycle() timestamp.

Stored here, so we do not need to pass it explicitly in API.

get_ohlcv(pair=None)[source]#

Get full OHLCV price feed for a trading pair.

Returns:

DataFrame with open, high, low, close, volume columns.

Parameters:

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

Return type:

DataFrame

get_price(pair=None, data_lag_tolerance=Timedelta('7 days 00:00:00'), index=- 1, timestamp=None, column='close')[source]#

Read the available close price of a trading pair.

  • Returns the latest available close price.

  • Does not return the current price in the decision_cycle, because any decision must be made based on the previous price to avoid lookahead bias.

Parameters:

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

    The trading pair for which we query the price.

    Give as id object or human description tuple format.

    E.g. (ChainId.centralised_exchange, “binance”, “ETH”, “USDT”).

  • data_lag_tolerance – In the case the data has issues (no recent price), then accept a price that’s this old.

  • index (int) –

    Access a specific previous timeframe item.

    If not given, always return the previous available value. Timeframe = candle bar here.

    Uses Python list access notation. - -1 is the last item (previous time frame value, yesterday). - -2 is the item before previous time frame (the day before yesterday). - 0 is looking to the future (the value at the end of the current day that has not yet passed)

  • timestamp (pandas._libs.tslibs.timestamps.Timestamp | None) –

    Look price at a specific timestamp.

    Manually calculate lookback. There is no timeshift for this value, so unless you are careful you may case lookahead bias.

    index parameter is ignored.

  • column –

    Which column to read from the price series.

    E.g. “volume”.

Returns:

The latest available price.

None if no price information is yet available at this point of time for the strategy.

Return type:

float | None

get_tvl(pair=None, data_lag_tolerance=Timedelta('7 days 00:00:00'), index=- 1, timestamp=None)[source]#

Read the available TVL of a trading pair.

  • Returns the latest available TVL/liquidity sample.

  • Does not return the current liquidity in the decision_cycle, because any decision must be made based on the previous price to avoid lookahead bias.

See also get_price()

Parameters:

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

    The trading pair for which we query the price.

    Give as id object or human description tuple format.

    E.g. (ChainId.centralised_exchange, “binance”, “ETH”, “USDT”).

  • data_lag_tolerance – In the case the data has issues (no recent price), then accept a price that’s this old.

  • index (int) –

    Access a specific previous timeframe item.

    If not given, always return the previous available value. Timeframe = candle bar here.

    Uses Python list access notation. - -1 is the last item (previous time frame value, yesterday). - -2 is the item before previous time frame (the day before yesterday). - 0 is looking to the future (the value at the end of the current day that has not yet passed)

  • timestamp (pandas._libs.tslibs.timestamps.Timestamp | None) –

    Look price at a specific timestamp.

    Manually calculate lookback. There is no timeshift for this value, so unless you are careful you may case lookahead bias.

    index parameter is ignored.

Returns:

The latest available TVL.

None if no price information is yet available at this point of time for the strategy.

Return type:

float | None

get_indicator_value(name, column=None, pair=None, index=- 1, clock_shift=Timedelta('0 days 00:00:00'), data_delay_tolerance='auto')[source]#

Read the available value of an indicator.

  • Returns the latest available indicator value.

  • Does not return the current timestamp value in the decision_cycle, because any decision must be made based on the previous price.

  • Normalises missing inputs, NaNs and other data issues to Python None.

Single pair example with a single series indicator (RSI):

def create_indicators(parameters: StrategyParameters, indicators: IndicatorSet, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext):
    indicators.add("rsi", pandas_ta.rsi, {"length": parameters.rsi_length})

#
# Then in decide_traces()
#

# Read the RSI value of our only trading pair
indicator_value = input.indicators.get_indicator_value("rsi")

Single pair example with a multi-series indicator (Bollinger band):

def create_indicators(parameters: StrategyParameters, indicators: IndicatorSet, strategy_universe: TradingStrategyUniverse, execution_context: ExecutionContext):
    indicators.add("bb", pandas_ta.bbands, {"length": parameters.bb_length})

#
# Then in decide_traces()
#

# Read bollinger band value for the current trading pair.
# Bollinger band look up length was 20 and standard deviation 2.0.
bb_value = input.indicators.get_indicator_value("bb", "BBL_20_2.0")

Example accessing latest and previous values for cross over test:

current_rsi_values[pair] = indicators.get_indicator_value("rsi", pair=pair)
previous_rsi_values[pair] = indicators.get_indicator_value("rsi", index=-2, pair=pair)

# Check for RSI crossing our threshold values in this cycle, compared to the previous cycle
if current_rsi_values[pair] and previous_rsi_values[pair]:
    rsi_cross_above = current_rsi_values[pair] >= parameters.rsi_high and previous_rsi_values[btc_pair] < parameters.rsi_high
    rsi_cross_below = current_rsi_values[pair] < parameters.rsi_low and previous_rsi_values[pair] > parameters.rsi_low
Parameters:

  • name (str) – Indicator name as defined in create_indicators.

  • column (str | None) –

    The name of the sub-column to read.

    For multicolumn indicators like Bollinger Bands, which produce multiple series of data from one column of price data.

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

    Trading pair.

    Must be given if the working with a multipair strategy.

  • index (int) –

    Access a specific previous timeframe item.

    If not given, always return the previous available value. Timeframe = candle bar here.

    Uses Python list access notation. - -1 is the last item (previous time frame value, yesterday). - -2 is the item before previous time frame (the day before yesterday). - 0 is looking to the future (the value at the end of the current day that has not yet passed)

  • clock_shift (Timedelta) – Used in time-shifted backtesting.

  • data_delay_tolerance (Timedelta) –

    If we do not have an exact timestamp match in the data series, look for the previous value.

    Look back max data_delay_tolerance days / hours to get a previous value using forward-fill technique.

    We need to do this when there is a mismatch between the indicator timeframe (e.g. daily) and decision cycle / price time frame (e.g. 15 minutes).

    Set to None to always return indicator value for the exact timestamp match.

    Set to `auto to try to figure out mismatch between indicator data and candle data automatically.s

Returns:

The latest available indicator value.

Any NaN, NA or not a number value in the indicator data is translated to Python None.

Return None if value not yet available when asked at the current decision moment.

Raises:

IndicatorDataNotFoundWithinDataTolerance – We asked data_delay_tolerance look backwards, but there wasn’t any samples within the tolerance.

Return type:

float | None

get_indicator_series(name, column=None, pair=None, unlimited=False)[source]#

Get the whole indicator data series.

By default, return data that is only available before the current timestamp.

Parameters:
Returns:

Indicator data.

Data may contain NaN values.

Return None if any data is not yet available before this stamp.

Return type:

pandas.core.series.Series | None

get_price_series(column='close', pair=None)[source]#

Get the whole price series.

  • Use for visualisation and other checks

  • Not useful inside decide_trades, as includes future data

Parameters:
Returns:

Indicator data.

Data may contain NaN values.

Return type:

Series

get_recent_price_series(column='close', pair=None)[source]#

Similar to get_price_series, but only returns the past data.

Parameters:
Returns:

Price series up to the current timestamp.

get_indicator_dataframe(name, pair=None)[source]#

Get the whole raw indicator data for DataFrame-like indicator with multiple columns.

See also get_indicator_series()

Returns:

DataFrame for a multicolumn indicator like Bollinger Bands or ADX

Parameters:
Return type:

DataFrame

resolve_indicator_data(name, column=None, pair=None, unlimited=False)[source]#

Get access to indicator data series/frame.

Throw friendly error messages for pitfalls.

Parameters:
Return type:

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

get_indicator_data_pairs_combined(name, pair_filter=None)[source]#

Get indicator data for all pairs in a single blob.

Example code:

class Parameters:

rsi_length = 20

def create_indicators(

timestamp, parameters, strategy_universe, execution_context,

) -> IndicatorSet:

indicator_set = IndicatorSet() indicator_set.add(“rsi”, pandas_ta.rsi, {“length”: parameters.rsi_length}) return indicator_set

# Calculate indicators - will spawn multiple worker processed, # or load cached results from the disk indicators = calculate_and_load_indicators_inline(

strategy_universe=strategy_universe, parameters=StrategyParameters.from_class(Parameters), create_indicators=create_indicators,

)

series = indicators.get_indicator_data_pairs_combined(“rsi”) print(series)

assert isinstance(series, pd.Series) assert isinstance(series.index, pd.MultiIndex)

pair_ids = series.index.get_level_values(“pair_id”) assert list(pair_ids.unique()) == [1, 2]

Output:

pair_id  timestamp
1        2021-06-01          NaN
         2021-06-02          NaN
         2021-06-03          NaN
         2021-06-04          NaN
         2021-06-05          NaN
                         ...
2        2021-12-27    54.956639
         2021-12-28    54.843694
         2021-12-29    48.713734
         2021-12-30    45.456219
         2021-12-31    49.308201
Parameters:

  • name (str) – Indicator name. Must be pd.Series based data.

  • pair_filter (Optional[Callable]) –

    Function pair_filter(TradingPairIdentifier).

    If give only include trading pairs for which this function returns True.

Returns:

Series data with (pair_id, timestamp) index.

Return type:

Series

prepare_decision_cycle(cycle, timestamp)[source]#

Called for each decision cycle by the framework..

  • Instead of making a copy of this data structure each time, we just bump the timestamp

Parameters:

  • cycle (int) –

  • timestamp (Timestamp) –

__init__(strategy_universe, available_indicators, indicator_results, timestamp=None)#
Parameters:
Return type:

None