# source: https://raw.githubusercontent.com/DerSalvador/freqtrade-helm-chart/f80d4d8b77c53435e9c0a9045636f1bfb2b8c539/chart/deployed_strategies/binance-futures-k8s-namespace/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_DerSalvador_freqtrade_helm_chart__Uptrend__20260416_224245(IStrategy): INTERFACE_VERSION = 3 entry_params = {'entry_rsi_uplimit': 50} entry_rsi_uplimit = IntParameter(50, 90, default=entry_params['entry_rsi_uplimit'], optimize=False, space='entry') # 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.1 # 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_exit_signal = True exit_profit_only = False ignore_roi_if_entry_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_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # uptrend # Make sure Volume is not 0 dataframe.loc[(dataframe['rsi'] < 80) & (dataframe['mama'] > dataframe['fama']) & (dataframe['mama_diff_ratio'] > 0.04) & (dataframe['volume'] > 0), 'entry'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Make sure Volume is not 0 dataframe.loc[(dataframe['mama_diff_ratio'] < 0.01) & (dataframe['volume'] > 0), 'exit'] = 1 return dataframe import random from functools import reduce class SuperBuy(Github_DerSalvador_freqtrade_helm_chart__Uptrend__20260416_224245): """ Idea is to build random entry signales from populate_indicators, with luck we'll get a good entry signal """ generator = IntParameter(0, 100000000000, default=99295874569, optimize=True, space='entry') # generate unique matrix of conditions for your dataframe operators_used_to_compare_between_columns = IntParameter(0, 3, default=3, optimize=True, space='entry') # number of conditions you will keep to build entry signal operators_used_to_with_best_point = IntParameter(0, 3, default=1, optimize=True, space='entry') # number of conditions you will keep to build entry signal condition_selector = IntParameter(0, 100, default=50, optimize=True, space='entry') # how to select the desired conditions beteween all conditions generated (seed random) best_entry_point = None best_entry_point_dict = dict() bad_entry_point_dict = dict() all_points_dict = dict() entry_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: '!='} # best point criteria # parameters selection criteria top_index_criteria = {'min_close_hh_ratio': 0.08, 'max_candles_to_get_ratio': 8, 'candles_after_dip_to_entry': 0, '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_entry'] 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_entry_point_dict: self.best_entry_point_dict[current_pair], self.bad_entry_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_entry_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_entry_point_dict: continue if all_best_points is None: all_best_points = self.best_entry_point_dict[pair] all_bad_points = self.bad_entry_point_dict[pair] all_points = self.all_points_dict[pair] else: all_best_points = all_best_points.append(self.best_entry_point_dict[pair]) all_bad_points = all_bad_points.append(self.bad_entry_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 ['entry', 'entry_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) # the part of best points should be at least 3% for the value # proportion of value is X* more important in best points than in bad points # minimum number of the value (because we don't want close=1.121213243482902183 as result) values = df_all.query(f'part_of_best_points > 3 & part_of_best_points_percent > 13 & best_points > 12').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 ['entry', 'entry_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 entry signal:') print('( # main entry 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_superentry_signal(self, dataframe: DataFrame, metadata: dict) -> list: # every indicators names columns = list(dataframe.columns) columns.remove('date') columns.remove('exit') columns.remove('entry') columns.remove('entry_tag') columns = [column for column in columns if not 'date' in column] # generated random conditions entry_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 entry 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_entry_point is None: try: top_index, _ = self.find_best_entry_point(dataframe, metadata)['shifted_index'].iloc[0] self.best_entry_point = dataframe.iloc[top_index] print(f"pair used as reference is {metadata['pair']}") print(top_index) except: self.best_entry_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_entry_point is not None: for column in self.columns_to_compare_to_best_point: if str(self.best_entry_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 entry 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 + "'])") entry_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 + "'])") entry_conds.append("(dataframe['" + left_elt + "'] " + self.operators[int(generator)] + " dataframe['" + right_elt + "'])") entry_conds_best_point = [] # generate entry conditions with best entry point for column in self.columns_to_compare_to_best_point: if self.best_entry_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_entry_point['" + column + "']))") # print(eval("best_entry_point['" + column + "']")) entry_conds_best_point.append("(dataframe['" + column + "'] " + self.operators[int(generator)] + ' ' + str(self.best_entry_point[column]) + ')') # select a few entry conditions random.seed(self.condition_selector.value) try: entry_conds = random.sample(entry_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: entry_conds += random.sample(entry_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.entry_signal_already_printed != entry_conds: print(entry_conds) self.entry_signal_already_printed = entry_conds try: entry_conds = [eval(entry_cond, globals(), {'dataframe': dataframe, 'best_entry_point': self.best_entry_point}) for entry_cond in entry_conds] except: return [] return entry_conds def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: entry_conds = self.generate_superentry_signal(dataframe, metadata) if self.config['runmode'].value in 'backtest': # backtest, we want to check must common entry tags... dataframe = super().populate_entry_trend(dataframe, metadata) # get entry tags self.common_points_for_every_best_entry(dataframe, metadata, self.columns + ['entry', 'entry_tag']) elif self.config['runmode'].value in 'hyperopt': # hyperopt, we want to test new entry signals # main entry signals found # protections # (dataframe['ema_50_lin'] < dataframe['ema_26_lin']) | # (dataframe['sma_15'] > dataframe['ema_50_1h']) | # (dataframe['ema_slow'] <= dataframe['sup1']) | is_additional_check = (dataframe['not_res1_1h'] == True) & ((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['res1'] >= dataframe['bb_upperband2_1h']) if entry_conds: dataframe.loc[is_additional_check & reduce(lambda x, y: x & y, entry_conds), 'entry'] = 1 # THIS STRAT SHOULD NOT BE USED IN LIVE/DRYRUN MODE return dataframe