# source: https://raw.githubusercontent.com/Industrial/autofreq/2531895e1048fe30a1b85dfa1bc34f23b2bc529d/user_data/strategies/HyperOptableStrategy.py # -*- coding: utf-8 -*- import json import os from datetime import datetime from typing import Any, Dict, List, Optional from pandas import DataFrame, concat from freqtrade.constants import Config from freqtrade.enums import RunMode from freqtrade.persistence import Trade from freqtrade.strategy import IStrategy from user_data.dictionary import deep_get, deep_set from user_data.symbol import filename_to_symbol class github_Industrial_autofreq__HyperOptableStrategy__20230415_222203(IStrategy): # This var holds the temporary value for fetching the options per pair. It # will get set from different points in the code. last_pair = "BTC/USDT" # Holds all the opts per pair. opt = {} # Set the parameters. parameters: List[str] = [] def _merge_pair_opts(self, default_values: Dict, hyperopted_values: Dict): """ Merges the hyperopted values over the default ones. """ params = {} default_params = default_values.get("params") if default_params: for k in default_params: params[k] = default_params[k] hyperopted_params = hyperopted_values.get("params") if hyperopted_params: for k in hyperopted_params: params[k] = hyperopted_params[k] return {"params": params} def _load_opts_from_file(self): """ Loads hyperopted parameters from a file. """ class_name = self.__class__.__name__ timeframe = self.timeframe opt_directory_path = ( f"{os.getcwd()}/user_data/strategies/{class_name}/opt/{timeframe}" ) for file_name in os.listdir(opt_directory_path): file_path = f"{opt_directory_path}/{file_name}" if os.path.isfile(file_path): with open(file_path, "r") as file: name = filename_to_symbol(os.path.splitext(file_name)[0]) hyperopted_values = json.load(file) default_values = self.opt.get(name) or {} self.opt[name] = self._merge_pair_opts( default_values, hyperopted_values ) def _get_opt(self, key: str) -> Any: """ Gets a dot separated key from the options for the current pair. """ options = self.opt[self.last_pair] value = deep_get(options, key) return value def _set_opt(self, key: str, value: Any) -> None: """ Sets a dot separated key to the value in the options for the current pair. """ options = deep_get(self.opt, self.last_pair, {}) deep_set(options, key, value) deep_set(self.opt, self.last_pair, options) def _get_parameter_value(self, name: str): """ Returns the value of a parameter. In Hyperopt mode returns the randomized value used in the hyperopt process. In backtest and live mode returns the saved hyperopted value. """ mode = self.config["runmode"] attribute = getattr(self, name) category = attribute.category if mode == RunMode.HYPEROPT: return attribute.value else: return self._get_opt(f"params.{category}.{name}") # TODO: Refactor. Unfriendly API. def _add_indicator(self, dataframe: DataFrame, name: str, prefix: str, fn): dfs = [dataframe] value = self._get_parameter_value(name) df = DataFrame() df[f"{prefix}{value}"] = fn(dataframe, value) dfs.append(df) for i in getattr(self, name).range: if i != value: df = DataFrame() df[f"{prefix}{i}"] = fn(dataframe, i) dfs.append(df) dataframe = concat(dfs, axis=1) return dataframe def _set_parameters(self): for parameter_name in self.parameters: parameter = getattr(self, parameter_name) self._set_opt( f"params.{parameter.category}.{parameter_name}", parameter.value ) def __init__(self, config: Config) -> None: super().__init__(config) self.timeframe = str(config["timeframe"]) #print("timeframe", self.timeframe) for pair in self.config["pairs"]: self.last_pair = pair self.set_defaults() self._set_parameters() self._load_opts_from_file() self.last_pair = config["pairs"][0] def set_defaults(self): """ Sets default values of minimal_roi, stoploss and trailing_stop variables. This will get called for each pair. """ self.minimal_roi = { "0": 0.05, "5": 0.04, "10": 0.03, "30": 0.02, "60": 0.01, } self.stoploss = -0.03 self.trailing_stop = True self.trailing_stop_positive_offset = 0.01 self.trailing_stop_positive = 0.005 self.trailing_only_offset_is_reached = False def bot_start(self, **kwargs) -> None: """ Called only once after bot instantiation. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. """ for pair in self.config["pairs"]: self.last_pair = pair self.minimal_roi = {int(k): v for k, v in self.minimal_roi.items()} pass @property def minimal_roi(self): return self._get_opt("params.roi") @minimal_roi.setter def minimal_roi(self, value): self._set_opt("params.roi", value) @property def stoploss(self): return self._get_opt("params.stoploss.stoploss") @stoploss.setter def stoploss(self, value): self._set_opt("params.stoploss.stoploss", value) @property def trailing_stop(self): return self._get_opt("params.trailing.trailing_stop") @trailing_stop.setter def trailing_stop(self, value): self._set_opt("params.trailing.trailing_stop", value) @property def trailing_stop_positive_offset(self): return self._get_opt("params.trailing.trailing_stop_positive_offset") @trailing_stop_positive_offset.setter def trailing_stop_positive_offset(self, value): self._set_opt("params.trailing.trailing_stop_positive_offset", value) @property def trailing_stop_positive(self): return self._get_opt("params.trailing.trailing_stop_positive") @trailing_stop_positive.setter def trailing_stop_positive(self, value): self._set_opt("params.trailing.trailing_stop_positive", value) @property def trailing_only_offset_is_reached(self): return self._get_opt("params.trailing.trailing_only_offset_is_reached") @trailing_only_offset_is_reached.setter def trailing_only_offset_is_reached(self, value): self._set_opt("params.trailing.trailing_only_offset_is_reached", value) def ft_stoploss_adjust( self, current_rate: float, trade: Trade, current_time: datetime, current_profit: float, force_stoploss: float, low: Optional[float] = None, high: Optional[float] = None, ) -> None: self.last_pair = trade.pair super().ft_stoploss_adjust( current_rate, trade, current_time, current_profit, force_stoploss ) def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs, ) -> bool: self.last_pair = pair return super().confirm_trade_entry( pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side ) def confirm_trade_exit( self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs, ) -> bool: self.last_pair = pair return super().confirm_trade_exit( pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, ) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.last_pair = metadata["pair"] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.last_pair = metadata["pair"] return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.last_pair = metadata["pair"] return dataframe def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs, ) -> float: self.last_pair = pair return proposed_stake