chore: improve typing in percentChange Pairlist

This commit is contained in:
Matthias
2024-12-11 07:25:07 +01:00
parent b9aa78b987
commit 34835752f8
2 changed files with 21 additions and 10 deletions

View File

@@ -8,7 +8,7 @@ defined period or as coming from ticker
import logging import logging
from datetime import timedelta from datetime import timedelta
from typing import Any from typing import TypedDict
from cachetools import TTLCache from cachetools import TTLCache
from pandas import DataFrame from pandas import DataFrame
@@ -24,6 +24,11 @@ from freqtrade.util import dt_now, format_ms_time
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class SymbolWithPercentage(TypedDict):
symbol: str
percentage: float | None
class PercentChangePairList(IPairList): class PercentChangePairList(IPairList):
is_pairlist_generator = True is_pairlist_generator = True
supports_backtesting = SupportsBacktesting.NO supports_backtesting = SupportsBacktesting.NO
@@ -211,10 +216,12 @@ class PercentChangePairList(IPairList):
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
:return: new whitelist :return: new whitelist
""" """
filtered_tickers: list[dict[str, Any]] = [{"symbol": k} for k in pairlist] filtered_tickers: list[SymbolWithPercentage] = [
{"symbol": k, "percentage": None} for k in pairlist
]
if self._use_range: if self._use_range:
# calculating using lookback_period # calculating using lookback_period
self.fetch_percent_change_from_lookback_period(filtered_tickers) filtered_tickers = self.fetch_percent_change_from_lookback_period(filtered_tickers)
else: else:
# Fetching 24h change by default from supported exchange tickers # Fetching 24h change by default from supported exchange tickers
filtered_tickers = self.fetch_percent_change_from_tickers(filtered_tickers, tickers) filtered_tickers = self.fetch_percent_change_from_tickers(filtered_tickers, tickers)
@@ -227,7 +234,7 @@ class PercentChangePairList(IPairList):
sorted_tickers = sorted( sorted_tickers = sorted(
filtered_tickers, filtered_tickers,
reverse=self._sort_direction == "desc", reverse=self._sort_direction == "desc",
key=lambda t: t["percentage"], key=lambda t: t["percentage"], # type: ignore
) )
# Validate whitelist to only have active market pairs # Validate whitelist to only have active market pairs
@@ -239,7 +246,7 @@ class PercentChangePairList(IPairList):
return pairs return pairs
def fetch_candles_for_lookback_period( def fetch_candles_for_lookback_period(
self, filtered_tickers: list[dict[str, str]] self, filtered_tickers: list[SymbolWithPercentage]
) -> dict[PairWithTimeframe, DataFrame]: ) -> dict[PairWithTimeframe, DataFrame]:
since_ms = ( since_ms = (
int( int(
@@ -275,7 +282,9 @@ class PercentChangePairList(IPairList):
candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms) candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms)
return candles return candles
def fetch_percent_change_from_lookback_period(self, filtered_tickers: list[dict[str, Any]]): def fetch_percent_change_from_lookback_period(
self, filtered_tickers: list[SymbolWithPercentage]
) -> list[SymbolWithPercentage]:
# get lookback period in ms, for exchange ohlcv fetch # get lookback period in ms, for exchange ohlcv fetch
candles = self.fetch_candles_for_lookback_period(filtered_tickers) candles = self.fetch_candles_for_lookback_period(filtered_tickers)
@@ -300,11 +309,12 @@ class PercentChangePairList(IPairList):
filtered_tickers[i]["percentage"] = pct_change filtered_tickers[i]["percentage"] = pct_change
else: else:
filtered_tickers[i]["percentage"] = 0 filtered_tickers[i]["percentage"] = 0
return filtered_tickers
def fetch_percent_change_from_tickers( def fetch_percent_change_from_tickers(
self, filtered_tickers: list[dict[str, Any]], tickers self, filtered_tickers: list[SymbolWithPercentage], tickers
) -> list[dict[str, Any]]: ) -> list[SymbolWithPercentage]:
valid_tickers: list[dict[str, Any]] = [] valid_tickers: list[SymbolWithPercentage] = []
for p in filtered_tickers: for p in filtered_tickers:
# Filter out assets # Filter out assets
if ( if (

View File

@@ -362,7 +362,8 @@ def test_gen_pairlist_from_tickers(mocker, rpl_config, tickers):
remote_pairlist = pairlistmanager._pairlist_handlers[0] remote_pairlist = pairlistmanager._pairlist_handlers[0]
# The generator returns BTC ETH and TKN - filtering the first ensures removing pairs in this step ain't problematic. # The generator returns BTC ETH and TKN - filtering the first ensures removing pairs
# in this step ain't problematic.
def _validate_pair(pair, ticker): def _validate_pair(pair, ticker):
if pair == "BTC/USDT": if pair == "BTC/USDT":
return False return False