chore: update plugins to modern typing syntax

This commit is contained in:
Matthias
2024-10-04 07:02:20 +02:00
parent e9a6ba03f9
commit d8e41fa8b1
24 changed files with 116 additions and 122 deletions
+9 -9
View File
@@ -1,7 +1,7 @@
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Dict, List, Optional from typing import Optional
import pandas as pd import pandas as pd
@@ -43,7 +43,7 @@ except ImportError:
exit(1) exit(1)
def init_plotscript(config, markets: List, startup_candles: int = 0): def init_plotscript(config, markets: list, startup_candles: int = 0):
""" """
Initialize objects needed for plotting Initialize objects needed for plotting
:return: Dict with candle (OHLCV) data, trades and pairs :return: Dict with candle (OHLCV) data, trades and pairs
@@ -103,7 +103,7 @@ def init_plotscript(config, markets: List, startup_candles: int = 0):
} }
def add_indicators(fig, row, indicators: Dict[str, Dict], data: pd.DataFrame) -> make_subplots: def add_indicators(fig, row, indicators: dict[str, dict], data: pd.DataFrame) -> make_subplots:
""" """
Generate all the indicators selected by the user for a specific row, based on the configuration Generate all the indicators selected by the user for a specific row, based on the configuration
:param fig: Plot figure to append to :param fig: Plot figure to append to
@@ -301,8 +301,8 @@ def plot_trades(fig, trades: pd.DataFrame) -> make_subplots:
def create_plotconfig( def create_plotconfig(
indicators1: List[str], indicators2: List[str], plot_config: Dict[str, Dict] indicators1: list[str], indicators2: list[str], plot_config: dict[str, dict]
) -> Dict[str, Dict]: ) -> dict[str, dict]:
""" """
Combines indicators 1 and indicators 2 into plot_config if necessary Combines indicators 1 and indicators 2 into plot_config if necessary
:param indicators1: List containing Main plot indicators :param indicators1: List containing Main plot indicators
@@ -434,9 +434,9 @@ def generate_candlestick_graph(
data: pd.DataFrame, data: pd.DataFrame,
trades: Optional[pd.DataFrame] = None, trades: Optional[pd.DataFrame] = None,
*, *,
indicators1: Optional[List[str]] = None, indicators1: Optional[list[str]] = None,
indicators2: Optional[List[str]] = None, indicators2: Optional[list[str]] = None,
plot_config: Optional[Dict[str, Dict]] = None, plot_config: Optional[dict[str, dict]] = None,
) -> go.Figure: ) -> go.Figure:
""" """
Generate the graph from the data generated by Backtesting or from DB Generate the graph from the data generated by Backtesting or from DB
@@ -521,7 +521,7 @@ def generate_candlestick_graph(
def generate_profit_graph( def generate_profit_graph(
pairs: str, pairs: str,
data: Dict[str, pd.DataFrame], data: dict[str, pd.DataFrame],
trades: pd.DataFrame, trades: pd.DataFrame,
timeframe: str, timeframe: str,
stake_currency: str, stake_currency: str,
+4 -4
View File
@@ -5,7 +5,7 @@ Minimum age (days listed) pair list filter
import logging import logging
from copy import deepcopy from copy import deepcopy
from datetime import timedelta from datetime import timedelta
from typing import Dict, List, Optional from typing import Optional
from pandas import DataFrame from pandas import DataFrame
@@ -27,7 +27,7 @@ class AgeFilter(IPairList):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
# Checked symbols cache (dictionary of ticker symbol => timestamp) # Checked symbols cache (dictionary of ticker symbol => timestamp)
self._symbolsChecked: Dict[str, int] = {} self._symbolsChecked: dict[str, int] = {}
self._symbolsCheckFailed = PeriodicCache(maxsize=1000, ttl=86_400) self._symbolsCheckFailed = PeriodicCache(maxsize=1000, ttl=86_400)
self._min_days_listed = self._pairlistconfig.get("min_days_listed", 10) self._min_days_listed = self._pairlistconfig.get("min_days_listed", 10)
@@ -78,7 +78,7 @@ class AgeFilter(IPairList):
return "Filter pairs by age (days listed)." return "Filter pairs by age (days listed)."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"min_days_listed": { "min_days_listed": {
"type": "number", "type": "number",
@@ -94,7 +94,7 @@ class AgeFilter(IPairList):
}, },
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
:param pairlist: pairlist to filter or sort :param pairlist: pairlist to filter or sort
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -3,7 +3,6 @@ Full trade slots pair list filter
""" """
import logging import logging
from typing import List
from freqtrade.exchange.exchange_types import Tickers from freqtrade.exchange.exchange_types import Tickers
from freqtrade.persistence import Trade from freqtrade.persistence import Trade
@@ -35,7 +34,7 @@ class FullTradesFilter(IPairList):
def description() -> str: def description() -> str:
return "Shrink whitelist when trade slots are full." return "Shrink whitelist when trade slots are full."
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the allowlist again. Filters and sorts pairlist and returns the allowlist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+13 -13
View File
@@ -6,7 +6,7 @@ import logging
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from copy import deepcopy from copy import deepcopy
from enum import Enum from enum import Enum
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union from typing import Any, Literal, Optional, TypedDict, Union
from freqtrade.constants import Config from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
@@ -36,12 +36,12 @@ class __StringPairlistParameter(__PairlistParameterBase):
class __OptionPairlistParameter(__PairlistParameterBase): class __OptionPairlistParameter(__PairlistParameterBase):
type: Literal["option"] type: Literal["option"]
default: Union[str, None] default: Union[str, None]
options: List[str] options: list[str]
class __ListPairListParamenter(__PairlistParameterBase): class __ListPairListParamenter(__PairlistParameterBase):
type: Literal["list"] type: Literal["list"]
default: Union[List[str], None] default: Union[list[str], None]
class __BoolPairlistParameter(__PairlistParameterBase): class __BoolPairlistParameter(__PairlistParameterBase):
@@ -78,7 +78,7 @@ class IPairList(LoggingMixin, ABC):
exchange: Exchange, exchange: Exchange,
pairlistmanager, pairlistmanager,
config: Config, config: Config,
pairlistconfig: Dict[str, Any], pairlistconfig: dict[str, Any],
pairlist_pos: int, pairlist_pos: int,
) -> None: ) -> None:
""" """
@@ -126,7 +126,7 @@ class IPairList(LoggingMixin, ABC):
return "" return ""
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
""" """
Return parameters used by this Pairlist Handler, and their type Return parameters used by this Pairlist Handler, and their type
contains a dictionary with the parameter name as key, and a dictionary contains a dictionary with the parameter name as key, and a dictionary
@@ -136,7 +136,7 @@ class IPairList(LoggingMixin, ABC):
return {} return {}
@staticmethod @staticmethod
def refresh_period_parameter() -> Dict[str, PairlistParameter]: def refresh_period_parameter() -> dict[str, PairlistParameter]:
return { return {
"refresh_period": { "refresh_period": {
"type": "number", "type": "number",
@@ -166,7 +166,7 @@ class IPairList(LoggingMixin, ABC):
""" """
raise NotImplementedError() raise NotImplementedError()
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist. Generate the pairlist.
@@ -185,7 +185,7 @@ class IPairList(LoggingMixin, ABC):
"at the first position in the list of Pairlist Handlers." "at the first position in the list of Pairlist Handlers."
) )
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
@@ -209,7 +209,7 @@ class IPairList(LoggingMixin, ABC):
return pairlist return pairlist
def verify_blacklist(self, pairlist: List[str], logmethod) -> List[str]: def verify_blacklist(self, pairlist: list[str], logmethod) -> list[str]:
""" """
Proxy method to verify_blacklist for easy access for child classes. Proxy method to verify_blacklist for easy access for child classes.
:param pairlist: Pairlist to validate :param pairlist: Pairlist to validate
@@ -219,8 +219,8 @@ class IPairList(LoggingMixin, ABC):
return self._pairlistmanager.verify_blacklist(pairlist, logmethod) return self._pairlistmanager.verify_blacklist(pairlist, logmethod)
def verify_whitelist( def verify_whitelist(
self, pairlist: List[str], logmethod, keep_invalid: bool = False self, pairlist: list[str], logmethod, keep_invalid: bool = False
) -> List[str]: ) -> list[str]:
""" """
Proxy method to verify_whitelist for easy access for child classes. Proxy method to verify_whitelist for easy access for child classes.
:param pairlist: Pairlist to validate :param pairlist: Pairlist to validate
@@ -230,7 +230,7 @@ class IPairList(LoggingMixin, ABC):
""" """
return self._pairlistmanager.verify_whitelist(pairlist, logmethod, keep_invalid) return self._pairlistmanager.verify_whitelist(pairlist, logmethod, keep_invalid)
def _whitelist_for_active_markets(self, pairlist: List[str]) -> List[str]: def _whitelist_for_active_markets(self, pairlist: list[str]) -> list[str]:
""" """
Check available markets and remove pair from whitelist if necessary Check available markets and remove pair from whitelist if necessary
:param pairlist: the sorted list of pairs the user might want to trade :param pairlist: the sorted list of pairs the user might want to trade
@@ -243,7 +243,7 @@ class IPairList(LoggingMixin, ABC):
"Markets not loaded. Make sure that exchange is initialized correctly." "Markets not loaded. Make sure that exchange is initialized correctly."
) )
sanitized_whitelist: List[str] = [] sanitized_whitelist: list[str] = []
for pair in pairlist: for pair in pairlist:
# pair is not in the generated dynamic market or has the wrong stake currency # pair is not in the generated dynamic market or has the wrong stake currency
if pair not in markets: if pair not in markets:
@@ -5,7 +5,6 @@ Provides dynamic pair list based on Market Cap
""" """
import logging import logging
from typing import Dict, List
from cachetools import TTLCache from cachetools import TTLCache
@@ -83,7 +82,7 @@ class MarketCapPairList(IPairList):
return "Provides pair list based on CoinGecko's market cap rank." return "Provides pair list based on CoinGecko's market cap rank."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"number_assets": { "number_assets": {
"type": "number", "type": "number",
@@ -114,7 +113,7 @@ class MarketCapPairList(IPairList):
}, },
} }
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -143,7 +142,7 @@ class MarketCapPairList(IPairList):
return pairlist return pairlist
def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: dict) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+2 -3
View File
@@ -3,7 +3,6 @@ Offset pair list filter
""" """
import logging import logging
from typing import Dict, List
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
from freqtrade.exchange.exchange_types import Tickers from freqtrade.exchange.exchange_types import Tickers
@@ -47,7 +46,7 @@ class OffsetFilter(IPairList):
return "Offset pair list filter." return "Offset pair list filter."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"offset": { "offset": {
"type": "number", "type": "number",
@@ -63,7 +62,7 @@ class OffsetFilter(IPairList):
}, },
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
@@ -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, Dict, List, Optional from typing import Any, Optional
from cachetools import TTLCache from cachetools import TTLCache
from pandas import DataFrame from pandas import DataFrame
@@ -115,7 +115,7 @@ class PercentChangePairList(IPairList):
return "Provides dynamic pair list based on percentage change." return "Provides dynamic pair list based on percentage change."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"number_assets": { "number_assets": {
"type": "number", "type": "number",
@@ -163,7 +163,7 @@ class PercentChangePairList(IPairList):
}, },
} }
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -204,7 +204,7 @@ class PercentChangePairList(IPairList):
return pairlist return pairlist
def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: dict) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
@@ -212,7 +212,7 @@ 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[dict[str, Any]] = [{"symbol": k} 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) self.fetch_percent_change_from_lookback_period(filtered_tickers)
@@ -240,8 +240,8 @@ 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[dict[str, str]]
) -> Dict[PairWithTimeframe, DataFrame]: ) -> dict[PairWithTimeframe, DataFrame]:
since_ms = ( since_ms = (
int( int(
timeframe_to_prev_date( timeframe_to_prev_date(
@@ -277,7 +277,7 @@ 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[dict[str, Any]]):
# 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)
@@ -301,7 +301,7 @@ class PercentChangePairList(IPairList):
else: else:
filtered_tickers[i]["percentage"] = 0 filtered_tickers[i]["percentage"] = 0
def fetch_percent_change_from_tickers(self, filtered_tickers: List[Dict[str, Any]], tickers): def fetch_percent_change_from_tickers(self, filtered_tickers: list[dict[str, Any]], tickers):
for i, p in enumerate(filtered_tickers): for i, p in enumerate(filtered_tickers):
# Filter out assets # Filter out assets
if not self._validate_pair( if not self._validate_pair(
@@ -3,7 +3,6 @@ Performance pair list filter
""" """
import logging import logging
from typing import Dict, List
import pandas as pd import pandas as pd
@@ -44,7 +43,7 @@ class PerformanceFilter(IPairList):
return "Filter pairs by performance." return "Filter pairs by performance."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"minutes": { "minutes": {
"type": "number", "type": "number",
@@ -60,7 +59,7 @@ class PerformanceFilter(IPairList):
}, },
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the allowlist again. Filters and sorts pairlist and returns the allowlist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+2 -2
View File
@@ -3,7 +3,7 @@ Price pair list filter
""" """
import logging import logging
from typing import Dict, Optional from typing import Optional
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
from freqtrade.exchange.exchange_types import Ticker from freqtrade.exchange.exchange_types import Ticker
@@ -71,7 +71,7 @@ class PriceFilter(IPairList):
return "Filter pairs by price." return "Filter pairs by price."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"low_price_ratio": { "low_price_ratio": {
"type": "number", "type": "number",
@@ -5,7 +5,7 @@ Provides pair list from Leader data
""" """
import logging import logging
from typing import Dict, List, Optional from typing import Optional
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
from freqtrade.exchange.exchange_types import Tickers from freqtrade.exchange.exchange_types import Tickers
@@ -64,7 +64,7 @@ class ProducerPairList(IPairList):
return "Get a pairlist from an upstream bot." return "Get a pairlist from an upstream bot."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"number_assets": { "number_assets": {
"type": "number", "type": "number",
@@ -83,7 +83,7 @@ class ProducerPairList(IPairList):
}, },
} }
def _filter_pairlist(self, pairlist: Optional[List[str]]): def _filter_pairlist(self, pairlist: Optional[list[str]]):
upstream_pairlist = self._pairlistmanager._dataprovider.get_producer_pairs( upstream_pairlist = self._pairlistmanager._dataprovider.get_producer_pairs(
self._producer_name self._producer_name
) )
@@ -97,7 +97,7 @@ class ProducerPairList(IPairList):
return pairs return pairs
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -108,7 +108,7 @@ class ProducerPairList(IPairList):
pairs = self._whitelist_for_active_markets(self.verify_whitelist(pairs, logger.info)) pairs = self._whitelist_for_active_markets(self.verify_whitelist(pairs, logger.info))
return pairs return pairs
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+10 -10
View File
@@ -6,7 +6,7 @@ Provides pair list fetched from a remote source
import logging import logging
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Tuple from typing import Any
import rapidjson import rapidjson
import requests import requests
@@ -54,7 +54,7 @@ class RemotePairList(IPairList):
self._bearer_token = self._pairlistconfig.get("bearer_token", "") self._bearer_token = self._pairlistconfig.get("bearer_token", "")
self._init_done = False self._init_done = False
self._save_to_file = self._pairlistconfig.get("save_to_file", None) self._save_to_file = self._pairlistconfig.get("save_to_file", None)
self._last_pairlist: List[Any] = list() self._last_pairlist: list[Any] = list()
if self._mode not in ["whitelist", "blacklist"]: if self._mode not in ["whitelist", "blacklist"]:
raise OperationalException( raise OperationalException(
@@ -93,7 +93,7 @@ class RemotePairList(IPairList):
return "Retrieve pairs from a remote API or local file." return "Retrieve pairs from a remote API or local file."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"pairlist_url": { "pairlist_url": {
"type": "string", "type": "string",
@@ -148,7 +148,7 @@ class RemotePairList(IPairList):
}, },
} }
def process_json(self, jsonparse) -> List[str]: def process_json(self, jsonparse) -> list[str]:
pairlist = jsonparse.get("pairs", []) pairlist = jsonparse.get("pairs", [])
remote_refresh_period = int(jsonparse.get("refresh_period", self._refresh_period)) remote_refresh_period = int(jsonparse.get("refresh_period", self._refresh_period))
@@ -166,7 +166,7 @@ class RemotePairList(IPairList):
return pairlist return pairlist
def return_last_pairlist(self) -> List[str]: def return_last_pairlist(self) -> list[str]:
if self._keep_pairlist_on_failure: if self._keep_pairlist_on_failure:
pairlist = self._last_pairlist pairlist = self._last_pairlist
self.log_once("Keeping last fetched pairlist", logger.info) self.log_once("Keeping last fetched pairlist", logger.info)
@@ -175,7 +175,7 @@ class RemotePairList(IPairList):
return pairlist return pairlist
def fetch_pairlist(self) -> Tuple[List[str], float]: def fetch_pairlist(self) -> tuple[list[str], float]:
headers = {"User-Agent": "Freqtrade/" + __version__ + " Remotepairlist"} headers = {"User-Agent": "Freqtrade/" + __version__ + " Remotepairlist"}
if self._bearer_token: if self._bearer_token:
@@ -207,14 +207,14 @@ class RemotePairList(IPairList):
return pairlist, time_elapsed return pairlist, time_elapsed
def _handle_error(self, error: str) -> List[str]: def _handle_error(self, error: str) -> list[str]:
if self._init_done: if self._init_done:
self.log_once("Error: " + error, logger.info) self.log_once("Error: " + error, logger.info)
return self.return_last_pairlist() return self.return_last_pairlist()
else: else:
raise OperationalException(error) raise OperationalException(error)
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -278,7 +278,7 @@ class RemotePairList(IPairList):
return pairlist return pairlist
def save_pairlist(self, pairlist: List[str], filename: str) -> None: def save_pairlist(self, pairlist: list[str], filename: str) -> None:
pairlist_data = {"pairs": pairlist} pairlist_data = {"pairs": pairlist}
try: try:
file_path = Path(filename) file_path = Path(filename)
@@ -288,7 +288,7 @@ class RemotePairList(IPairList):
except Exception as e: except Exception as e:
logger.error(f"Error saving processed pairlist to {filename}: {e}") logger.error(f"Error saving processed pairlist to {filename}: {e}")
def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: dict) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+3 -3
View File
@@ -4,7 +4,7 @@ Shuffle pair list filter
import logging import logging
import random import random
from typing import Dict, List, Literal from typing import Literal
from freqtrade.enums import RunMode from freqtrade.enums import RunMode
from freqtrade.exchange import timeframe_to_seconds from freqtrade.exchange import timeframe_to_seconds
@@ -61,7 +61,7 @@ class ShuffleFilter(IPairList):
return "Randomize pairlist order." return "Randomize pairlist order."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"shuffle_frequency": { "shuffle_frequency": {
"type": "option", "type": "option",
@@ -78,7 +78,7 @@ class ShuffleFilter(IPairList):
}, },
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
+2 -2
View File
@@ -3,7 +3,7 @@ Spread pair list filter
""" """
import logging import logging
from typing import Dict, Optional from typing import Optional
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
from freqtrade.exchange.exchange_types import Ticker from freqtrade.exchange.exchange_types import Ticker
@@ -51,7 +51,7 @@ class SpreadFilter(IPairList):
return "Filter by bid/ask difference." return "Filter by bid/ask difference."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"max_spread_ratio": { "max_spread_ratio": {
"type": "number", "type": "number",
+3 -4
View File
@@ -6,7 +6,6 @@ Provides pair white list as it configured in config
import logging import logging
from copy import deepcopy from copy import deepcopy
from typing import Dict, List
from freqtrade.exchange.exchange_types import Tickers from freqtrade.exchange.exchange_types import Tickers
from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter, SupportsBacktesting from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter, SupportsBacktesting
@@ -45,7 +44,7 @@ class StaticPairList(IPairList):
return "Use pairlist as configured in config." return "Use pairlist as configured in config."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"allow_inactive": { "allow_inactive": {
"type": "boolean", "type": "boolean",
@@ -55,7 +54,7 @@ class StaticPairList(IPairList):
}, },
} }
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -71,7 +70,7 @@ class StaticPairList(IPairList):
# proper warnings in the log # proper warnings in the log
return self._whitelist_for_active_markets(wl) return self._whitelist_for_active_markets(wl)
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
@@ -5,7 +5,7 @@ Volatility pairlist filter
import logging import logging
import sys import sys
from datetime import timedelta from datetime import timedelta
from typing import Dict, List, Optional from typing import Optional
import numpy as np import numpy as np
from cachetools import TTLCache from cachetools import TTLCache
@@ -79,7 +79,7 @@ class VolatilityFilter(IPairList):
return "Filter pairs by their recent volatility." return "Filter pairs by their recent volatility."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"lookback_days": { "lookback_days": {
"type": "number", "type": "number",
@@ -109,7 +109,7 @@ class VolatilityFilter(IPairList):
**IPairList.refresh_period_parameter(), **IPairList.refresh_period_parameter(),
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Validate trading range Validate trading range
:param pairlist: pairlist to filter or sort :param pairlist: pairlist to filter or sort
@@ -123,8 +123,8 @@ class VolatilityFilter(IPairList):
since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days)) since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days))
candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms) candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)
resulting_pairlist: List[str] = [] resulting_pairlist: list[str] = []
volatilitys: Dict[str, float] = {} volatilitys: dict[str, float] = {}
for p in pairlist: for p in pairlist:
daily_candles = candles.get((p, "1d", self._def_candletype), None) daily_candles = candles.get((p, "1d", self._def_candletype), None)
+5 -5
View File
@@ -6,7 +6,7 @@ Provides dynamic pair list based on trade volumes
import logging import logging
from datetime import timedelta from datetime import timedelta
from typing import Any, Dict, List, Literal from typing import Any, Literal
from cachetools import TTLCache from cachetools import TTLCache
@@ -122,7 +122,7 @@ class VolumePairList(IPairList):
return "Provides dynamic pair list based on trade volumes." return "Provides dynamic pair list based on trade volumes."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"number_assets": { "number_assets": {
"type": "number", "type": "number",
@@ -170,7 +170,7 @@ class VolumePairList(IPairList):
}, },
} }
def gen_pairlist(self, tickers: Tickers) -> List[str]: def gen_pairlist(self, tickers: Tickers) -> list[str]:
""" """
Generate the pairlist Generate the pairlist
:param tickers: Tickers (from exchange.get_tickers). May be cached. :param tickers: Tickers (from exchange.get_tickers). May be cached.
@@ -212,7 +212,7 @@ class VolumePairList(IPairList):
return pairlist return pairlist
def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: dict) -> list[str]:
""" """
Filters and sorts pairlist and returns the whitelist again. Filters and sorts pairlist and returns the whitelist again.
Called on each bot iteration - please use internal caching if necessary Called on each bot iteration - please use internal caching if necessary
@@ -222,7 +222,7 @@ class VolumePairList(IPairList):
""" """
if self._use_range: if self._use_range:
# Create bare minimum from tickers structure. # Create bare minimum from tickers structure.
filtered_tickers: List[Dict[str, Any]] = [{"symbol": k} for k in pairlist] filtered_tickers: list[dict[str, Any]] = [{"symbol": k} for k in pairlist]
# get lookback period in ms, for exchange ohlcv fetch # get lookback period in ms, for exchange ohlcv fetch
since_ms = ( since_ms = (
@@ -1,12 +1,11 @@
import re import re
from typing import List
from freqtrade.constants import Config from freqtrade.constants import Config
def expand_pairlist( def expand_pairlist(
wildcardpl: List[str], available_pairs: List[str], keep_invalid: bool = False wildcardpl: list[str], available_pairs: list[str], keep_invalid: bool = False
) -> List[str]: ) -> list[str]:
""" """
Expand pairlist potentially containing wildcards based on available markets. Expand pairlist potentially containing wildcards based on available markets.
This will implicitly filter all pairs in the wildcard-list which are not in available_pairs. This will implicitly filter all pairs in the wildcard-list which are not in available_pairs.
@@ -41,7 +40,7 @@ def expand_pairlist(
return result return result
def dynamic_expand_pairlist(config: Config, markets: List[str]) -> List[str]: def dynamic_expand_pairlist(config: Config, markets: list[str]) -> list[str]:
expanded_pairs = expand_pairlist(config["pairs"], markets) expanded_pairs = expand_pairlist(config["pairs"], markets)
if config.get("freqai", {}).get("enabled", False): if config.get("freqai", {}).get("enabled", False):
corr_pairlist = config["freqai"]["feature_parameters"]["include_corr_pairlist"] corr_pairlist = config["freqai"]["feature_parameters"]["include_corr_pairlist"]
@@ -4,7 +4,7 @@ Rate of change pairlist filter
import logging import logging
from datetime import timedelta from datetime import timedelta
from typing import Dict, List, Optional from typing import Optional
from cachetools import TTLCache from cachetools import TTLCache
from pandas import DataFrame from pandas import DataFrame
@@ -76,7 +76,7 @@ class RangeStabilityFilter(IPairList):
return "Filters pairs by their rate of change." return "Filters pairs by their rate of change."
@staticmethod @staticmethod
def available_parameters() -> Dict[str, PairlistParameter]: def available_parameters() -> dict[str, PairlistParameter]:
return { return {
"lookback_days": { "lookback_days": {
"type": "number", "type": "number",
@@ -106,7 +106,7 @@ class RangeStabilityFilter(IPairList):
**IPairList.refresh_period_parameter(), **IPairList.refresh_period_parameter(),
} }
def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]: def filter_pairlist(self, pairlist: list[str], tickers: Tickers) -> list[str]:
""" """
Validate trading range Validate trading range
:param pairlist: pairlist to filter or sort :param pairlist: pairlist to filter or sort
@@ -120,8 +120,8 @@ class RangeStabilityFilter(IPairList):
since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days + 1)) since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days + 1))
candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms) candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)
resulting_pairlist: List[str] = [] resulting_pairlist: list[str] = []
pct_changes: Dict[str, float] = {} pct_changes: dict[str, float] = {}
for p in pairlist: for p in pairlist:
daily_candles = candles.get((p, "1d", self._def_candletype), None) daily_candles = candles.get((p, "1d", self._def_candletype), None)
+15 -15
View File
@@ -4,7 +4,7 @@ PairList manager class
import logging import logging
from functools import partial from functools import partial
from typing import Dict, List, Optional from typing import Optional
from cachetools import TTLCache, cached from cachetools import TTLCache, cached
@@ -31,7 +31,7 @@ class PairListManager(LoggingMixin):
self._config = config self._config = config
self._whitelist = self._config["exchange"].get("pair_whitelist") self._whitelist = self._config["exchange"].get("pair_whitelist")
self._blacklist = self._config["exchange"].get("pair_blacklist", []) self._blacklist = self._config["exchange"].get("pair_blacklist", [])
self._pairlist_handlers: List[IPairList] = [] self._pairlist_handlers: list[IPairList] = []
self._tickers_needed = False self._tickers_needed = False
self._dataprovider: Optional[DataProvider] = dataprovider self._dataprovider: Optional[DataProvider] = dataprovider
for pairlist_handler_config in self._config.get("pairlists", []): for pairlist_handler_config in self._config.get("pairlists", []):
@@ -67,9 +67,9 @@ class PairListManager(LoggingMixin):
if self._config["runmode"] not in (RunMode.BACKTEST, RunMode.EDGE, RunMode.HYPEROPT): if self._config["runmode"] not in (RunMode.BACKTEST, RunMode.EDGE, RunMode.HYPEROPT):
return return
pairlist_errors: List[str] = [] pairlist_errors: list[str] = []
noaction_pairlists: List[str] = [] noaction_pairlists: list[str] = []
biased_pairlists: List[str] = [] biased_pairlists: list[str] = []
for pairlist_handler in self._pairlist_handlers: for pairlist_handler in self._pairlist_handlers:
if pairlist_handler.supports_backtesting == SupportsBacktesting.NO: if pairlist_handler.supports_backtesting == SupportsBacktesting.NO:
pairlist_errors.append(pairlist_handler.name) pairlist_errors.append(pairlist_handler.name)
@@ -97,12 +97,12 @@ class PairListManager(LoggingMixin):
) )
@property @property
def whitelist(self) -> List[str]: def whitelist(self) -> list[str]:
"""The current whitelist""" """The current whitelist"""
return self._whitelist return self._whitelist
@property @property
def blacklist(self) -> List[str]: def blacklist(self) -> list[str]:
""" """
The current blacklist The current blacklist
-> no need to overwrite in subclasses -> no need to overwrite in subclasses
@@ -110,16 +110,16 @@ class PairListManager(LoggingMixin):
return self._blacklist return self._blacklist
@property @property
def expanded_blacklist(self) -> List[str]: def expanded_blacklist(self) -> list[str]:
"""The expanded blacklist (including wildcard expansion)""" """The expanded blacklist (including wildcard expansion)"""
return expand_pairlist(self._blacklist, self._exchange.get_markets().keys()) return expand_pairlist(self._blacklist, self._exchange.get_markets().keys())
@property @property
def name_list(self) -> List[str]: def name_list(self) -> list[str]:
"""Get list of loaded Pairlist Handler names""" """Get list of loaded Pairlist Handler names"""
return [p.name for p in self._pairlist_handlers] return [p.name for p in self._pairlist_handlers]
def short_desc(self) -> List[Dict]: def short_desc(self) -> list[dict]:
"""List of short_desc for each Pairlist Handler""" """List of short_desc for each Pairlist Handler"""
return [{p.name: p.short_desc()} for p in self._pairlist_handlers] return [{p.name: p.short_desc()} for p in self._pairlist_handlers]
@@ -130,7 +130,7 @@ class PairListManager(LoggingMixin):
def refresh_pairlist(self) -> None: def refresh_pairlist(self) -> None:
"""Run pairlist through all configured Pairlist Handlers.""" """Run pairlist through all configured Pairlist Handlers."""
# Tickers should be cached to avoid calling the exchange on each call. # Tickers should be cached to avoid calling the exchange on each call.
tickers: Dict = {} tickers: dict = {}
if self._tickers_needed: if self._tickers_needed:
tickers = self._get_cached_tickers() tickers = self._get_cached_tickers()
@@ -150,7 +150,7 @@ class PairListManager(LoggingMixin):
self._whitelist = pairlist self._whitelist = pairlist
def verify_blacklist(self, pairlist: List[str], logmethod) -> List[str]: def verify_blacklist(self, pairlist: list[str], logmethod) -> list[str]:
""" """
Verify and remove items from pairlist - returning a filtered pairlist. Verify and remove items from pairlist - returning a filtered pairlist.
Logs a warning or info depending on `aswarning`. Logs a warning or info depending on `aswarning`.
@@ -173,8 +173,8 @@ class PairListManager(LoggingMixin):
return pairlist return pairlist
def verify_whitelist( def verify_whitelist(
self, pairlist: List[str], logmethod, keep_invalid: bool = False self, pairlist: list[str], logmethod, keep_invalid: bool = False
) -> List[str]: ) -> list[str]:
""" """
Verify and remove items from pairlist - returning a filtered pairlist. Verify and remove items from pairlist - returning a filtered pairlist.
Logs a warning or info depending on `aswarning`. Logs a warning or info depending on `aswarning`.
@@ -193,7 +193,7 @@ class PairListManager(LoggingMixin):
return whitelist return whitelist
def create_pair_list( def create_pair_list(
self, pairs: List[str], timeframe: Optional[str] = None self, pairs: list[str], timeframe: Optional[str] = None
) -> ListPairsWithTimeframes: ) -> ListPairsWithTimeframes:
""" """
Create list of pair tuples with (pair, timeframe) Create list of pair tuples with (pair, timeframe)
+6 -6
View File
@@ -4,7 +4,7 @@ Protection manager class
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Any, Dict, List, Optional from typing import Any, Optional
from freqtrade.constants import Config, LongShort from freqtrade.constants import Config, LongShort
from freqtrade.exceptions import ConfigurationError from freqtrade.exceptions import ConfigurationError
@@ -18,10 +18,10 @@ logger = logging.getLogger(__name__)
class ProtectionManager: class ProtectionManager:
def __init__(self, config: Config, protections: List) -> None: def __init__(self, config: Config, protections: list) -> None:
self._config = config self._config = config
self._protection_handlers: List[IProtection] = [] self._protection_handlers: list[IProtection] = []
self.validate_protections(protections) self.validate_protections(protections)
for protection_handler_config in protections: for protection_handler_config in protections:
protection_handler = ProtectionResolver.load_protection( protection_handler = ProtectionResolver.load_protection(
@@ -35,13 +35,13 @@ class ProtectionManager:
logger.info("No protection Handlers defined.") logger.info("No protection Handlers defined.")
@property @property
def name_list(self) -> List[str]: def name_list(self) -> list[str]:
""" """
Get list of loaded Protection Handler names Get list of loaded Protection Handler names
""" """
return [p.name for p in self._protection_handlers] return [p.name for p in self._protection_handlers]
def short_desc(self) -> List[Dict]: def short_desc(self) -> list[dict]:
""" """
List of short_desc for each Pairlist Handler List of short_desc for each Pairlist Handler
""" """
@@ -80,7 +80,7 @@ class ProtectionManager:
return result return result
@staticmethod @staticmethod
def validate_protections(protections: List[Dict[str, Any]]) -> None: def validate_protections(protections: list[dict[str, Any]]) -> None:
""" """
Validate protection setup validity Validate protection setup validity
""" """
+3 -3
View File
@@ -2,7 +2,7 @@ import logging
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional from typing import Any, Optional
from freqtrade.constants import Config, LongShort from freqtrade.constants import Config, LongShort
from freqtrade.exchange import timeframe_to_minutes from freqtrade.exchange import timeframe_to_minutes
@@ -28,7 +28,7 @@ class IProtection(LoggingMixin, ABC):
# Can stop trading for one pair # Can stop trading for one pair
has_local_stop: bool = False has_local_stop: bool = False
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: dict[str, Any]) -> None:
self._config = config self._config = config
self._protection_config = protection_config self._protection_config = protection_config
self._stop_duration_candles: Optional[int] = None self._stop_duration_candles: Optional[int] = None
@@ -119,7 +119,7 @@ class IProtection(LoggingMixin, ABC):
If true, this pair will be locked with <reason> until <until> If true, this pair will be locked with <reason> until <until>
""" """
def calculate_lock_end(self, trades: List[LocalTrade]) -> datetime: def calculate_lock_end(self, trades: list[LocalTrade]) -> datetime:
""" """
Get lock end time Get lock end time
Implicitly uses `self._stop_duration` or `self._unlock_at` depending on the configuration. Implicitly uses `self._stop_duration` or `self._unlock_at` depending on the configuration.
@@ -1,6 +1,6 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Optional
from freqtrade.constants import Config, LongShort from freqtrade.constants import Config, LongShort
from freqtrade.persistence import Trade from freqtrade.persistence import Trade
@@ -14,7 +14,7 @@ class LowProfitPairs(IProtection):
has_global_stop: bool = False has_global_stop: bool = False
has_local_stop: bool = True has_local_stop: bool = True
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get("trade_limit", 1) self._trade_limit = protection_config.get("trade_limit", 1)
@@ -1,6 +1,6 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Optional
import pandas as pd import pandas as pd
@@ -17,7 +17,7 @@ class MaxDrawdown(IProtection):
has_global_stop: bool = True has_global_stop: bool = True
has_local_stop: bool = False has_local_stop: bool = False
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get("trade_limit", 1) self._trade_limit = protection_config.get("trade_limit", 1)
@@ -1,6 +1,6 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Optional
from freqtrade.constants import Config, LongShort from freqtrade.constants import Config, LongShort
from freqtrade.enums import ExitType from freqtrade.enums import ExitType
@@ -15,7 +15,7 @@ class StoplossGuard(IProtection):
has_global_stop: bool = True has_global_stop: bool = True
has_local_stop: bool = True has_local_stop: bool = True
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get("trade_limit", 10) self._trade_limit = protection_config.get("trade_limit", 10)