Merged feat/short into lev-strat

tests/strategy/strats/informative_decorator_strategy.py

@@ -0,0 +1,75 @@
+# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
+
+from pandas import DataFrame
+
+from freqtrade.strategy import informative, merge_informative_pair
+from freqtrade.strategy.interface import IStrategy
+
+
+class InformativeDecoratorTest(IStrategy):
+    """
+    Strategy used by tests freqtrade bot.
+    Please do not modify this strategy, it's  intended for internal use only.
+    Please look at the SampleStrategy in the user_data/strategy directory
+    or strategy repository https://github.com/freqtrade/freqtrade-strategies
+    for samples and inspiration.
+    """
+    INTERFACE_VERSION = 2
+    stoploss = -0.10
+    timeframe = '5m'
+    startup_candle_count: int = 20
+
+    def informative_pairs(self):
+        return [('BTC/USDT', '5m')]
+
+    def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['buy'] = 0
+        return dataframe
+
+    def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['sell'] = 0
+        return dataframe
+
+    # Decorator stacking test.
+    @informative('30m')
+    @informative('1h')
+    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['rsi'] = 14
+        return dataframe
+
+    # Simple informative test.
+    @informative('1h', 'BTC/{stake}')
+    def populate_indicators_btc_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['rsi'] = 14
+        return dataframe
+
+    # Quote currency different from stake currency test.
+    @informative('1h', 'ETH/BTC')
+    def populate_indicators_eth_btc_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['rsi'] = 14
+        return dataframe
+
+    # Formatting test.
+    @informative('30m', 'BTC/{stake}', '{column}_{BASE}_{QUOTE}_{base}_{quote}_{asset}_{timeframe}')
+    def populate_indicators_btc_1h_2(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['rsi'] = 14
+        return dataframe
+
+    # Custom formatter test
+    @informative('30m', 'ETH/{stake}', fmt=lambda column, **kwargs: column + '_from_callable')
+    def populate_indicators_eth_30m(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        dataframe['rsi'] = 14
+        return dataframe
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        # Strategy timeframe indicators for current pair.
+        dataframe['rsi'] = 14
+        # Informative pairs are available in this method.
+        dataframe['rsi_less'] = dataframe['rsi'] < dataframe['rsi_1h']
+
+        # Mixing manual informative pairs with decorators.
+        informative = self.dp.get_pair_dataframe('BTC/USDT', '5m')
+        informative['rsi'] = 14
+        dataframe = merge_informative_pair(dataframe, informative, self.timeframe, '5m', ffill=True)
+
+        return dataframe

tests/strategy/test_interface.py

@@ -648,7 +648,7 @@ def test_is_informative_pairs_callback(default_conf):
     strategy = StrategyResolver.load_strategy(default_conf)
     # Should return empty
     # Uses fallback to base implementation
-    assert [] == strategy.informative_pairs()
+    assert [] == strategy.gather_informative_pairs()
 
 
 @pytest.mark.parametrize('error', [

tests/strategy/test_strategy_helpers.py

@@ -4,7 +4,9 @@ import numpy as np
 import pandas as pd
 import pytest
 
-from freqtrade.strategy import merge_informative_pair, stoploss_from_open, timeframe_to_minutes
+from freqtrade.data.dataprovider import DataProvider
+from freqtrade.strategy import (merge_informative_pair, stoploss_from_absolute, stoploss_from_open,
+                                timeframe_to_minutes)
 
 
 def generate_test_data(timeframe: str, size: int):
@@ -132,3 +134,65 @@ def test_stoploss_from_open():
                         assert stoploss == 0
                     else:
                         assert isclose(stop_price, expected_stop_price, rel_tol=0.00001)
+
+
+def test_stoploss_from_absolute():
+    assert stoploss_from_absolute(90, 100) == 1 - (90 / 100)
+    assert stoploss_from_absolute(100, 100) == 0
+    assert stoploss_from_absolute(110, 100) == 0
+    assert stoploss_from_absolute(100, 0) == 1
+    assert stoploss_from_absolute(0, 100) == 1
+
+
+def test_informative_decorator(mocker, default_conf):
+    test_data_5m = generate_test_data('5m', 40)
+    test_data_30m = generate_test_data('30m', 40)
+    test_data_1h = generate_test_data('1h', 40)
+    data = {
+        ('XRP/USDT', '5m'): test_data_5m,
+        ('XRP/USDT', '30m'): test_data_30m,
+        ('XRP/USDT', '1h'): test_data_1h,
+        ('LTC/USDT', '5m'): test_data_5m,
+        ('LTC/USDT', '30m'): test_data_30m,
+        ('LTC/USDT', '1h'): test_data_1h,
+        ('BTC/USDT', '30m'): test_data_30m,
+        ('BTC/USDT', '5m'): test_data_5m,
+        ('BTC/USDT', '1h'): test_data_1h,
+        ('ETH/USDT', '1h'): test_data_1h,
+        ('ETH/USDT', '30m'): test_data_30m,
+        ('ETH/BTC', '1h'): test_data_1h,
+    }
+    from .strats.informative_decorator_strategy import InformativeDecoratorTest
+    default_conf['stake_currency'] = 'USDT'
+    strategy = InformativeDecoratorTest(config=default_conf)
+    strategy.dp = DataProvider({}, None, None)
+    mocker.patch.object(strategy.dp, 'current_whitelist', return_value=[
+        'XRP/USDT', 'LTC/USDT', 'BTC/USDT'
+    ])
+
+    assert len(strategy._ft_informative) == 6   # Equal to number of decorators used
+    informative_pairs = [('XRP/USDT', '1h'), ('LTC/USDT', '1h'), ('XRP/USDT', '30m'),
+                         ('LTC/USDT', '30m'), ('BTC/USDT', '1h'), ('BTC/USDT', '30m'),
+                         ('BTC/USDT', '5m'), ('ETH/BTC', '1h'), ('ETH/USDT', '30m')]
+    for inf_pair in informative_pairs:
+        assert inf_pair in strategy.gather_informative_pairs()
+
+    def test_historic_ohlcv(pair, timeframe):
+        return data[(pair, timeframe or strategy.timeframe)].copy()
+    mocker.patch('freqtrade.data.dataprovider.DataProvider.historic_ohlcv',
+                 side_effect=test_historic_ohlcv)
+
+    analyzed = strategy.advise_all_indicators(
+        {p: data[(p, strategy.timeframe)] for p in ('XRP/USDT', 'LTC/USDT')})
+    expected_columns = [
+        'rsi_1h', 'rsi_30m',                    # Stacked informative decorators
+        'btc_usdt_rsi_1h',                      # BTC 1h informative
+        'rsi_BTC_USDT_btc_usdt_BTC/USDT_30m',   # Column formatting
+        'rsi_from_callable',                    # Custom column formatter
+        'eth_btc_rsi_1h',                       # Quote currency not matching stake currency
+        'rsi', 'rsi_less',                      # Non-informative columns
+        'rsi_5m',                               # Manual informative dataframe
+    ]
+    for _, dataframe in analyzed.items():
+        for col in expected_columns:
+            assert col in dataframe.columns

tests/strategy/test_strategy_loading.py

@@ -35,7 +35,7 @@ def test_search_all_strategies_no_failed():
     directory = Path(__file__).parent / "strats"
     strategies = StrategyResolver.search_all_objects(directory, enum_failed=False)
     assert isinstance(strategies, list)
-    assert len(strategies) == 3
+    assert len(strategies) == 4
     assert isinstance(strategies[0], dict)
 
 
@@ -43,10 +43,10 @@ def test_search_all_strategies_with_failed():
     directory = Path(__file__).parent / "strats"
     strategies = StrategyResolver.search_all_objects(directory, enum_failed=True)
     assert isinstance(strategies, list)
-    assert len(strategies) == 4
+    assert len(strategies) == 5
     # with enum_failed=True search_all_objects() shall find 2 good strategies
     # and 1 which fails to load
-    assert len([x for x in strategies if x['class'] is not None]) == 3
+    assert len([x for x in strategies if x['class'] is not None]) == 4
     assert len([x for x in strategies if x['class'] is None]) == 1