import logging import numpy as np import pandas as pd import talib.abstract as ta import time # from freqtrade.persistence import Trade from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, merge_informative_pair from functools import reduce from pandas import DataFrame from scipy.signal import find_peaks from technical import qtpylib # from user_data.litmus import label_helpers logger = logging.getLogger(__name__) class LitmusGoodMinMaxClassificationStrategy(IStrategy): minimal_roi = {"0": 0.1, "600": -1} plot_config = { "main_plot": {}, "subplots": { "do_predict": { "do_predict": {"color": "brown"}, "DI_values": {"color": "grey"}, }, "Long": { "good_entry_long": {"color": "SpringGreen"}, "good_exit_long": {"color": "MediumVioletRed"}, }, "Short": { "good_entry_short": {"color": "SpringGreen"}, "good_exit_short": {"color": "MediumVioletRed"}, }, "True": { "true_good_entry_long": {"color": "#4dff4d"}, "true_good_exit_long": {"color": "#ff471a"}, "true_bad_entry_long": {"color": "#e6ffe6"}, "true_bad_exit_long": {"color": "#ffebe6"}, }, "Prec": { "pr_auc_is_minima": {"color": "Thistle"}, "pr_auc_is_maxima": {"color": "SteelBlue"}, "pr_auc_missed_minima": {"color": "Wheat"}, "pr_auc_missed_maxima": {"color": "Plum"} }, }, } # Stop loss config stoploss = -0.05 trailing_stop = True trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.03 trailing_only_offset_is_reached = True process_only_new_candles = True use_exit_signal = True startup_candle_count = 300 can_short = True linear_roi_offset = DecimalParameter( 0.00, 0.02, default=0.005, space="sell", optimize=False, load=True ) max_roi_time_long = IntParameter(0, 800, default=400, space="sell", optimize=False, load=True) def informative_pairs(self): whitelist_pairs = self.dp.current_whitelist() corr_pairs = self.config["freqai"]["feature_parameters"]["include_corr_pairlist"] informative_pairs = [] for tf in self.config["freqai"]["feature_parameters"]["include_timeframes"]: for pair in whitelist_pairs: informative_pairs.append((pair, tf)) for pair in corr_pairs: if pair in whitelist_pairs: continue # avoid duplication informative_pairs.append((pair, tf)) return informative_pairs def populate_any_indicators( self, pair, df, tf, informative=None, set_generalized_indicators=False ): """ Function designed to automatically generate, name and merge features from user indicated timeframes in the configuration file. User controls the indicators passed to the training/prediction by prepending indicators with `'%-' + coin ` (see convention below). I.e. user should not prepend any supporting metrics (e.g. bb_lowerband below) with % unless they explicitly want to pass that metric to the model. :param pair: pair to be used as informative :param df: strategy dataframe which will receive merges from informatives :param tf: timeframe of the dataframe which will modify the feature names :param informative: the dataframe associated with the informative pair """ coin = pair.split('/')[0] if informative is None: informative = self.dp.get_pair_dataframe(pair, tf) # first loop is automatically duplicating indicators for time periods for t in self.freqai_info["feature_parameters"]["indicator_periods_candles"]: t = int(t) informative[f"%-{coin}-rsi-period_{t}"] = ta.RSI(informative, timeperiod=t) informative[f"%-{coin}-mfi-period_{t}"] = ta.MFI(informative, timeperiod=t) informative[f"%-{coin}-adx-period_{t}"] = ta.ADX(informative, window=t) informative[f"{coin}-sma-period_{t}"] = ta.SMA(informative, timeperiod=t) informative[f"{coin}-ema-period_{t}"] = ta.EMA(informative, timeperiod=t) informative[f"%-{coin}-close_over_sma-period_{t}"] = ( informative["close"] / informative[f"{coin}-sma-period_{t}"] ) informative[f"%-{coin}-mfi-period_{t}"] = ta.MFI(informative, timeperiod=t) bollinger = qtpylib.bollinger_bands( qtpylib.typical_price(informative), window=t, stds=2.2 ) informative[f"{coin}-bb_lowerband-period_{t}"] = bollinger["lower"] informative[f"{coin}-bb_middleband-period_{t}"] = bollinger["mid"] informative[f"{coin}-bb_upperband-period_{t}"] = bollinger["upper"] informative[f"%-{coin}-bb_width-period_{t}"] = ( informative[f"{coin}-bb_upperband-period_{t}"] - informative[f"{coin}-bb_lowerband-period_{t}"] ) / informative[f"{coin}-bb_middleband-period_{t}"] informative[f"%-{coin}-close-bb_lower-period_{t}"] = ( informative["close"] / informative[f"{coin}-bb_lowerband-period_{t}"] ) informative[f"%-{coin}-roc-period_{t}"] = ta.ROC(informative, timeperiod=t) informative[f"%-{coin}-relative_volume-period_{t}"] = ( informative["volume"] / informative["volume"].rolling(t).mean() ) informative[f"%-{coin}-pct-change"] = informative["close"].pct_change() informative[f"%-{coin}-raw_volume"] = informative["volume"] informative[f"%-{coin}-raw_price"] = informative["close"] indicators = [col for col in informative if col.startswith("%")] # This loop duplicates and shifts all indicators to add a sense of recency to data for n in range(self.freqai_info["feature_parameters"]["include_shifted_candles"] + 1): if n == 0: continue informative_shift = informative[indicators].shift(n) informative_shift = informative_shift.add_suffix("_shift-" + str(n)) informative = pd.concat((informative, informative_shift), axis=1) df = merge_informative_pair(df, informative, self.config["timeframe"], tf, ffill=True) skip_columns = [ (s + "_" + tf) for s in ["date", "open", "high", "low", "close", "volume"] ] df = df.drop(columns=skip_columns) # Add generalized indicators here (because in live, it will call this # function to populate indicators during training). Notice how we ensure not to # add them multiple times if set_generalized_indicators: df["%-day_of_week"] = df["date"].dt.dayofweek df["%-hour_of_day"] = df["date"].dt.hour # Find all good entries and exits for longs min_duration = 60 min_growth = 0.02 long_labels = entry_exit_labeler( df["close"], min_growth=min_growth, max_duration=min_duration, suffix="_long" ) """short_labels = label_helpers.entry_exit_labeler( df["close"], min_growth=min_growth, max_duration=min_duration, suffix="_short" )""" df = pd.concat([df, long_labels], axis=1) return df def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.freqai_info = self.config["freqai"] dataframe = self.freqai.start(dataframe, metadata, self) # enter_mul = 2.6 # exit_mul = 1.7 trigger_window = 300 print(dataframe.columns) dataframe["test"] = dataframe["close"].rolling(trigger_window).mean() """dataframe["is_maxima_mean"] = dataframe["is_maxima"].rolling(trigger_window).mean() dataframe["is_maxima_std"] = dataframe["is_maxima"].rolling(trigger_window).std() dataframe["is_minima_mean"] = dataframe["is_minima"].rolling(trigger_window).mean() dataframe["is_minima_std"] = dataframe["is_minima"].rolling(trigger_window).std() dataframe["missed_maxima_mean"] = dataframe["missed_maxima"].rolling(trigger_window).mean() dataframe["missed_maxima_std"] = dataframe["missed_maxima"].rolling(trigger_window).std() dataframe["missed_minima_mean"] = dataframe["missed_minima"].rolling(trigger_window).mean() dataframe["missed_minima_std"] = dataframe["missed_minima"].rolling(trigger_window).std() # Short dataframe["short_entry_target"] = ( dataframe["is_maxima_mean"] + dataframe["is_maxima_std"] * enter_mul) dataframe["short_exit_target"] = ( dataframe["is_minima_mean"] + dataframe["is_minima_std"] * exit_mul) dataframe["missed_short_entry_target"] = ( dataframe["missed_maxima_mean"] + dataframe["missed_maxima_std"] * enter_mul) dataframe["missed_short_exit_target"] = ( dataframe["missed_minima_mean"] + dataframe["missed_minima_std"] * exit_mul) # Long dataframe["long_entry_target"] = ( dataframe["is_minima_mean"] + dataframe["is_minima_std"] * enter_mul) dataframe["long_exit_target"] = ( dataframe["is_maxima_mean"] + dataframe["is_maxima_std"] * exit_mul) dataframe["missed_long_entry_target"] = ( dataframe["missed_minima_mean"] + dataframe["missed_minima_std"] * enter_mul) dataframe["missed_long_exit_target"] = ( dataframe["missed_maxima_mean"] + dataframe["missed_maxima_std"] * exit_mul)""" return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: # Long Entry conditions = [df["do_predict"] == 1, df["test"] > 0] if conditions: df.loc[ reduce(lambda x, y: x & y, conditions), ["enter_long", "enter_tag"] ] = (1, "is_minima") # Short Entry conditions = [df["do_predict"] == 1, df["test"] > 0] if conditions: df.loc[ reduce(lambda x, y: x & y, conditions), ["enter_short", "enter_tag"] ] = (1, "is_maxima") return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: # Long Exit conditions = [1 == 1, df["test"] > 0] if conditions: df.loc[ reduce(lambda x, y: x & y, conditions), ["exit_long", "exit_tag"] ] = (1, "is_maxima") # Short Exit conditions = [1 == 1, df["test"] > 0] if conditions: df.loc[ reduce(lambda x, y: x & y, conditions), ["exit_short", "exit_tag"] ] = (1, "is_minima") return df def get_ticker_indicator(self): return int(self.config["timeframe"][:-1]) """def custom_exit( self, pair: str, trade: Trade, current_time, current_rate, current_profit, **kwargs ): dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) trade_date = timeframe_to_prev_date(self.config["timeframe"], trade.open_date_utc) trade_candle = dataframe.loc[(dataframe["date"] == trade_date)] if trade_candle.empty: return None trade_candle = trade_candle.squeeze() follow_mode = self.config.get("freqai", {}).get("follow_mode", False) if not follow_mode: pair_dict = self.freqai.dd.pair_dict else: pair_dict = self.freqai.dd.follower_dict entry_tag = trade.enter_tag if ( "prediction" + entry_tag not in pair_dict[pair] or pair_dict[pair]["prediction" + entry_tag] > 0 ): with self.freqai.lock: pair_dict[pair]["prediction" + entry_tag] = abs(trade_candle["&-s_close"]) if not follow_mode: self.freqai.dd.save_drawer_to_disk() else: self.freqai.dd.save_follower_dict_to_disk() roi_price = pair_dict[pair]["prediction" + entry_tag] roi_time = self.max_roi_time_long.value roi_decay = roi_price * ( 1 - ((current_time - trade.open_date_utc).seconds) / (roi_time * 60) ) if roi_decay < 0: roi_decay = self.linear_roi_offset.value else: roi_decay += self.linear_roi_offset.value if current_profit > roi_decay: return "roi_custom_win" if current_profit < -roi_decay: return "roi_custom_loss" """ """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, **kwargs, ) -> bool: entry_tag = trade.enter_tag follow_mode = self.config.get("freqai", {}).get("follow_mode", False) if not follow_mode: pair_dict = self.freqai.dd.pair_dict else: pair_dict = self.freqai.dd.follower_dict with self.freqai.lock: pair_dict[pair]["prediction" + entry_tag] = 0 if not follow_mode: self.freqai.dd.save_drawer_to_disk() else: self.freqai.dd.save_follower_dict_to_disk() return True""" def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs, ) -> bool: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = df.iloc[-1].squeeze() if side == "long": if rate > (last_candle["close"] * (1 + 0.0025)): return False else: if rate < (last_candle["close"] * (1 - 0.0025)): return False return True def minmax_growth_np(ts): """Just for testing without numba... very slow for large ts""" ts = ts.astype(np.float32) res = np.empty((ts.shape[0], ts.shape[0]), dtype=ts.dtype) for i in range(ts.shape[0]): for j in range(ts.shape[0]): r = (ts[j] - ts[i]) / ts[i] res[i, j] = r return res def entry_exit_labeler(close_df, min_growth, max_duration, suffix=None): """Label timeseries for good entries and exits at peaks""" logger.info(f"Labeling for {len(close_df)} samples") logger.info(f"Labeling settings: min_growth {min_growth}, max_duration {max_duration}") start_time = time.time() close = close_df.values # Find all peaks and valleys exit_idx = find_peaks(close)[0] entry_idx = find_peaks(-close)[0] # Mask valleys & peaks min_mask = np.zeros(close.shape, dtype=bool) min_mask[entry_idx] = True max_mask = np.zeros(close.shape, dtype=bool) max_mask[exit_idx] = True # Compute distance matrix between all close points dist_matrix = minmax_growth_np(close) # Scope dist_matrix to distances > min_threshold growth_mask = (dist_matrix - min_growth) > 0 dist_matrix = dist_matrix * growth_mask # Limit to max_duration & remove lower triangle dist_matrix = np.tril(dist_matrix, k=max_duration) dist_matrix = np.triu(dist_matrix) # Scope to only entries & get idx dist_matrix = dist_matrix * min_mask.reshape(-1, 1) good_entry_idx, _ = np.where(dist_matrix > 0) # Scope to entries & exits & find idx dist_matrix = dist_matrix * max_mask _, good_exit_idx = np.where(dist_matrix > 0) # Populate DataFrame and return col_names = ["&target", "true_good_entry", "true_good_exit", "true_bad_entry", "true_bad_exit"] col_names = [i + suffix for i in col_names] label_names = ["good_entry", "good_exit", "bad_entry", "bad_exit", "transition"] label_names = [i + suffix for i in label_names] df = pd.DataFrame(columns=col_names, index=close_df.index) df[col_names[1:]] = 0 # Bad entries df.loc[[i for i in entry_idx if i not in good_entry_idx], col_names[0]] = label_names[2] df.loc[[i for i in entry_idx if i not in good_entry_idx], col_names[3]] = 1 # Good entries df.loc[good_entry_idx, col_names[0]] = label_names[0] df.loc[good_entry_idx, col_names[1]] = 1 # Bad exit df.loc[[i for i in exit_idx if i not in good_exit_idx], col_names[0]] = label_names[3] df.loc[[i for i in exit_idx if i not in good_exit_idx], col_names[4]] = 1 # Good exit df.loc[good_exit_idx, col_names[0]] = label_names[1] df.loc[good_exit_idx, col_names[2]] = 1 # Fill the rest with filler_value df[col_names[0]].fillna(value=label_names[4], inplace=True) end_time = time.time() - start_time logger.info(f"Time taken to label data: {end_time} seconds") print(df) return df