# source: https://raw.githubusercontent.com/cyberjunky/willemstijn_user_data/758a23f80ce7a5dca176e939c4f3f8562d11a304/stratscorer/all_strategies/Uptrend.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # --- Do not remove these libs --- import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame, Series # noqa from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter) # -------------------------------- # Add your lib to import here import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from datetime import datetime class Github_cyberjunky_willemstijn_user_data__Uptrend__20230822_152729(IStrategy): INTERFACE_VERSION = 2 buy_params = { 'buy_rsi_uplimit': 50, } buy_rsi_uplimit = IntParameter(50, 90, default=buy_params['buy_rsi_uplimit'], optimize=False, space='buy') # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi". minimal_roi = { "60": 0.1, "30": 0.02, "0": 0.04 } # Optimal stoploss designed for the strategy. # This attribute will be overridden if the config file contains "stoploss". stoploss = -0.10 # Trailing stoploss trailing_stop = False # trailing_only_offset_is_reached = False # trailing_stop_positive = 0.01 # trailing_stop_positive_offset = 0.0 # Disabled / not configured # Optimal timeframe for the strategy. timeframe = '5m' # Run "populate_indicators()" only for new candle. process_only_new_candles = False # These values can be overridden in the "ask_strategy" section in the config. use_sell_signal = True sell_profit_only = False ignore_roi_if_buy_signal = False # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 30 use_custom_stoploss = True def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: sl_new = 1 if (current_profit > 0.2): sl_new = 0.05 elif (current_profit > 0.1): sl_new = 0.03 elif (current_profit > 0.06): sl_new = 0.02 elif (current_profit > 0.03): sl_new = 0.015 elif (current_profit > 0.015): sl_new = 0.0075 return sl_new def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['hl2'] = (dataframe['high'] + dataframe['low']) / 2 dataframe['mama'], dataframe['fama'] = ta.MAMA(dataframe['hl2'], 0.5, 0.05) dataframe['mama_diff'] = dataframe['mama'] - dataframe['fama'] dataframe['mama_diff_ratio'] = dataframe['mama_diff'] / dataframe['hl2'] dataframe['zero'] = 0 dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) # EMA 50 dataframe['ema50'] = ta.EMA(dataframe['close'], timeperiod=50) # EMA 200 dataframe['ema200'] = ta.EMA(dataframe['close'], timeperiod=200) return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['rsi'] < 80) & (dataframe['mama'] > dataframe['fama']) & # uptrend (dataframe['mama_diff_ratio'] > 0.04) & (dataframe['volume'] > 0) # Make sure Volume is not 0 ), 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['mama_diff_ratio'] < 0.01) & (dataframe['volume'] > 0) # Make sure Volume is not 0 ), 'sell'] = 1 return dataframe import random from functools import reduce class SuperBuy(Github_cyberjunky_willemstijn_user_data__Uptrend__20230822_152729): """ Idea is to build random buy signales from populate_indicators, with luck we'll get a good buy signal """ generator = IntParameter(0, 100000000000, default=99295874569, optimize=True, space='buy') # generate unique matrix of conditions for your dataframe operators_used_to_compare_between_columns = IntParameter(0, 3, default=3, optimize=True, space='buy') # number of conditions you will keep to build buy signal operators_used_to_with_best_point = IntParameter(0, 3, default=1, optimize=True, space='buy') # number of conditions you will keep to build buy signal condition_selector = IntParameter(0, 100, default=50, optimize=True, space='buy') # how to select the desired conditions beteween all conditions generated (seed random) best_buy_point = None best_buy_point_dict = dict() bad_buy_point_dict = dict() all_points_dict = dict() buy_signal_already_printed = False columns = [] columns_to_compare_to_best_point = [] columns_to_compare_to_volume = [] columns_to_compare_to_price = [] operators = { 0: '<', 1: '>', 2: '<=', 3: '>=', 4: '==', 5: '!=' } top_index_criteria = { # best point criteria 'min_close_hh_ratio': 0.08, 'max_candles_to_get_ratio': 8, 'candles_after_dip_to_buy': 0, # parameters selection criteria 'select_parameter_if_in_more_than_x_percent_of_best_points': 97, 'select_prameter_if_prop_is_x_percent_higher_in_best_points': 10, } def find_best_entry_point(self, dataframe: DataFrame, metadata: dict): lookahead_candles = self.top_index_criteria['max_candles_to_get_ratio'] workdataframe = dataframe.copy() workdataframe['higher_high'] = workdataframe['high'].rolling(lookahead_candles).max() workdataframe['close_shifted_lookehead'] = workdataframe['close'].shift(lookahead_candles) workdataframe['higher_high_close_ratio'] = workdataframe['higher_high'] / workdataframe['close_shifted_lookehead'] df_mask = workdataframe['higher_high_close_ratio'] >= 1 + self.top_index_criteria['min_close_hh_ratio'] # print(1 + self.top_index_criteria['min_close_hh_ratio']) filtered_df = workdataframe[df_mask] filtered_df = filtered_df.sort_values(by=["higher_high_close_ratio"], ascending=False) filtered_df["shifted_index"] = filtered_df.index - lookahead_candles + self.top_index_criteria['candles_after_dip_to_buy'] if filtered_df.empty: if self.config['runmode'].value != 'hyperopt': print("No entry point found for {}".format(metadata['pair'])) return filtered_df, workdataframe # print(metadata['pair']) # print(filtered_df[["date", "shifted_index", "higher_high_close_ratio", "close_shifted_lookehead", "close", "higher_high"]]) return filtered_df, workdataframe.drop(filtered_df['shifted_index']) def common_points_for_every_best_entry(self, dataframe: DataFrame, metadata: dict, columns: list) -> list: full_pairlist = self.dp.current_whitelist() current_pair = metadata['pair'] if current_pair not in self.best_buy_point_dict: self.best_buy_point_dict[current_pair], self.bad_buy_point_dict[current_pair] = self.find_best_entry_point(dataframe, metadata) self.all_points_dict[current_pair] = dataframe.copy() for pair in full_pairlist: current_df = self.dp.get_pair_dataframe(pair=pair, timeframe=self.timeframe) if (pair not in self.best_buy_point_dict) and not current_df.empty: # print("No entry point found for {}".format(pair)) return [] all_best_points = None all_bad_points = None all_points = None for pair in full_pairlist: # NO DATA FOR THIS PAIR if not pair in self.best_buy_point_dict: continue if all_best_points is None: all_best_points = self.best_buy_point_dict[pair] all_bad_points = self.bad_buy_point_dict[pair] all_points = self.all_points_dict[pair] else: all_best_points = all_best_points.append(self.best_buy_point_dict[pair]) all_bad_points = all_bad_points.append(self.bad_buy_point_dict[pair]) all_points = all_points.append(self.all_points_dict[pair]) print("HERE COMMON VALUES FOR ALL BEST POINTS !!!!!!!!!!") res = list() best_indicators = [] for column in columns: all_points_values_count = all_points[column].value_counts() all_bad_points_values_count = all_bad_points[column].value_counts() count = all_best_points[column].value_counts() # keep only values in more than x% of all points df_mask = count >= 1 / 100 * all_best_points.shape[0] count = count[df_mask] if not count.empty and not column in ['buy', 'buy_tag']: count_normalized = count / all_best_points.shape[0] all_bad_points_values_count_normalized = all_bad_points_values_count / all_bad_points.shape[0] df_all = count.to_frame(name='best_points').join(all_bad_points_values_count.to_frame(name='bad_points')) df_all['best_point_percent'] = count_normalized df_all['bad_point_percent'] = all_bad_points_values_count_normalized df_all['part_of_best_points'] = 100 * df_all['best_points'] / (df_all['best_points'] + df_all['bad_points']) df_all['part_of_best_points_percent'] = count_normalized / all_bad_points_values_count_normalized print(column) print(df_all) values = df_all.query( f"part_of_best_points > 3 & " # the part of best points should be at least 3% for the value "part_of_best_points_percent > 13 & " # proportion of value is X* more important in best points than in bad points "best_points > 12" # minimum number of the value (because we don't want close=1.121213243482902183 as result) ).index.tolist() # & df_all["part_of_best_points_percent"] > 10)] print(values) for elt in values: best_indicators.append(f"dataframe['{column}'] == {elt}") if column in ['buy', 'buy_tag']: print(column) print(all_best_points[column].value_counts()) elif not count.empty: res.append({ 'column': column, 'value': count.index[0], 'ratio_for_best': count.iloc[0] / all_best_points.shape[0], 'ratio_for_all': all_points_values_count[count.index[0]] / all_points.shape[0], 'ratio_diff': count.iloc[0] / all_best_points.shape[0] - all_points_values_count[count.index[0]] / all_points.shape[0] }) for item in sorted(res, key=lambda x: x['ratio_diff']): print(f"({item['column']} == {item['value']}), {100 * item['ratio_for_best']:.2f}% in best vs {100 * item['ratio_for_all']:.2f}% average") print("END OF COMMON VALUES FOR ALL BEST POINTS !!!!!!!!!!") print("Suggested buy signal:") print("( # main buy signals found") for item in best_indicators: print(f" ({item})|") print(")") print("& ( # protections") for item in sorted(res, key=lambda x: x['ratio_diff']): if (((100 * item['ratio_for_best']) - (100 * item['ratio_for_all'])) > self.top_index_criteria['select_prameter_if_prop_is_x_percent_higher_in_best_points']) and (100 * item['ratio_for_best']) > self.top_index_criteria['select_parameter_if_in_more_than_x_percent_of_best_points']: print(f"(dataframe['{item['column']}'] == {item['value']}) &") print(")") return [] def is_same_dimension_as_price(self, dataframe: DataFrame, column_name: str) -> bool: if dataframe['close'].dtype != dataframe[column_name].dtype: # prevent impossible comparisons return False return (dataframe[column_name].max() <= dataframe['high'].max() and dataframe[column_name].min() >= dataframe['low'].min()) def is_same_dimension_as_volume(self, dataframe: DataFrame, column_name: str) -> bool: if dataframe['volume'].dtype != dataframe[column_name].dtype: # prevent impossible comparisons return False if 'volume' in column_name: return True return False def generate_superbuy_signal(self, dataframe: DataFrame, metadata: dict) -> list: # every indicators names columns = list(dataframe.columns) columns.remove('date') columns.remove('sell') columns.remove('buy') columns.remove('buy_tag') columns = [column for column in columns if not 'date' in column] # generated random conditions buy_conds = [] # operators we will use as a string "123423232" which will be used to sequentially pick in operators generators = "" base_generators = str(self.generator.value) while len(generators) < len(columns) * len(columns): generators = generators + base_generators # get best buy point for first pair, will indicators will be used for each pair # THE PAIR YOU WANT TO USE AS REFERENCE MUST BE FIRST IN YOUR PAIRLIST !!!!!!!! if self.best_buy_point is None: try: top_index, _ = self.find_best_entry_point(dataframe, metadata)["shifted_index"].iloc[0] self.best_buy_point = dataframe.iloc[top_index] print(f"pair used as reference is {metadata['pair']}") print(top_index) except: self.best_buy_point = None pass # sort columns by category if len(self.columns_to_compare_to_best_point) == 0 and len(self.columns_to_compare_to_volume) == 0 and len(self.columns_to_compare_to_price) == 0: self.columns = columns for column in columns: if self.is_same_dimension_as_price(dataframe, column): self.columns_to_compare_to_price.append(column) elif self.is_same_dimension_as_volume(dataframe, column): self.columns_to_compare_to_volume.append(column) else: self.columns_to_compare_to_best_point.append(column) print(f"columns_to_compare_to_price : {self.columns_to_compare_to_price}") print(f"columns_to_compare_to_volume : {self.columns_to_compare_to_volume}") # remove NAN columns for best point... if self.best_buy_point is not None: for column in self.columns_to_compare_to_best_point: if str(self.best_buy_point[column]) == 'nan': self.columns_to_compare_to_best_point.remove(column) print(f"columns_to_compare_to_best_point : {self.columns_to_compare_to_best_point}") # generate matrix of all operators for all combinations of columns and create buy conditions index = 0 for left_elt in self.columns_to_compare_to_price: for right_elt in self.columns_to_compare_to_price: if index > len(generators): break generator = generators[index] index += 1 if left_elt == right_elt: continue if int(generator) not in self.operators: # pass if no operator is selected continue # print("(dataframe['" + left_elt + "'] " + self.operators[int(generator)] + " dataframe['" + right_elt + "'])") buy_conds.append( "(dataframe['" + left_elt + "'] " + self.operators[int(generator)] + " dataframe['" + right_elt + "'])" ) for left_elt in self.columns_to_compare_to_volume: for right_elt in self.columns_to_compare_to_volume: if index > len(generators): break generator = generators[index] index += 1 if left_elt == right_elt: continue if int(generator) not in self.operators: # pass if no operator is selected continue # print("(dataframe['" + left_elt + "'] " + self.operators[int(generator)] + " dataframe['" + right_elt + "'])") buy_conds.append( "(dataframe['" + left_elt + "'] " + self.operators[int(generator)] + " dataframe['" + right_elt + "'])" ) buy_conds_best_point = [] # generate buy conditions with best buy point for column in self.columns_to_compare_to_best_point: if self.best_buy_point is None: continue if index > len(generators): break generator = generators[index] index += 1 if int(generator) not in self.operators: # pass if no operator is selected continue # print("(dataframe['" + column + "'] " + self.operators[int(generator)] + " best_buy_point['" + column + "']))") # print(eval("best_buy_point['" + column + "']")) buy_conds_best_point.append( "(dataframe['" + column + "'] " + self.operators[int(generator)] + " " + str(self.best_buy_point[column]) + ")" ) # select a few buy conditions random.seed(self.condition_selector.value) try: buy_conds = random.sample(buy_conds, self.operators_used_to_compare_between_columns.value) except ValueError: print("not enough conditions to compare between columns") # Sample larger than population or is negative pass try: buy_conds += random.sample(buy_conds_best_point, self.operators_used_to_with_best_point.value) except ValueError as e: if self.config['runmode'].value != 'hyperopt': print("not enough conditions to compare with best point") # Sample larger than population or is negative pass if self.config['runmode'].value in ('backtest', 'hyperopt') and self.buy_signal_already_printed != buy_conds: print(buy_conds) self.buy_signal_already_printed = buy_conds try: buy_conds = [eval(buy_cond, globals(), {'dataframe': dataframe, 'best_buy_point': self.best_buy_point}) for buy_cond in buy_conds] except: return [] return buy_conds def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: buy_conds = self.generate_superbuy_signal(dataframe, metadata) if self.config['runmode'].value in ('backtest'): # backtest, we want to check must common buy tags... dataframe = super().populate_buy_trend(dataframe, metadata) # get buy tags self.common_points_for_every_best_entry(dataframe, metadata, self.columns + ['buy', 'buy_tag']) elif self.config['runmode'].value in ('hyperopt'): # hyperopt, we want to test new buy signals is_additional_check = ( ( # main buy signals found (dataframe['not_res1_1h'] == True) ) & ( # protections (dataframe['rsi_fast_lower_20'] == 0) & (dataframe['rsi_fast_lower_30'] == 0) & (dataframe['r_14_lower_minus_80'] == 0) & (dataframe['r_32_lower_minus_80'] == 0) & (dataframe['r_96_lower_minus_80'] == 0) ) & ( # (dataframe['ema_50_lin'] < dataframe['ema_26_lin']) | # (dataframe['sma_15'] > dataframe['ema_50_1h']) | # (dataframe['ema_slow'] <= dataframe['sup1']) | (dataframe['res1'] >= dataframe['bb_upperband2_1h']) ) ) if buy_conds: dataframe.loc[ is_additional_check & reduce(lambda x, y: x & y, buy_conds) , 'buy'] = 1 # THIS STRAT SHOULD NOT BE USED IN LIVE/DRYRUN MODE return dataframe