chore: update strategy to modern typing syntax

This commit is contained in:
Matthias
2024-10-04 07:07:32 +02:00
parent 1d4658e978
commit acc40c73f3
4 changed files with 47 additions and 45 deletions

View File

@@ -4,8 +4,9 @@ This module defines a base class for auto-hyperoptable strategies.
""" """
import logging import logging
from collections.abc import Iterator
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union from typing import Any, Optional, Union
from freqtrade.constants import Config from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
@@ -28,9 +29,9 @@ class HyperStrategyMixin:
Initialize hyperoptable strategy mixin. Initialize hyperoptable strategy mixin.
""" """
self.config = config self.config = config
self.ft_buy_params: List[BaseParameter] = [] self.ft_buy_params: list[BaseParameter] = []
self.ft_sell_params: List[BaseParameter] = [] self.ft_sell_params: list[BaseParameter] = []
self.ft_protection_params: List[BaseParameter] = [] self.ft_protection_params: list[BaseParameter] = []
params = self.load_params_from_file() params = self.load_params_from_file()
params = params.get("params", {}) params = params.get("params", {})
@@ -39,7 +40,7 @@ class HyperStrategyMixin:
def enumerate_parameters( def enumerate_parameters(
self, category: Optional[str] = None self, category: Optional[str] = None
) -> Iterator[Tuple[str, BaseParameter]]: ) -> Iterator[tuple[str, BaseParameter]]:
""" """
Find all optimizable parameters and return (name, attr) iterator. Find all optimizable parameters and return (name, attr) iterator.
:param category: :param category:
@@ -59,9 +60,9 @@ class HyperStrategyMixin:
yield par.name, par yield par.name, par
@classmethod @classmethod
def detect_all_parameters(cls) -> Dict: def detect_all_parameters(cls) -> dict:
"""Detect all parameters and return them as a list""" """Detect all parameters and return them as a list"""
params: Dict[str, Any] = { params: dict[str, Any] = {
"buy": list(detect_parameters(cls, "buy")), "buy": list(detect_parameters(cls, "buy")),
"sell": list(detect_parameters(cls, "sell")), "sell": list(detect_parameters(cls, "sell")),
"protection": list(detect_parameters(cls, "protection")), "protection": list(detect_parameters(cls, "protection")),
@@ -124,7 +125,7 @@ class HyperStrategyMixin:
self._ft_load_params(sell_params, "sell", hyperopt) self._ft_load_params(sell_params, "sell", hyperopt)
self._ft_load_params(protection_params, "protection", hyperopt) self._ft_load_params(protection_params, "protection", hyperopt)
def load_params_from_file(self) -> Dict: def load_params_from_file(self) -> dict:
filename_str = getattr(self, "__file__", "") filename_str = getattr(self, "__file__", "")
if not filename_str: if not filename_str:
return {} return {}
@@ -144,14 +145,14 @@ class HyperStrategyMixin:
return {} return {}
def _ft_load_params(self, params: Dict, space: str, hyperopt: bool = False) -> None: def _ft_load_params(self, params: dict, space: str, hyperopt: bool = False) -> None:
""" """
Set optimizable parameter values. Set optimizable parameter values.
:param params: Dictionary with new parameter values. :param params: Dictionary with new parameter values.
""" """
if not params: if not params:
logger.info(f"No params for {space} found, using default values.") logger.info(f"No params for {space} found, using default values.")
param_container: List[BaseParameter] = getattr(self, f"ft_{space}_params") param_container: list[BaseParameter] = getattr(self, f"ft_{space}_params")
for attr_name, attr in detect_parameters(self, space): for attr_name, attr in detect_parameters(self, space):
attr.name = attr_name attr.name = attr_name
@@ -173,11 +174,11 @@ class HyperStrategyMixin:
else: else:
logger.info(f"Strategy Parameter(default): {attr_name} = {attr.value}") logger.info(f"Strategy Parameter(default): {attr_name} = {attr.value}")
def get_no_optimize_params(self) -> Dict[str, Dict]: def get_no_optimize_params(self) -> dict[str, dict]:
""" """
Returns list of Parameters that are not part of the current optimize job Returns list of Parameters that are not part of the current optimize job
""" """
params: Dict[str, Dict] = { params: dict[str, dict] = {
"buy": {}, "buy": {},
"sell": {}, "sell": {},
"protection": {}, "protection": {},
@@ -189,8 +190,8 @@ class HyperStrategyMixin:
def detect_parameters( def detect_parameters(
obj: Union[HyperStrategyMixin, Type[HyperStrategyMixin]], category: str obj: Union[HyperStrategyMixin, type[HyperStrategyMixin]], category: str
) -> Iterator[Tuple[str, BaseParameter]]: ) -> Iterator[tuple[str, BaseParameter]]:
""" """
Detect all parameters for 'category' for "obj" Detect all parameters for 'category' for "obj"
:param obj: Strategy object or class :param obj: Strategy object or class

View File

@@ -1,5 +1,5 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, Union from typing import Any, Callable, Optional, Union
from pandas import DataFrame from pandas import DataFrame
@@ -73,7 +73,7 @@ def informative(
return decorator return decorator
def __get_pair_formats(market: Optional[Dict[str, Any]]) -> Dict[str, str]: def __get_pair_formats(market: Optional[dict[str, Any]]) -> dict[str, str]:
if not market: if not market:
return {} return {}
base = market["base"] base = market["base"]
@@ -86,7 +86,7 @@ def __get_pair_formats(market: Optional[Dict[str, Any]]) -> Dict[str, str]:
} }
def _format_pair_name(config, pair: str, market: Optional[Dict[str, Any]] = None) -> str: def _format_pair_name(config, pair: str, market: Optional[dict[str, Any]] = None) -> str:
return pair.format( return pair.format(
stake_currency=config["stake_currency"], stake_currency=config["stake_currency"],
stake=config["stake_currency"], stake=config["stake_currency"],

View File

@@ -8,7 +8,7 @@ from abc import ABC, abstractmethod
from collections import OrderedDict from collections import OrderedDict
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from math import isinf, isnan from math import isinf, isnan
from typing import Dict, List, Optional, Tuple, Union from typing import Optional, Union
from pandas import DataFrame from pandas import DataFrame
@@ -63,9 +63,9 @@ class IStrategy(ABC, HyperStrategyMixin):
# Version 3 - First version with short and leverage support # Version 3 - First version with short and leverage support
INTERFACE_VERSION: int = 3 INTERFACE_VERSION: int = 3
_ft_params_from_file: Dict _ft_params_from_file: dict
# associated minimal roi # associated minimal roi
minimal_roi: Dict = {} minimal_roi: dict = {}
# associated stoploss # associated stoploss
stoploss: float stoploss: float
@@ -87,7 +87,7 @@ class IStrategy(ABC, HyperStrategyMixin):
timeframe: str timeframe: str
# Optional order types # Optional order types
order_types: Dict = { order_types: dict = {
"entry": "limit", "entry": "limit",
"exit": "limit", "exit": "limit",
"stoploss": "limit", "stoploss": "limit",
@@ -96,7 +96,7 @@ class IStrategy(ABC, HyperStrategyMixin):
} }
# Optional time in force # Optional time in force
order_time_in_force: Dict = { order_time_in_force: dict = {
"entry": "GTC", "entry": "GTC",
"exit": "GTC", "exit": "GTC",
} }
@@ -123,7 +123,7 @@ class IStrategy(ABC, HyperStrategyMixin):
startup_candle_count: int = 0 startup_candle_count: int = 0
# Protections # Protections
protections: List = [] protections: list = []
# Class level variables (intentional) containing # Class level variables (intentional) containing
# the dataprovider (dp) (access to other candles, historic data, ...) # the dataprovider (dp) (access to other candles, historic data, ...)
@@ -136,24 +136,24 @@ class IStrategy(ABC, HyperStrategyMixin):
__source__: str = "" __source__: str = ""
# Definition of plot_config. See plotting documentation for more details. # Definition of plot_config. See plotting documentation for more details.
plot_config: Dict = {} plot_config: dict = {}
# A self set parameter that represents the market direction. filled from configuration # A self set parameter that represents the market direction. filled from configuration
market_direction: MarketDirection = MarketDirection.NONE market_direction: MarketDirection = MarketDirection.NONE
# Global cache dictionary # Global cache dictionary
_cached_grouped_trades_per_pair: Dict[ _cached_grouped_trades_per_pair: dict[
str, OrderedDict[Tuple[datetime, datetime], DataFrame] str, OrderedDict[tuple[datetime, datetime], DataFrame]
] = {} ] = {}
def __init__(self, config: Config) -> None: def __init__(self, config: Config) -> None:
self.config = config self.config = config
# Dict to determine if analysis is necessary # Dict to determine if analysis is necessary
self._last_candle_seen_per_pair: Dict[str, datetime] = {} self._last_candle_seen_per_pair: dict[str, datetime] = {}
super().__init__(config) super().__init__(config)
# Gather informative pairs from @informative-decorated methods. # Gather informative pairs from @informative-decorated methods.
self._ft_informative: List[Tuple[InformativeData, PopulateIndicators]] = [] self._ft_informative: list[tuple[InformativeData, PopulateIndicators]] = []
for attr_name in dir(self.__class__): for attr_name in dir(self.__class__):
cls_method = getattr(self.__class__, attr_name) cls_method = getattr(self.__class__, attr_name)
if not callable(cls_method): if not callable(cls_method):
@@ -627,7 +627,7 @@ class IStrategy(ABC, HyperStrategyMixin):
current_entry_profit: float, current_entry_profit: float,
current_exit_profit: float, current_exit_profit: float,
**kwargs, **kwargs,
) -> Union[Optional[float], Tuple[Optional[float], Optional[str]]]: ) -> Union[Optional[float], tuple[Optional[float], Optional[str]]]:
""" """
Custom trade adjustment logic, returning the stake amount that a trade should be Custom trade adjustment logic, returning the stake amount that a trade should be
increased or decreased. increased or decreased.
@@ -761,7 +761,7 @@ class IStrategy(ABC, HyperStrategyMixin):
return df return df
def feature_engineering_expand_all( def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: Dict, **kwargs self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame: ) -> DataFrame:
""" """
*Only functional with FreqAI enabled strategies* *Only functional with FreqAI enabled strategies*
@@ -789,7 +789,7 @@ class IStrategy(ABC, HyperStrategyMixin):
return dataframe return dataframe
def feature_engineering_expand_basic( def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: Dict, **kwargs self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame: ) -> DataFrame:
""" """
*Only functional with FreqAI enabled strategies* *Only functional with FreqAI enabled strategies*
@@ -820,7 +820,7 @@ class IStrategy(ABC, HyperStrategyMixin):
return dataframe return dataframe
def feature_engineering_standard( def feature_engineering_standard(
self, dataframe: DataFrame, metadata: Dict, **kwargs self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame: ) -> DataFrame:
""" """
*Only functional with FreqAI enabled strategies* *Only functional with FreqAI enabled strategies*
@@ -845,7 +845,7 @@ class IStrategy(ABC, HyperStrategyMixin):
""" """
return dataframe return dataframe
def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame: def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
""" """
*Only functional with FreqAI enabled strategies* *Only functional with FreqAI enabled strategies*
Required function to set the targets for the model. Required function to set the targets for the model.
@@ -880,7 +880,7 @@ class IStrategy(ABC, HyperStrategyMixin):
current_entry_profit: float, current_entry_profit: float,
current_exit_profit: float, current_exit_profit: float,
**kwargs, **kwargs,
) -> Tuple[Optional[float], str]: ) -> tuple[Optional[float], str]:
""" """
wrapper around adjust_trade_position to handle the return value wrapper around adjust_trade_position to handle the return value
""" """
@@ -1112,7 +1112,7 @@ class IStrategy(ABC, HyperStrategyMixin):
logger.warning("Empty dataframe for pair %s", pair) logger.warning("Empty dataframe for pair %s", pair)
return return
def analyze(self, pairs: List[str]) -> None: def analyze(self, pairs: list[str]) -> None:
""" """
Analyze all pairs using analyze_pair(). Analyze all pairs using analyze_pair().
:param pairs: List of pairs to analyze :param pairs: List of pairs to analyze
@@ -1121,7 +1121,7 @@ class IStrategy(ABC, HyperStrategyMixin):
self.analyze_pair(pair) self.analyze_pair(pair)
@staticmethod @staticmethod
def preserve_df(dataframe: DataFrame) -> Tuple[int, float, datetime]: def preserve_df(dataframe: DataFrame) -> tuple[int, float, datetime]:
"""keep some data for dataframes""" """keep some data for dataframes"""
return len(dataframe), dataframe["close"].iloc[-1], dataframe["date"].iloc[-1] return len(dataframe), dataframe["close"].iloc[-1], dataframe["date"].iloc[-1]
@@ -1152,7 +1152,7 @@ class IStrategy(ABC, HyperStrategyMixin):
pair: str, pair: str,
timeframe: str, timeframe: str,
dataframe: DataFrame, dataframe: DataFrame,
) -> Tuple[Optional[DataFrame], Optional[datetime]]: ) -> tuple[Optional[DataFrame], Optional[datetime]]:
""" """
Calculates current signal based based on the entry order or exit order Calculates current signal based based on the entry order or exit order
columns of the dataframe. columns of the dataframe.
@@ -1185,7 +1185,7 @@ class IStrategy(ABC, HyperStrategyMixin):
def get_exit_signal( def get_exit_signal(
self, pair: str, timeframe: str, dataframe: DataFrame, is_short: Optional[bool] = None self, pair: str, timeframe: str, dataframe: DataFrame, is_short: Optional[bool] = None
) -> Tuple[bool, bool, Optional[str]]: ) -> tuple[bool, bool, Optional[str]]:
""" """
Calculates current exit signal based based on the dataframe Calculates current exit signal based based on the dataframe
columns of the dataframe. columns of the dataframe.
@@ -1221,7 +1221,7 @@ class IStrategy(ABC, HyperStrategyMixin):
pair: str, pair: str,
timeframe: str, timeframe: str,
dataframe: DataFrame, dataframe: DataFrame,
) -> Tuple[Optional[SignalDirection], Optional[str]]: ) -> tuple[Optional[SignalDirection], Optional[str]]:
""" """
Calculates current entry signal based based on the dataframe signals Calculates current entry signal based based on the dataframe signals
columns of the dataframe. columns of the dataframe.
@@ -1292,7 +1292,7 @@ class IStrategy(ABC, HyperStrategyMixin):
low: Optional[float] = None, low: Optional[float] = None,
high: Optional[float] = None, high: Optional[float] = None,
force_stoploss: float = 0, force_stoploss: float = 0,
) -> List[ExitCheckTuple]: ) -> list[ExitCheckTuple]:
""" """
This function evaluates if one of the conditions required to trigger an exit order This function evaluates if one of the conditions required to trigger an exit order
has been reached, which can either be a stop-loss, ROI or exit-signal. has been reached, which can either be a stop-loss, ROI or exit-signal.
@@ -1301,7 +1301,7 @@ class IStrategy(ABC, HyperStrategyMixin):
:param force_stoploss: Externally provided stoploss :param force_stoploss: Externally provided stoploss
:return: List of exit reasons - or empty list. :return: List of exit reasons - or empty list.
""" """
exits: List[ExitCheckTuple] = [] exits: list[ExitCheckTuple] = []
current_rate = rate current_rate = rate
current_profit = trade.calc_profit_ratio(current_rate) current_profit = trade.calc_profit_ratio(current_rate)
current_profit_best = current_profit current_profit_best = current_profit
@@ -1520,7 +1520,7 @@ class IStrategy(ABC, HyperStrategyMixin):
return ExitCheckTuple(exit_type=ExitType.NONE) return ExitCheckTuple(exit_type=ExitType.NONE)
def min_roi_reached_entry(self, trade_dur: int) -> Tuple[Optional[int], Optional[float]]: def min_roi_reached_entry(self, trade_dur: int) -> tuple[Optional[int], Optional[float]]:
""" """
Based on trade duration defines the ROI entry that may have been reached. Based on trade duration defines the ROI entry that may have been reached.
:param trade_dur: trade duration in minutes :param trade_dur: trade duration in minutes
@@ -1573,7 +1573,7 @@ class IStrategy(ABC, HyperStrategyMixin):
pair=trade.pair, trade=trade, order=order, current_time=current_time pair=trade.pair, trade=trade, order=order, current_time=current_time
) )
def advise_all_indicators(self, data: Dict[str, DataFrame]) -> Dict[str, DataFrame]: def advise_all_indicators(self, data: dict[str, DataFrame]) -> dict[str, DataFrame]:
""" """
Populates indicators for given candle (OHLCV) data (for multiple pairs) Populates indicators for given candle (OHLCV) data (for multiple pairs)
Does not run advise_entry or advise_exit! Does not run advise_entry or advise_exit!
@@ -1611,7 +1611,7 @@ class IStrategy(ABC, HyperStrategyMixin):
config["timeframe"] = self.timeframe config["timeframe"] = self.timeframe
pair = metadata["pair"] pair = metadata["pair"]
# TODO: slice trades to size of dataframe for faster backtesting # TODO: slice trades to size of dataframe for faster backtesting
cached_grouped_trades: OrderedDict[Tuple[datetime, datetime], DataFrame] = ( cached_grouped_trades: OrderedDict[tuple[datetime, datetime], DataFrame] = (
self._cached_grouped_trades_per_pair.get(pair, OrderedDict()) self._cached_grouped_trades_per_pair.get(pair, OrderedDict())
) )
dataframe, cached_grouped_trades = populate_dataframe_with_trades( dataframe, cached_grouped_trades = populate_dataframe_with_trades(

View File

@@ -5,8 +5,9 @@ This module defines a base class for auto-hyperoptable strategies.
import logging import logging
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Sequence
from contextlib import suppress from contextlib import suppress
from typing import Any, Optional, Sequence, Union from typing import Any, Optional, Union
from freqtrade.enums import HyperoptState from freqtrade.enums import HyperoptState
from freqtrade.optimize.hyperopt_tools import HyperoptStateContainer from freqtrade.optimize.hyperopt_tools import HyperoptStateContainer