from freqtrade.strategy import IStrategy, CategoricalParameter, IntParameter, merge_informative_pair from pandas import DataFrame, Series import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from functools import reduce import numpy as np INDICATORS = ('EMA', 'SMA') TIMEFRAMES = ('5m', '15m', '1h', '4h', '1d') BASE_TIMEFRAME = TIMEFRAMES[0] INFO_TIMEFRAMES = TIMEFRAMES[1:] TIMEFRAMES_LEN = len(TIMEFRAMES) SELL_TIMEFRAME = '1h' PERIODS = [] n = 5 for i in range(1, 15): PERIODS.append(n) n += i PERIODS_LEN = len(PERIODS) MAX_CONDITIONS = TIMEFRAMES_LEN buy = { "buy_fperiod_0": 11, "buy_fperiod_1": 0, "buy_fperiod_2": 4, "buy_fperiod_3": 4, "buy_fperiod_4": 2, "buy_indicator_0": "SMA", "buy_indicator_1": "EMA", "buy_indicator_2": "SMA", "buy_indicator_3": "EMA", "buy_indicator_4": "SMA", "buy_speriod_0": 0, "buy_speriod_1": 1, "buy_speriod_2": 10, "buy_speriod_3": 7, "buy_speriod_4": 5 } sell = { "sell_fperiod_0": 1, "sell_indicator_0": "EMA", "sell_speriod_0": 10 } def normalize(df): df = (df-df.min())/(df.max()-df.min()) return df def apply_indicator(dataframe: DataFrame, key: str, indicator: str, period: int): if key in dataframe.keys(): return result = getattr(ta, indicator)(dataframe, timeperiod=period) dataframe[key] = result def greater_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return (dataframe[main_indicator] > dataframe[crossed_indicator]) def true_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return (dataframe['volume'] > 10) def close_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return (np.isclose(dataframe[main_indicator], dataframe[crossed_indicator])) def crossed_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return ( (qtpylib.crossed_below(dataframe[main_indicator], dataframe[crossed_indicator])) | (qtpylib.crossed_above( dataframe[main_indicator], dataframe[crossed_indicator])) ) def crossed_above_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return ( qtpylib.crossed_above( dataframe[main_indicator], dataframe[crossed_indicator]) ) def crossed_below_operator(dataframe: DataFrame, main_indicator: str, crossed_indicator: str): return ( qtpylib.crossed_below( dataframe[main_indicator], dataframe[crossed_indicator]) ) OPERATORS = { 'D': true_operator, '>': greater_operator, '=': close_operator, 'C': crossed_operator, 'CA': crossed_above_operator, 'CB': crossed_below_operator, } def apply_operator(dataframe: DataFrame, main_indicator, crossed_indicator, operator) -> tuple[Series, DataFrame]: condition = OPERATORS[operator]( dataframe, main_indicator, crossed_indicator) return condition, dataframe class DefaultValue: def __init__(self, value) -> None: self.value = value def get_parameter_keys(trend: str, condition_idx: int): k_1 = f'{trend}_indicator_{condition_idx}' k_2 = f'{trend}_fperiod_{condition_idx}' k_3 = f'{trend}_speriod_{condition_idx}' return k_1, k_2, k_3 def set_hyperopt_parameters(self): trend = 'buy' for condition_idx in range(MAX_CONDITIONS): k_1, k_2, k_3 = get_parameter_keys(trend, condition_idx) if condition_idx < 4: setattr(self, k_1, CategoricalParameter(INDICATORS, space=trend)) setattr(self, k_2, IntParameter( 0, PERIODS_LEN - 1, space=trend, default=0)) setattr(self, k_3, IntParameter( 0, PERIODS_LEN - 1, space=trend, default=0)) else: # use default for 1 day timeframe setattr(self, k_1, DefaultValue(buy[k_1])) setattr(self, k_2, DefaultValue(buy[k_2])) setattr(self, k_3, DefaultValue(buy[k_3])) trend = 'sell' condition_idx = 0 k_1, k_2, k_3 = get_parameter_keys(trend, condition_idx) setattr(self, k_1, CategoricalParameter(INDICATORS, space=trend)) setattr(self, k_2, IntParameter(0, PERIODS_LEN - 1, space=trend, default=0)) setattr(self, k_3, IntParameter(0, PERIODS_LEN - 1, space=trend, default=0)) return self @set_hyperopt_parameters class MyStrategyNew10(IStrategy): INTERFACE_VERSION = 2 minimal_roi = {"0": 1} trailing_stop = False trailing_stop_positive = 0.03 trailing_stop_positive_offset = 0.05 trailing_only_offset_is_reached = False stoploss = -1 timeframe = BASE_TIMEFRAME process_only_new_candles = True use_sell_signal = True sell_profit_only = False ignore_roi_if_buy_signal = True startup_candle_count = 500 def __init__(self, config: dict) -> None: super().__init__(config) print("Self:", self.__dict__) def get_hyperopt_parameters(self, trend: str, condition_idx: int): k_1, k_2, k_3 = get_parameter_keys(trend, condition_idx) indicator = getattr(self, k_1).value fperiod = getattr(self, k_2).value speriod = getattr(self, k_3).value return indicator, fperiod, speriod def get_indicators_pair(self, trend: str, condition_idx: int) -> tuple[str, str, str]: indicator, fperiod, speriod = self.get_hyperopt_parameters( trend, condition_idx) if fperiod == speriod: return None, None, None if trend == 'sell': operator = 'CB' sell_timeframe = SELL_TIMEFRAME main_indicator = f'{indicator}_{PERIODS[fperiod]}_{sell_timeframe}' crossed_indicator = f'{indicator}_{PERIODS[speriod]}_{sell_timeframe}' else: operator = 'CA' if condition_idx == 0 else '>' main_indicator = f'{indicator}_{PERIODS[fperiod]}_{TIMEFRAMES[condition_idx]}' crossed_indicator = f'{indicator}_{PERIODS[speriod]}_{TIMEFRAMES[condition_idx]}' return main_indicator, crossed_indicator, operator def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, "DataProvider is required for multiple timeframes." avalidable_indicators = set() avalidable_info_timeframes = set() avalidable_periods = set() run_mode = self.dp.runmode.value if run_mode in ('backtest', 'live', 'dry_run'): trend = 'buy' for condition_idx in range(MAX_CONDITIONS): indicator, fperiod, speriod = self.get_hyperopt_parameters( trend, condition_idx) avalidable_indicators.add(indicator) avalidable_info_timeframes.add(TIMEFRAMES[condition_idx]) avalidable_periods.add(PERIODS[fperiod]) avalidable_periods.add(PERIODS[speriod]) trend = 'sell' indicator, fperiod, speriod = self.get_hyperopt_parameters( trend, 0) avalidable_indicators.add(indicator) avalidable_periods.add(PERIODS[fperiod]) avalidable_periods.add(PERIODS[speriod]) else: avalidable_indicators = INDICATORS avalidable_info_timeframes = INFO_TIMEFRAMES avalidable_periods = PERIODS for info_timeframe in avalidable_info_timeframes: info_dataframe = self.dp.get_pair_dataframe( pair=metadata['pair'], timeframe=info_timeframe) for indicator in avalidable_indicators: for period in avalidable_periods: apply_indicator( info_dataframe, f'{indicator}_{period}', indicator, period) dataframe = merge_informative_pair( dataframe, info_dataframe, self.timeframe, info_timeframe, ffill=True) for indicator in avalidable_indicators: for period in avalidable_periods: apply_indicator( dataframe, f'{indicator}_{period}_{BASE_TIMEFRAME}', indicator, period) return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: trend = 'buy' conditions = list() for condition_idx in range(MAX_CONDITIONS): main_indicator, crossed_indicator, operator = self.get_indicators_pair( trend, condition_idx) if not operator: continue condition, dataframe = apply_operator( dataframe, main_indicator, crossed_indicator, operator) conditions.append(condition) if conditions: dataframe.loc[reduce(lambda x, y: x & y, conditions), 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = list() condition_idx = 0 main_indicator, crossed_indicator, operator = self.get_indicators_pair( 'sell', condition_idx) if operator: condition, dataframe = apply_operator( dataframe, main_indicator, crossed_indicator, operator) conditions.append(condition) # bitwaise not condition if conditions: dataframe.loc[reduce(lambda x, y: x & y, conditions), 'sell'] = 1 return dataframe