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]) –
timestamp (pandas._libs.tslibs.timestamps.Timestamp | None) –
- 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
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'), 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
- 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()
- 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:
name (str) – Indicator name
column (str | None) –
Column name for multi-column indicators.
”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.
unlimited – Allow loading of past and future data.
- 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
- 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]) –
timestamp (pandas._libs.tslibs.timestamps.Timestamp | None) –
- Return type:
None