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
For simple strategies calling
get_indicator_value()
should be only required here.
- __init__(strategy_universe, available_indicators, indicator_results, timestamp=None)#
- Parameters:
strategy_universe (TradingStrategyUniverse) –
available_indicators (IndicatorSet) –
indicator_results (dict[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey, tradeexecutor.strategy.pandas_trader.indicator.IndicatorResult]) –
- Return type:
None
Methods
__init__
(strategy_universe, ...[, timestamp])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_price
([pair, data_lag_tolerance])Read the available close price 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
Trading universe
Available indicators as defined in create_indicators()
Raw cached indicator results or ones calculated in the memory
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_price(pair=None, data_lag_tolerance=Timedelta('7 days 00:00:00'))[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
- 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.
- 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_indicator_value(name, column=None, pair=None, index=- 1, clock_shift=Timedelta('0 days 00:00:00'), data_delay_tolerance=None)[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.
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.
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 (Optional[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.
- 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
- resolve_indicator_data(name, column=None, pair=None)[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
- 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
- __init__(strategy_universe, available_indicators, indicator_results, timestamp=None)#
- Parameters:
strategy_universe (TradingStrategyUniverse) –
available_indicators (IndicatorSet) –
indicator_results (dict[tradeexecutor.strategy.pandas_trader.indicator.IndicatorKey, tradeexecutor.strategy.pandas_trader.indicator.IndicatorResult]) –
- Return type:
None