556 lines
23 KiB
Python
556 lines
23 KiB
Python
# pragma pylint: disable=too-many-instance-attributes, pointless-string-statement
|
|
|
|
"""
|
|
This module contains the hyperopt logic
|
|
"""
|
|
|
|
import locale
|
|
import logging
|
|
import sys
|
|
import warnings
|
|
from collections import OrderedDict
|
|
from operator import itemgetter
|
|
from pathlib import Path
|
|
from pprint import pprint
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import rapidjson
|
|
from colorama import Fore, Style
|
|
from colorama import init as colorama_init
|
|
from joblib import (Parallel, cpu_count, delayed, dump, load,
|
|
wrap_non_picklable_objects)
|
|
from pandas import DataFrame
|
|
|
|
from freqtrade import OperationalException
|
|
from freqtrade.data.history import get_timeframe, trim_dataframe
|
|
from freqtrade.misc import plural, round_dict
|
|
from freqtrade.optimize.backtesting import Backtesting
|
|
# Import IHyperOpt and IHyperOptLoss to allow unpickling classes from these modules
|
|
from freqtrade.optimize.hyperopt_interface import IHyperOpt # noqa: F4
|
|
from freqtrade.optimize.hyperopt_loss_interface import IHyperOptLoss # noqa: F4
|
|
from freqtrade.resolvers.hyperopt_resolver import (HyperOptLossResolver,
|
|
HyperOptResolver)
|
|
|
|
# Suppress scikit-learn FutureWarnings from skopt
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings("ignore", category=FutureWarning)
|
|
from skopt import Optimizer
|
|
from skopt.space import Dimension
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
INITIAL_POINTS = 30
|
|
|
|
# Keep no more than 2*SKOPT_MODELS_MAX_NUM models
|
|
# in the skopt models list
|
|
SKOPT_MODELS_MAX_NUM = 10
|
|
|
|
MAX_LOSS = 100000 # just a big enough number to be bad result in loss optimization
|
|
|
|
|
|
class Hyperopt:
|
|
"""
|
|
Hyperopt class, this class contains all the logic to run a hyperopt simulation
|
|
|
|
To run a backtest:
|
|
hyperopt = Hyperopt(config)
|
|
hyperopt.start()
|
|
"""
|
|
def __init__(self, config: Dict[str, Any]) -> None:
|
|
self.config = config
|
|
|
|
self.backtesting = Backtesting(self.config)
|
|
|
|
self.custom_hyperopt = HyperOptResolver(self.config).hyperopt
|
|
|
|
self.custom_hyperoptloss = HyperOptLossResolver(self.config).hyperoptloss
|
|
self.calculate_loss = self.custom_hyperoptloss.hyperopt_loss_function
|
|
|
|
self.trials_file = (self.config['user_data_dir'] /
|
|
'hyperopt_results' / 'hyperopt_results.pickle')
|
|
self.tickerdata_pickle = (self.config['user_data_dir'] /
|
|
'hyperopt_results' / 'hyperopt_tickerdata.pkl')
|
|
self.total_epochs = config.get('epochs', 0)
|
|
|
|
self.current_best_loss = 100
|
|
|
|
if not self.config.get('hyperopt_continue'):
|
|
self.clean_hyperopt()
|
|
else:
|
|
logger.info("Continuing on previous hyperopt results.")
|
|
|
|
self.num_trials_saved = 0
|
|
|
|
# Previous evaluations
|
|
self.trials: List = []
|
|
|
|
# Populate functions here (hasattr is slow so should not be run during "regular" operations)
|
|
if hasattr(self.custom_hyperopt, 'populate_indicators'):
|
|
self.backtesting.strategy.advise_indicators = \
|
|
self.custom_hyperopt.populate_indicators # type: ignore
|
|
if hasattr(self.custom_hyperopt, 'populate_buy_trend'):
|
|
self.backtesting.strategy.advise_buy = \
|
|
self.custom_hyperopt.populate_buy_trend # type: ignore
|
|
if hasattr(self.custom_hyperopt, 'populate_sell_trend'):
|
|
self.backtesting.strategy.advise_sell = \
|
|
self.custom_hyperopt.populate_sell_trend # type: ignore
|
|
|
|
# Use max_open_trades for hyperopt as well, except --disable-max-market-positions is set
|
|
if self.config.get('use_max_market_positions', True):
|
|
self.max_open_trades = self.config['max_open_trades']
|
|
else:
|
|
logger.debug('Ignoring max_open_trades (--disable-max-market-positions was used) ...')
|
|
self.max_open_trades = 0
|
|
self.position_stacking = self.config.get('position_stacking', False)
|
|
|
|
if self.has_space('sell'):
|
|
# Make sure use_sell_signal is enabled
|
|
if 'ask_strategy' not in self.config:
|
|
self.config['ask_strategy'] = {}
|
|
self.config['ask_strategy']['use_sell_signal'] = True
|
|
|
|
self.print_all = self.config.get('print_all', False)
|
|
self.print_colorized = self.config.get('print_colorized', False)
|
|
self.print_json = self.config.get('print_json', False)
|
|
|
|
@staticmethod
|
|
def get_lock_filename(config) -> str:
|
|
|
|
return str(config['user_data_dir'] / 'hyperopt.lock')
|
|
|
|
def clean_hyperopt(self):
|
|
"""
|
|
Remove hyperopt pickle files to restart hyperopt.
|
|
"""
|
|
for f in [self.tickerdata_pickle, self.trials_file]:
|
|
p = Path(f)
|
|
if p.is_file():
|
|
logger.info(f"Removing `{p}`.")
|
|
p.unlink()
|
|
|
|
def _get_params_dict(self, raw_params: List[Any]) -> Dict:
|
|
|
|
dimensions: List[Dimension] = self.dimensions
|
|
|
|
# Ensure the number of dimensions match
|
|
# the number of parameters in the list.
|
|
if len(raw_params) != len(dimensions):
|
|
raise ValueError('Mismatch in number of search-space dimensions.')
|
|
|
|
# Return a dict where the keys are the names of the dimensions
|
|
# and the values are taken from the list of parameters.
|
|
return {d.name: v for d, v in zip(dimensions, raw_params)}
|
|
|
|
def save_trials(self, final: bool = False) -> None:
|
|
"""
|
|
Save hyperopt trials to file
|
|
"""
|
|
num_trials = len(self.trials)
|
|
if num_trials > self.num_trials_saved:
|
|
logger.info(f"Saving {num_trials} {plural(num_trials, 'epoch')}.")
|
|
dump(self.trials, self.trials_file)
|
|
self.num_trials_saved = num_trials
|
|
if final:
|
|
logger.info(f"{num_trials} {plural(num_trials, 'epoch')} "
|
|
f"saved to '{self.trials_file}'.")
|
|
|
|
@staticmethod
|
|
def _read_trials(trials_file) -> List:
|
|
"""
|
|
Read hyperopt trials file
|
|
"""
|
|
logger.info("Reading Trials from '%s'", trials_file)
|
|
trials = load(trials_file)
|
|
return trials
|
|
|
|
def _get_params_details(self, params: Dict) -> Dict:
|
|
"""
|
|
Return the params for each space
|
|
"""
|
|
result: Dict = {}
|
|
|
|
if self.has_space('buy'):
|
|
result['buy'] = {p.name: params.get(p.name)
|
|
for p in self.hyperopt_space('buy')}
|
|
if self.has_space('sell'):
|
|
result['sell'] = {p.name: params.get(p.name)
|
|
for p in self.hyperopt_space('sell')}
|
|
if self.has_space('roi'):
|
|
result['roi'] = self.custom_hyperopt.generate_roi_table(params)
|
|
if self.has_space('stoploss'):
|
|
result['stoploss'] = {p.name: params.get(p.name)
|
|
for p in self.hyperopt_space('stoploss')}
|
|
if self.has_space('trailing'):
|
|
result['trailing'] = self.custom_hyperopt.generate_trailing_params(params)
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
def print_epoch_details(results, total_epochs, print_json: bool,
|
|
no_header: bool = False, header_str: str = None) -> None:
|
|
"""
|
|
Display details of the hyperopt result
|
|
"""
|
|
params = results.get('params_details', {})
|
|
|
|
# Default header string
|
|
if header_str is None:
|
|
header_str = "Best result"
|
|
|
|
if not no_header:
|
|
explanation_str = Hyperopt._format_explanation_string(results, total_epochs)
|
|
print(f"\n{header_str}:\n\n{explanation_str}\n")
|
|
|
|
if print_json:
|
|
result_dict: Dict = {}
|
|
for s in ['buy', 'sell', 'roi', 'stoploss', 'trailing']:
|
|
Hyperopt._params_update_for_json(result_dict, params, s)
|
|
print(rapidjson.dumps(result_dict, default=str, number_mode=rapidjson.NM_NATIVE))
|
|
|
|
else:
|
|
Hyperopt._params_pretty_print(params, 'buy', "Buy hyperspace params:")
|
|
Hyperopt._params_pretty_print(params, 'sell', "Sell hyperspace params:")
|
|
Hyperopt._params_pretty_print(params, 'roi', "ROI table:")
|
|
Hyperopt._params_pretty_print(params, 'stoploss', "Stoploss:")
|
|
Hyperopt._params_pretty_print(params, 'trailing', "Trailing stop:")
|
|
|
|
@staticmethod
|
|
def _params_update_for_json(result_dict, params, space: str):
|
|
if space in params:
|
|
space_params = Hyperopt._space_params(params, space)
|
|
if space in ['buy', 'sell']:
|
|
result_dict.setdefault('params', {}).update(space_params)
|
|
elif space == 'roi':
|
|
# Convert keys in min_roi dict to strings because
|
|
# rapidjson cannot dump dicts with integer keys...
|
|
# OrderedDict is used to keep the numeric order of the items
|
|
# in the dict.
|
|
result_dict['minimal_roi'] = OrderedDict(
|
|
(str(k), v) for k, v in space_params.items()
|
|
)
|
|
else: # 'stoploss', 'trailing'
|
|
result_dict.update(space_params)
|
|
|
|
@staticmethod
|
|
def _params_pretty_print(params, space: str, header: str):
|
|
if space in params:
|
|
space_params = Hyperopt._space_params(params, space, 5)
|
|
if space == 'stoploss':
|
|
print(header, space_params.get('stoploss'))
|
|
else:
|
|
print(header)
|
|
pprint(space_params, indent=4)
|
|
|
|
@staticmethod
|
|
def _space_params(params, space: str, r: int = None) -> Dict:
|
|
d = params[space]
|
|
# Round floats to `r` digits after the decimal point if requested
|
|
return round_dict(d, r) if r else d
|
|
|
|
@staticmethod
|
|
def is_best_loss(results, current_best_loss) -> bool:
|
|
return results['loss'] < current_best_loss
|
|
|
|
def print_results(self, results) -> None:
|
|
"""
|
|
Log results if it is better than any previous evaluation
|
|
"""
|
|
is_best = results['is_best']
|
|
if not self.print_all:
|
|
# Print '\n' after each 100th epoch to separate dots from the log messages.
|
|
# Otherwise output is messy on a terminal.
|
|
print('.', end='' if results['current_epoch'] % 100 != 0 else None) # type: ignore
|
|
sys.stdout.flush()
|
|
|
|
if self.print_all or is_best:
|
|
if not self.print_all:
|
|
# Separate the results explanation string from dots
|
|
print("\n")
|
|
self.print_results_explanation(results, self.total_epochs, self.print_all,
|
|
self.print_colorized)
|
|
|
|
@staticmethod
|
|
def print_results_explanation(results, total_epochs, highlight_best: bool,
|
|
print_colorized: bool) -> None:
|
|
"""
|
|
Log results explanation string
|
|
"""
|
|
explanation_str = Hyperopt._format_explanation_string(results, total_epochs)
|
|
# Colorize output
|
|
if print_colorized:
|
|
if results['total_profit'] > 0:
|
|
explanation_str = Fore.GREEN + explanation_str
|
|
if highlight_best and results['is_best']:
|
|
explanation_str = Style.BRIGHT + explanation_str
|
|
print(explanation_str)
|
|
|
|
@staticmethod
|
|
def _format_explanation_string(results, total_epochs) -> str:
|
|
return (("*" if results['is_initial_point'] else " ") +
|
|
f"{results['current_epoch']:5d}/{total_epochs}: " +
|
|
f"{results['results_explanation']} " +
|
|
f"Objective: {results['loss']:.5f}")
|
|
|
|
def has_space(self, space: str) -> bool:
|
|
"""
|
|
Tell if the space value is contained in the configuration
|
|
"""
|
|
# The 'trailing' space is not included in the 'default' set of spaces
|
|
if space == 'trailing':
|
|
return any(s in self.config['spaces'] for s in [space, 'all'])
|
|
else:
|
|
return any(s in self.config['spaces'] for s in [space, 'all', 'default'])
|
|
|
|
def hyperopt_space(self, space: Optional[str] = None) -> List[Dimension]:
|
|
"""
|
|
Return the dimensions in the hyperoptimization space.
|
|
:param space: Defines hyperspace to return dimensions for.
|
|
If None, then the self.has_space() will be used to return dimensions
|
|
for all hyperspaces used.
|
|
"""
|
|
spaces: List[Dimension] = []
|
|
|
|
if space == 'buy' or (space is None and self.has_space('buy')):
|
|
logger.debug("Hyperopt has 'buy' space")
|
|
spaces += self.custom_hyperopt.indicator_space()
|
|
|
|
if space == 'sell' or (space is None and self.has_space('sell')):
|
|
logger.debug("Hyperopt has 'sell' space")
|
|
spaces += self.custom_hyperopt.sell_indicator_space()
|
|
|
|
if space == 'roi' or (space is None and self.has_space('roi')):
|
|
logger.debug("Hyperopt has 'roi' space")
|
|
spaces += self.custom_hyperopt.roi_space()
|
|
|
|
if space == 'stoploss' or (space is None and self.has_space('stoploss')):
|
|
logger.debug("Hyperopt has 'stoploss' space")
|
|
spaces += self.custom_hyperopt.stoploss_space()
|
|
|
|
if space == 'trailing' or (space is None and self.has_space('trailing')):
|
|
logger.debug("Hyperopt has 'trailing' space")
|
|
spaces += self.custom_hyperopt.trailing_space()
|
|
|
|
return spaces
|
|
|
|
def generate_optimizer(self, raw_params: List[Any], iteration=None) -> Dict:
|
|
"""
|
|
Used Optimize function. Called once per epoch to optimize whatever is configured.
|
|
Keep this function as optimized as possible!
|
|
"""
|
|
params_dict = self._get_params_dict(raw_params)
|
|
params_details = self._get_params_details(params_dict)
|
|
|
|
if self.has_space('roi'):
|
|
self.backtesting.strategy.minimal_roi = \
|
|
self.custom_hyperopt.generate_roi_table(params_dict)
|
|
|
|
if self.has_space('buy'):
|
|
self.backtesting.strategy.advise_buy = \
|
|
self.custom_hyperopt.buy_strategy_generator(params_dict)
|
|
|
|
if self.has_space('sell'):
|
|
self.backtesting.strategy.advise_sell = \
|
|
self.custom_hyperopt.sell_strategy_generator(params_dict)
|
|
|
|
if self.has_space('stoploss'):
|
|
self.backtesting.strategy.stoploss = params_dict['stoploss']
|
|
|
|
if self.has_space('trailing'):
|
|
d = self.custom_hyperopt.generate_trailing_params(params_dict)
|
|
self.backtesting.strategy.trailing_stop = d['trailing_stop']
|
|
self.backtesting.strategy.trailing_stop_positive = d['trailing_stop_positive']
|
|
self.backtesting.strategy.trailing_stop_positive_offset = \
|
|
d['trailing_stop_positive_offset']
|
|
self.backtesting.strategy.trailing_only_offset_is_reached = \
|
|
d['trailing_only_offset_is_reached']
|
|
|
|
processed = load(self.tickerdata_pickle)
|
|
|
|
min_date, max_date = get_timeframe(processed)
|
|
|
|
backtesting_results = self.backtesting.backtest(
|
|
{
|
|
'stake_amount': self.config['stake_amount'],
|
|
'processed': processed,
|
|
'max_open_trades': self.max_open_trades,
|
|
'position_stacking': self.position_stacking,
|
|
'start_date': min_date,
|
|
'end_date': max_date,
|
|
}
|
|
)
|
|
return self._get_results_dict(backtesting_results, min_date, max_date,
|
|
params_dict, params_details)
|
|
|
|
def _get_results_dict(self, backtesting_results, min_date, max_date,
|
|
params_dict, params_details):
|
|
results_metrics = self._calculate_results_metrics(backtesting_results)
|
|
results_explanation = self._format_results_explanation_string(results_metrics)
|
|
|
|
trade_count = results_metrics['trade_count']
|
|
total_profit = results_metrics['total_profit']
|
|
|
|
# If this evaluation contains too short amount of trades to be
|
|
# interesting -- consider it as 'bad' (assigned max. loss value)
|
|
# in order to cast this hyperspace point away from optimization
|
|
# path. We do not want to optimize 'hodl' strategies.
|
|
loss: float = MAX_LOSS
|
|
if trade_count >= self.config['hyperopt_min_trades']:
|
|
loss = self.calculate_loss(results=backtesting_results, trade_count=trade_count,
|
|
min_date=min_date.datetime, max_date=max_date.datetime)
|
|
return {
|
|
'loss': loss,
|
|
'params_dict': params_dict,
|
|
'params_details': params_details,
|
|
'results_metrics': results_metrics,
|
|
'results_explanation': results_explanation,
|
|
'total_profit': total_profit,
|
|
}
|
|
|
|
def _calculate_results_metrics(self, backtesting_results: DataFrame) -> Dict:
|
|
return {
|
|
'trade_count': len(backtesting_results.index),
|
|
'avg_profit': backtesting_results.profit_percent.mean() * 100.0,
|
|
'total_profit': backtesting_results.profit_abs.sum(),
|
|
'profit': backtesting_results.profit_percent.sum() * 100.0,
|
|
'duration': backtesting_results.trade_duration.mean(),
|
|
}
|
|
|
|
def _format_results_explanation_string(self, results_metrics: Dict) -> str:
|
|
"""
|
|
Return the formatted results explanation in a string
|
|
"""
|
|
stake_cur = self.config['stake_currency']
|
|
return (f"{results_metrics['trade_count']:6d} trades. "
|
|
f"Avg profit {results_metrics['avg_profit']: 6.2f}%. "
|
|
f"Total profit {results_metrics['total_profit']: 11.8f} {stake_cur} "
|
|
f"({results_metrics['profit']: 7.2f}\N{GREEK CAPITAL LETTER SIGMA}%). "
|
|
f"Avg duration {results_metrics['duration']:5.1f} mins."
|
|
).encode(locale.getpreferredencoding(), 'replace').decode('utf-8')
|
|
|
|
def get_optimizer(self, dimensions: List[Dimension], cpu_count) -> Optimizer:
|
|
return Optimizer(
|
|
dimensions,
|
|
base_estimator="ET",
|
|
acq_optimizer="auto",
|
|
n_initial_points=INITIAL_POINTS,
|
|
acq_optimizer_kwargs={'n_jobs': cpu_count},
|
|
random_state=self.config.get('hyperopt_random_state', None),
|
|
)
|
|
|
|
def fix_optimizer_models_list(self):
|
|
"""
|
|
WORKAROUND: Since skopt is not actively supported, this resolves problems with skopt
|
|
memory usage, see also: https://github.com/scikit-optimize/scikit-optimize/pull/746
|
|
|
|
This may cease working when skopt updates if implementation of this intrinsic
|
|
part changes.
|
|
"""
|
|
n = len(self.opt.models) - SKOPT_MODELS_MAX_NUM
|
|
# Keep no more than 2*SKOPT_MODELS_MAX_NUM models in the skopt models list,
|
|
# remove the old ones. These are actually of no use, the current model
|
|
# from the estimator is the only one used in the skopt optimizer.
|
|
# Freqtrade code also does not inspect details of the models.
|
|
if n >= SKOPT_MODELS_MAX_NUM:
|
|
logger.debug(f"Fixing skopt models list, removing {n} old items...")
|
|
del self.opt.models[0:n]
|
|
|
|
def run_optimizer_parallel(self, parallel, asked, i) -> List:
|
|
return parallel(delayed(
|
|
wrap_non_picklable_objects(self.generate_optimizer))(v, i) for v in asked)
|
|
|
|
@staticmethod
|
|
def load_previous_results(trials_file) -> List:
|
|
"""
|
|
Load data for epochs from the file if we have one
|
|
"""
|
|
trials: List = []
|
|
if trials_file.is_file() and trials_file.stat().st_size > 0:
|
|
trials = Hyperopt._read_trials(trials_file)
|
|
if trials[0].get('is_best') is None:
|
|
raise OperationalException(
|
|
"The file with Hyperopt results is incompatible with this version "
|
|
"of Freqtrade and cannot be loaded.")
|
|
logger.info(f"Loaded {len(trials)} previous evaluations from disk.")
|
|
return trials
|
|
|
|
def start(self) -> None:
|
|
data, timerange = self.backtesting.load_bt_data()
|
|
|
|
preprocessed = self.backtesting.strategy.tickerdata_to_dataframe(data)
|
|
|
|
# Trim startup period from analyzed dataframe
|
|
for pair, df in preprocessed.items():
|
|
preprocessed[pair] = trim_dataframe(df, timerange)
|
|
min_date, max_date = get_timeframe(data)
|
|
|
|
logger.info(
|
|
'Hyperopting with data from %s up to %s (%s days)..',
|
|
min_date.isoformat(), max_date.isoformat(), (max_date - min_date).days
|
|
)
|
|
dump(preprocessed, self.tickerdata_pickle)
|
|
|
|
# We don't need exchange instance anymore while running hyperopt
|
|
self.backtesting.exchange = None # type: ignore
|
|
|
|
self.trials = self.load_previous_results(self.trials_file)
|
|
|
|
cpus = cpu_count()
|
|
logger.info(f"Found {cpus} CPU cores. Let's make them scream!")
|
|
config_jobs = self.config.get('hyperopt_jobs', -1)
|
|
logger.info(f'Number of parallel jobs set as: {config_jobs}')
|
|
|
|
self.dimensions: List[Dimension] = self.hyperopt_space()
|
|
self.opt = self.get_optimizer(self.dimensions, config_jobs)
|
|
|
|
if self.print_colorized:
|
|
colorama_init(autoreset=True)
|
|
|
|
try:
|
|
with Parallel(n_jobs=config_jobs) as parallel:
|
|
jobs = parallel._effective_n_jobs()
|
|
logger.info(f'Effective number of parallel workers used: {jobs}')
|
|
EVALS = max(self.total_epochs // jobs, 1)
|
|
for i in range(EVALS):
|
|
asked = self.opt.ask(n_points=jobs)
|
|
f_val = self.run_optimizer_parallel(parallel, asked, i)
|
|
self.opt.tell(asked, [v['loss'] for v in f_val])
|
|
self.fix_optimizer_models_list()
|
|
for j in range(jobs):
|
|
# Use human-friendly indexes here (starting from 1)
|
|
current = i * jobs + j + 1
|
|
val = f_val[j]
|
|
val['current_epoch'] = current
|
|
val['is_initial_point'] = current <= INITIAL_POINTS
|
|
logger.debug(f"Optimizer epoch evaluated: {val}")
|
|
|
|
is_best = self.is_best_loss(val, self.current_best_loss)
|
|
# This value is assigned here and not in the optimization method
|
|
# to keep proper order in the list of results. That's because
|
|
# evaluations can take different time. Here they are aligned in the
|
|
# order they will be shown to the user.
|
|
val['is_best'] = is_best
|
|
|
|
self.print_results(val)
|
|
|
|
if is_best:
|
|
self.current_best_loss = val['loss']
|
|
self.trials.append(val)
|
|
# Save results after each best epoch and every 100 epochs
|
|
if is_best or current % 100 == 0:
|
|
self.save_trials()
|
|
except KeyboardInterrupt:
|
|
print('User interrupted..')
|
|
|
|
self.save_trials(final=True)
|
|
|
|
if self.trials:
|
|
sorted_trials = sorted(self.trials, key=itemgetter('loss'))
|
|
results = sorted_trials[0]
|
|
self.print_epoch_details(results, self.total_epochs, self.print_json)
|
|
else:
|
|
# This is printed when Ctrl+C is pressed quickly, before first epochs have
|
|
# a chance to be evaluated.
|
|
print("No epochs evaluated yet, no best result.")
|