193 lines
7.8 KiB
Python
193 lines
7.8 KiB
Python
"""
|
|
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
|
|
|
|
from freqtrade.enums import RunMode
|
|
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__)
|
|
|
|
|
|
class HyperStrategyMixin:
|
|
"""
|
|
A helper base class which allows HyperOptAuto class to reuse implementations of buy/sell
|
|
strategy logic.
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any], *args, **kwargs):
|
|
"""
|
|
Initialize hyperoptable strategy mixin.
|
|
"""
|
|
self.config = config
|
|
self.ft_buy_params: List[BaseParameter] = []
|
|
self.ft_sell_params: List[BaseParameter] = []
|
|
self.ft_protection_params: List[BaseParameter] = []
|
|
|
|
params = self.load_params_from_file()
|
|
params = params.get('params', {})
|
|
self._ft_params_from_file = params
|
|
|
|
if config.get('runmode') != RunMode.BACKTEST:
|
|
self.ft_load_hyper_params(config.get('runmode') == RunMode.HYPEROPT)
|
|
|
|
def enumerate_parameters(self, category: str = None) -> Iterator[Tuple[str, BaseParameter]]:
|
|
"""
|
|
Find all optimizable parameters and return (name, attr) iterator.
|
|
:param category:
|
|
:return:
|
|
"""
|
|
if category not in ('buy', 'sell', 'protection', None):
|
|
raise OperationalException(
|
|
'Category must be one of: "buy", "sell", "protection", None.')
|
|
|
|
if category is None:
|
|
params = self.ft_buy_params + self.ft_sell_params + self.ft_protection_params
|
|
else:
|
|
params = getattr(self, f"ft_{category}_params")
|
|
|
|
for par in params:
|
|
yield par.name, par
|
|
|
|
@classmethod
|
|
def detect_parameters(cls, category: str) -> Iterator[Tuple[str, BaseParameter]]:
|
|
""" Detect all parameters for 'category' """
|
|
for attr_name in dir(cls):
|
|
if not attr_name.startswith('__'): # Ignore internals, not strictly necessary.
|
|
attr = getattr(cls, 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
|
|
|
|
@classmethod
|
|
def detect_all_parameters(cls) -> Dict:
|
|
""" Detect all parameters and return them as a list"""
|
|
params: Dict[str, Any] = {
|
|
'buy': list(cls.detect_parameters('buy')),
|
|
'sell': list(cls.detect_parameters('sell')),
|
|
'protection': list(cls.detect_parameters('protection')),
|
|
}
|
|
params.update({
|
|
'count': len(params['buy'] + params['sell'] + params['protection'])
|
|
})
|
|
|
|
return params
|
|
|
|
def ft_load_params_from_file(self) -> None:
|
|
"""
|
|
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))
|
|
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))
|
|
|
|
def ft_load_hyper_params(self, hyperopt: bool = False) -> None:
|
|
"""
|
|
Load Hyperoptable parameters
|
|
Prevalence:
|
|
* Parameters from parameter file
|
|
* Parameters defined in parameters objects (buy_params, sell_params, ...)
|
|
* Parameter defaults
|
|
"""
|
|
|
|
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', {}),
|
|
getattr(self, 'protection_params', {}))
|
|
|
|
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__:
|
|
raise OperationalException('Invalid parameter file provided.')
|
|
return params
|
|
except ValueError:
|
|
logger.warning("Invalid parameter file format.")
|
|
return {}
|
|
logger.info("Found no parameter file.")
|
|
|
|
return {}
|
|
|
|
def _ft_load_params(self, params: Dict, space: str, hyperopt: bool = False) -> None:
|
|
"""
|
|
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.")
|
|
param_container: List[BaseParameter] = getattr(self, f"ft_{space}_params")
|
|
|
|
for attr_name, attr in self.detect_parameters(space):
|
|
attr.name = attr_name
|
|
attr.in_space = hyperopt and HyperoptTools.has_space(self.config, space)
|
|
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}')
|
|
|
|
def get_no_optimize_params(self):
|
|
"""
|
|
Returns list of Parameters that are not part of the current optimize job
|
|
"""
|
|
params: Dict[str, Dict] = {
|
|
'buy': {},
|
|
'sell': {},
|
|
'protection': {},
|
|
}
|
|
for name, p in self.enumerate_parameters():
|
|
if not p.optimize or not p.in_space:
|
|
params[p.category][name] = p.value
|
|
return params
|