Merge pull request #9821 from freqtrade/feat/volumepairlist_caching

improve volumepairlist "advanced filter mode" caching

docs/includes/pairlists.md

@@ -68,7 +68,7 @@ When used in the leading position of the chain of Pairlist Handlers, the `pair_w
 
 The `refresh_period` setting allows to define the period (in seconds), at which the pairlist will be refreshed. Defaults to 1800s (30 minutes).
 The pairlist cache (`refresh_period`) on `VolumePairList` is only applicable to generating pairlists.
-Filtering instances (not the first position in the list) will not apply any cache and will always use up-to-date data.
+Filtering instances (not the first position in the list) will not apply any cache (beyond caching candles for the duration of the candle in advanced mode) and will always use up-to-date data.
 
 `VolumePairList` is per default based on the ticker data from exchange, as reported by the ccxt library:
 

freqtrade/exchange/exchange.py

@@ -43,6 +43,7 @@ from freqtrade.misc import (chunks, deep_merge_dicts, file_dump_json, file_load_
 from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist
 from freqtrade.util import dt_from_ts, dt_now
 from freqtrade.util.datetime_helpers import dt_humanize, dt_ts
+from freqtrade.util.periodic_cache import PeriodicCache
 
 
 logger = logging.getLogger(__name__)
@@ -131,6 +132,7 @@ class Exchange:
 
         # Holds candles
         self._klines: Dict[PairWithTimeframe, DataFrame] = {}
+        self._expiring_candle_cache: Dict[str, PeriodicCache] = {}
 
         # Holds all open sell orders for dry_run
         self._dry_run_open_orders: Dict[str, Any] = {}
@@ -2124,6 +2126,39 @@ class Exchange:
 
         return results_df
 
+    def refresh_ohlcv_with_cache(
+        self,
+        pairs: List[PairWithTimeframe],
+        since_ms: int
+    ) -> Dict[PairWithTimeframe, DataFrame]:
+        """
+        Refresh ohlcv data for all pairs in needed_pairs if necessary.
+        Caches data with expiring per timeframe.
+        Should only be used for pairlists which need "on time" expirarion, and no longer cache.
+        """
+
+        timeframes = [p[1] for p in pairs]
+        for timeframe in timeframes:
+            if timeframe not in self._expiring_candle_cache:
+                timeframe_in_sec = timeframe_to_seconds(timeframe)
+                # Initialise cache
+                self._expiring_candle_cache[timeframe] = PeriodicCache(ttl=timeframe_in_sec,
+                                                                       maxsize=1000)
+
+        # Get candles from cache
+        candles = {
+            c: self._expiring_candle_cache[c[1]].get(c, None) for c in pairs
+            if c in self._expiring_candle_cache[c[1]]
+        }
+        pairs_to_download = [p for p in pairs if p not in candles]
+        if pairs_to_download:
+            candles = self.refresh_latest_ohlcv(
+                pairs_to_download, since_ms=since_ms, cache=False
+            )
+            for c, val in candles.items():
+                self._expiring_candle_cache[c[1]][c] = val
+        return candles
+
     def _now_is_time_to_refresh(self, pair: str, timeframe: str, candle_type: CandleType) -> bool:
         # Timeframe in seconds
         interval_in_sec = timeframe_to_seconds(timeframe)

freqtrade/plugins/pairlist/VolatilityFilter.py

@@ -104,10 +104,7 @@ class VolatilityFilter(IPairList):
 
         since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days))
         # Get all candles
-        candles = {}
-        if needed_pairs:
-            candles = self._exchange.refresh_latest_ohlcv(needed_pairs, since_ms=since_ms,
-                                                          cache=False)
+        candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)
 
         if self._enabled:
             for p in deepcopy(pairlist):

freqtrade/plugins/pairlist/VolumePairList.py

@@ -229,12 +229,8 @@ class VolumePairList(IPairList):
                 if p not in self._pair_cache
             ]
 
-            # Get all candles
-            candles = {}
-            if needed_pairs:
-                candles = self._exchange.refresh_latest_ohlcv(
-                    needed_pairs, since_ms=since_ms, cache=False
-                )
+            candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms)
+
             for i, p in enumerate(filtered_tickers):
                 contract_size = self._exchange.markets[p['symbol']].get('contractSize', 1.0) or 1.0
                 pair_candles = candles[

freqtrade/plugins/pairlist/rangestabilityfilter.py

@@ -101,11 +101,7 @@ class RangeStabilityFilter(IPairList):
             (p, '1d', self._def_candletype) for p in pairlist if p not in self._pair_cache]
 
         since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days - 1))
-        # Get all candles
-        candles = {}
-        if needed_pairs:
-            candles = self._exchange.refresh_latest_ohlcv(needed_pairs, since_ms=since_ms,
-                                                          cache=False)
+        candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)
 
         if self._enabled:
             for p in deepcopy(pairlist):

tests/exchange/test_exchange.py

@@ -2303,6 +2303,66 @@ def test_refresh_latest_ohlcv_cache(mocker, default_conf, candle_type, time_mach
     assert res[pair2].at[0, 'open']
 
 
+def test_refresh_ohlcv_with_cache(mocker, default_conf, time_machine) -> None:
+    start = datetime(2021, 8, 1, 0, 0, 0, 0, tzinfo=timezone.utc)
+    ohlcv = generate_test_data_raw('1h', 100, start.strftime('%Y-%m-%d'))
+    time_machine.move_to(start, tick=False)
+    pairs = [
+        ('ETH/BTC', '1d', CandleType.SPOT),
+        ('TKN/BTC', '1d', CandleType.SPOT),
+        ('LTC/BTC', '1d', CandleType.SPOT),
+        ('LTC/BTC', '5m', CandleType.SPOT),
+        ('LTC/BTC', '1h', CandleType.SPOT),
+    ]
+
+    ohlcv_data = {
+        p: ohlcv for p in pairs
+    }
+    ohlcv_mock = mocker.patch(f"{EXMS}.refresh_latest_ohlcv", return_value=ohlcv_data)
+    mocker.patch(f"{EXMS}.ohlcv_candle_limit", return_value=100)
+    exchange = get_patched_exchange(mocker, default_conf)
+
+    assert len(exchange._expiring_candle_cache) == 0
+
+    res = exchange.refresh_ohlcv_with_cache(pairs, start.timestamp())
+    assert ohlcv_mock.call_count == 1
+    assert ohlcv_mock.call_args_list[0][0][0] == pairs
+    assert len(ohlcv_mock.call_args_list[0][0][0]) == 5
+
+    assert len(res) == 5
+    # length of 3 - as we have 3 different timeframes
+    assert len(exchange._expiring_candle_cache) == 3
+
+    ohlcv_mock.reset_mock()
+    res = exchange.refresh_ohlcv_with_cache(pairs, start.timestamp())
+    assert ohlcv_mock.call_count == 0
+
+    # Expire 5m cache
+    time_machine.move_to(start + timedelta(minutes=6), tick=False)
+
+    ohlcv_mock.reset_mock()
+    res = exchange.refresh_ohlcv_with_cache(pairs, start.timestamp())
+    assert ohlcv_mock.call_count == 1
+    assert len(ohlcv_mock.call_args_list[0][0][0]) == 1
+
+    # Expire 5m and 1h cache
+    time_machine.move_to(start + timedelta(hours=2), tick=False)
+
+    ohlcv_mock.reset_mock()
+    res = exchange.refresh_ohlcv_with_cache(pairs, start.timestamp())
+    assert ohlcv_mock.call_count == 1
+    assert len(ohlcv_mock.call_args_list[0][0][0]) == 2
+
+    # Expire all caches
+    time_machine.move_to(start + timedelta(days=1, hours=2), tick=False)
+
+    ohlcv_mock.reset_mock()
+    res = exchange.refresh_ohlcv_with_cache(pairs, start.timestamp())
+    assert ohlcv_mock.call_count == 1
+    assert len(ohlcv_mock.call_args_list[0][0][0]) == 5
+    assert ohlcv_mock.call_args_list[0][0][0] == pairs
+
+
 @pytest.mark.parametrize("exchange_name", EXCHANGES)
 async def test__async_get_candle_history(default_conf, mocker, caplog, exchange_name):
     ohlcv = [

tests/plugins/test_pairlist.py

@@ -621,13 +621,20 @@ def test_VolumePairList_whitelist_gen(mocker, whitelist_conf, shitcoinmarkets, t
     ([{"method": "VolumePairList", "number_assets": 5, "sort_key": "quoteVolume",
        "lookback_timeframe": "1d", "lookback_period": 6, "refresh_period": 86400}],
      "BTC", "binance", ['LTC/BTC', 'XRP/BTC', 'ETH/BTC', 'HOT/BTC', 'NEO/BTC']),
+    # VolumePairlist in range mode as filter.
+    # TKN/BTC is removed because it doesn't have enough candles
+    ([{"method": "VolumePairList", "number_assets": 5},
+      {"method": "VolumePairList", "number_assets": 5, "sort_key": "quoteVolume",
+       "lookback_timeframe": "1d", "lookback_period": 2, "refresh_period": 86400}],
+     "BTC", "binance", ['LTC/BTC', 'XRP/BTC', 'ETH/BTC', 'TKN/BTC', 'HOT/BTC']),
     # ftx data is already in Quote currency, therefore won't require conversion
     # ([{"method": "VolumePairList", "number_assets": 5, "sort_key": "quoteVolume",
     #    "lookback_timeframe": "1d", "lookback_period": 1, "refresh_period": 86400}],
     #  "BTC", "ftx", ['HOT/BTC', 'LTC/BTC', 'ETH/BTC', 'TKN/BTC', 'XRP/BTC']),
 ])
-def test_VolumePairList_range(mocker, whitelist_conf, shitcoinmarkets, tickers, ohlcv_history,
-                              pairlists, base_currency, exchange, volumefilter_result) -> None:
+def test_VolumePairList_range(
+        mocker, whitelist_conf, shitcoinmarkets, tickers, ohlcv_history,
+        pairlists, base_currency, exchange, volumefilter_result, time_machine) -> None:
     whitelist_conf['pairlists'] = pairlists
     whitelist_conf['stake_currency'] = base_currency
     whitelist_conf['exchange']['name'] = exchange
@@ -686,23 +693,36 @@ def test_VolumePairList_range(mocker, whitelist_conf, shitcoinmarkets, tickers,
             get_tickers=tickers,
             markets=PropertyMock(return_value=shitcoinmarkets)
         )
-
+        start_dt = dt_now()
+        time_machine.move_to(start_dt)
         # remove ohlcv when looback_timeframe != 1d
         # to enforce fallback to ticker data
         if 'lookback_timeframe' in pairlists[0]:
             if pairlists[0]['lookback_timeframe'] != '1d':
-                ohlcv_data = []
+                ohlcv_data = {}
 
-        mocker.patch.multiple(
-            EXMS,
-            refresh_latest_ohlcv=MagicMock(return_value=ohlcv_data),
-        )
+        ohclv_mock = mocker.patch(f"{EXMS}.refresh_latest_ohlcv", return_value=ohlcv_data)
 
         freqtrade.pairlists.refresh_pairlist()
         whitelist = freqtrade.pairlists.whitelist
+        assert ohclv_mock.call_count == 1
 
         assert isinstance(whitelist, list)
         assert whitelist == volumefilter_result
+        # Test caching
+        ohclv_mock.reset_mock()
+        freqtrade.pairlists.refresh_pairlist()
+        # in "filter" mode, caching is disabled.
+        assert ohclv_mock.call_count == 0
+        whitelist = freqtrade.pairlists.whitelist
+        assert whitelist == volumefilter_result
+
+        time_machine.move_to(start_dt + timedelta(days=2))
+        ohclv_mock.reset_mock()
+        freqtrade.pairlists.refresh_pairlist()
+        assert ohclv_mock.call_count == 1
+        whitelist = freqtrade.pairlists.whitelist
+        assert whitelist == volumefilter_result
 
 
 def test_PrecisionFilter_error(mocker, whitelist_conf) -> None: