stable/freqtrade/strategy/hyper.py

201 lines
8.0 KiB
Python
Raw Normal View History

"""
IHyperStrategy interface, hyperoptable Parameter class.
This module defines a base class for auto-hyperoptable strategies.
"""
import logging
from pathlib import Path
from typing import Any, Dict, Iterator, List, Tuple, Type, Union
2022-09-18 11:20:36 +00:00
from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException
from freqtrade.misc import deep_merge_dicts, json_load
from freqtrade.optimize.hyperopt_tools import HyperoptTools
from freqtrade.strategy.parameters import BaseParameter
logger = logging.getLogger(__name__)
2021-11-11 11:06:18 +00:00
class HyperStrategyMixin:
"""
2021-06-25 13:45:49 +00:00
A helper base class which allows HyperOptAuto class to reuse implementations of buy/sell
strategy logic.
"""
2022-09-18 11:20:36 +00:00
def __init__(self, config: Config, *args, **kwargs):
"""
Initialize hyperoptable strategy mixin.
"""
self.config = config
2021-05-02 08:45:21 +00:00
self.ft_buy_params: List[BaseParameter] = []
self.ft_sell_params: List[BaseParameter] = []
2021-08-03 05:10:04 +00:00
self.ft_protection_params: List[BaseParameter] = []
2021-05-02 08:45:21 +00:00
2022-05-29 13:58:40 +00:00
params = self.load_params_from_file()
params = params.get('params', {})
self._ft_params_from_file = params
# Init/loading of parameters is done as part of ft_bot_start().
def enumerate_parameters(self, category: str = None) -> Iterator[Tuple[str, BaseParameter]]:
"""
2021-06-25 13:45:49 +00:00
Find all optimizable parameters and return (name, attr) iterator.
:param category:
:return:
"""
2021-08-03 05:10:04 +00:00
if category not in ('buy', 'sell', 'protection', None):
2021-08-04 18:01:28 +00:00
raise OperationalException(
'Category must be one of: "buy", "sell", "protection", None.')
2021-05-02 08:45:21 +00:00
if category is None:
2021-08-03 05:10:04 +00:00
params = self.ft_buy_params + self.ft_sell_params + self.ft_protection_params
2021-05-02 08:45:21 +00:00
else:
params = getattr(self, f"ft_{category}_params")
for par in params:
yield par.name, par
@classmethod
def detect_all_parameters(cls) -> Dict:
""" Detect all parameters and return them as a list"""
2022-05-25 18:43:43 +00:00
params: Dict[str, Any] = {
'buy': list(detect_parameters(cls, 'buy')),
'sell': list(detect_parameters(cls, 'sell')),
'protection': list(detect_parameters(cls, 'protection')),
}
params.update({
2021-08-03 05:10:04 +00:00
'count': len(params['buy'] + params['sell'] + params['protection'])
})
return params
2022-05-30 04:52:44 +00:00
def ft_load_params_from_file(self) -> None:
2022-05-30 04:32:35 +00:00
"""
Load Parameters from parameter file
Should/must run before config values are loaded in strategy_resolver.
"""
if self._ft_params_from_file:
# Set parameters from Hyperopt results file
params = self._ft_params_from_file
self.minimal_roi = params.get('roi', getattr(self, 'minimal_roi', {}))
self.stoploss = params.get('stoploss', {}).get(
'stoploss', getattr(self, 'stoploss', -0.1))
self.max_open_trades = params.get('max_open_trades', {}).get(
'max_open_trades', getattr(self, 'max_open_trades', -1))
trailing = params.get('trailing', {})
self.trailing_stop = trailing.get(
'trailing_stop', getattr(self, 'trailing_stop', False))
self.trailing_stop_positive = trailing.get(
'trailing_stop_positive', getattr(self, 'trailing_stop_positive', None))
self.trailing_stop_positive_offset = trailing.get(
'trailing_stop_positive_offset',
getattr(self, 'trailing_stop_positive_offset', 0))
self.trailing_only_offset_is_reached = trailing.get(
'trailing_only_offset_is_reached',
getattr(self, 'trailing_only_offset_is_reached', 0.0))
2022-05-29 13:58:40 +00:00
def ft_load_hyper_params(self, hyperopt: bool = False) -> None:
2021-04-24 05:00:33 +00:00
"""
Load Hyperoptable parameters
2022-05-29 13:58:40 +00:00
Prevalence:
* Parameters from parameter file
* Parameters defined in parameters objects (buy_params, sell_params, ...)
* Parameter defaults
2021-04-24 05:00:33 +00:00
"""
2022-05-29 13:58:40 +00:00
buy_params = deep_merge_dicts(self._ft_params_from_file.get('buy', {}),
getattr(self, 'buy_params', {}))
sell_params = deep_merge_dicts(self._ft_params_from_file.get('sell', {}),
getattr(self, 'sell_params', {}))
protection_params = deep_merge_dicts(self._ft_params_from_file.get('protection', {}),
2021-08-03 05:10:04 +00:00
getattr(self, 'protection_params', {}))
2021-04-24 05:00:33 +00:00
2022-05-29 13:58:40 +00:00
self._ft_load_params(buy_params, 'buy', hyperopt)
self._ft_load_params(sell_params, 'sell', hyperopt)
self._ft_load_params(protection_params, 'protection', hyperopt)
def load_params_from_file(self) -> Dict:
filename_str = getattr(self, '__file__', '')
if not filename_str:
return {}
filename = Path(filename_str).with_suffix('.json')
if filename.is_file():
logger.info(f"Loading parameters from file {filename}")
try:
with filename.open('r') as f:
params = json_load(f)
if params.get('strategy_name') != self.__class__.__name__:
2021-06-30 04:43:49 +00:00
raise OperationalException('Invalid parameter file provided.')
return params
except ValueError:
2021-06-30 04:43:49 +00:00
logger.warning("Invalid parameter file format.")
return {}
logger.info("Found no parameter file.")
return {}
2022-05-29 13:58:40 +00:00
def _ft_load_params(self, params: Dict, space: str, hyperopt: bool = False) -> None:
"""
2021-06-25 13:45:49 +00:00
Set optimizable parameter values.
:param params: Dictionary with new parameter values.
"""
if not params:
logger.info(f"No params for {space} found, using default values.")
2021-05-02 08:45:21 +00:00
param_container: List[BaseParameter] = getattr(self, f"ft_{space}_params")
for attr_name, attr in detect_parameters(self, space):
2021-05-02 08:45:21 +00:00
attr.name = attr_name
attr.in_space = hyperopt and HyperoptTools.has_space(self.config, space)
2021-05-02 08:45:21 +00:00
if not attr.category:
attr.category = space
param_container.append(attr)
if params and attr_name in params:
if attr.load:
attr.value = params[attr_name]
logger.info(f'Strategy Parameter: {attr_name} = {attr.value}')
else:
logger.warning(f'Parameter "{attr_name}" exists, but is disabled. '
f'Default value "{attr.value}" used.')
else:
logger.info(f'Strategy Parameter(default): {attr_name} = {attr.value}')
2021-05-02 08:45:21 +00:00
def get_no_optimize_params(self):
2021-05-02 08:45:21 +00:00
"""
Returns list of Parameters that are not part of the current optimize job
"""
2022-05-25 18:43:43 +00:00
params: Dict[str, Dict] = {
2021-05-02 08:45:21 +00:00
'buy': {},
2021-08-03 05:10:04 +00:00
'sell': {},
'protection': {},
2021-05-02 08:45:21 +00:00
}
for name, p in self.enumerate_parameters():
if not p.optimize or not p.in_space:
params[p.category][name] = p.value
return params
def detect_parameters(
obj: Union[HyperStrategyMixin, Type[HyperStrategyMixin]],
category: str
) -> Iterator[Tuple[str, BaseParameter]]:
"""
Detect all parameters for 'category' for "obj"
:param obj: Strategy object or class
:param category: category - usually `'buy', 'sell', 'protection',...
"""
for attr_name in dir(obj):
if not attr_name.startswith('__'): # Ignore internals, not strictly necessary.
attr = getattr(obj, attr_name)
if issubclass(attr.__class__, BaseParameter):
if (attr_name.startswith(category + '_')
and attr.category is not None and attr.category != category):
raise OperationalException(
f'Inconclusive parameter name {attr_name}, category: {attr.category}.')
if (category == attr.category or
(attr_name.startswith(category + '_') and attr.category is None)):
yield attr_name, attr