# source: https://raw.githubusercontent.com/nateemma/strategies/4527005ccba228325ad6a6ec33e64fea8be1663a/binanceus/DWT_Predict.py from datetime import datetime from functools import reduce import numpy as np # Get rid of pandas warnings during backtesting import pandas as pd from pandas import DataFrame, Series from sklearn.ensemble import GradientBoostingRegressor from sklearn.preprocessing import RobustScaler import freqtrade.vendor.qtpylib.indicators as qtpylib from freqtrade.strategy import (IStrategy, DecimalParameter) pd.options.mode.chained_assignment = None # default='warn' # Strategy specific imports, files must reside in same folder as strategy import sys from pathlib import Path sys.path.append(str(Path(__file__).parent)) import logging import warnings log = logging.getLogger(__name__) # log.setLevel(logging.DEBUG) warnings.simplefilter(action='ignore', category=pd.errors.PerformanceWarning) from DataframeUtils import DataframeUtils, ScalerType import pywt import talib.abstract as ta """ #################################################################################### github_nateemma_strategies__DWT_Predict__20230624_173842 - use a Discreet Wavelet Transform to model the price, and an sklearn regression algorithm trained on the DWT coefficients, which is then used to predict future prices. Unfortunately, this must all be done in a rolling fashion to avoid lookahead bias - so it is pretty slow #################################################################################### """ class github_nateemma_strategies__DWT_Predict__20230624_173842(IStrategy): # Do *not* hyperopt for the roi and stoploss spaces # ROI table: minimal_roi = { "0": 0.06 } # Stoploss: stoploss = -0.10 # Trailing stop: trailing_stop = False trailing_stop_positive = None trailing_stop_positive_offset = 0.0 trailing_only_offset_is_reached = False timeframe = '5m' inf_timeframe = '15m' use_custom_stoploss = False # Recommended use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True # Required startup_candle_count: int = 128 # must be power of 2 process_only_new_candles = False custom_trade_info = {} ################################### # Strategy Specific Variable Storage ## Hyperopt Variables dwt_window = startup_candle_count lookahead = 6 df_coeffs: DataFrame = None coeff_model = None dataframeUtils = None scaler = RobustScaler() # DWT hyperparams entry_dwt_diff = DecimalParameter(0.0, 5.0, decimals=1, default=0.4, space='buy', load=True, optimize=True) exit_dwt_diff = DecimalParameter(-5.0, 0.0, decimals=1, default=-0.3, space='sell', load=True, optimize=True) plot_config = { 'main_plot': { 'close': {'color': 'cornflowerblue'}, # 'dwt_model': {'color': 'lightsalmon'}, 'dwt_predict': {'color': 'mediumaquamarine'}, }, 'subplots': { "Diff": { 'model_diff': {'color': 'brown'}, }, } } ################################### """ Informative Pair Definitions """ def informative_pairs(self): return [] ################################### """ Indicator Definitions """ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Base pair dataframe timeframe indicators curr_pair = metadata['pair'] #print("") #print(curr_pair) #print("") if self.dataframeUtils is None: self.dataframeUtils = DataframeUtils() self.dataframeUtils.set_scaler_type(ScalerType.Robust) # # build the DWT # #print(" Building DWT...") # dataframe['dwt_model'] = dataframe['close'].rolling(window=self.dwt_window).apply(self.model) # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.dwt_window) # Williams %R dataframe['wr'] = 0.02 * (self.williams_r(dataframe, period=14) + 50.0) # Fisher RSI rsi = 0.1 * (dataframe['rsi'] - 50) dataframe['fisher_rsi'] = (np.exp(2 * rsi) - 1) / (np.exp(2 * rsi) + 1) # Combined Fisher RSI and Williams %R dataframe['fisher_wr'] = (dataframe['wr'] + dataframe['fisher_rsi']) / 2.0 # build the list of model coefficients - added to self.df_coeffs #print(" Building coefficients dataframe...") self.df_coeffs = None # reset for each pair coeffs = dataframe['close'].rolling(window=self.dwt_window).apply(self.add_coeffs) #print(" Merging coefficients into dataframe...") dataframe = self.merge_data(dataframe) #print(" Training Model...") dataframe['dwt_predict'] = dataframe['close'] self.train_model(dataframe) # add the predictions #print(" Making predictions...") if self.dp.runmode.value not in ('hyperopt', 'backtest', 'plot'): dataframe = self.add_predictions(dataframe) else: dataframe = self.add_rolling_predictions(dataframe) dataframe['model_diff'] = 100.0 * (dataframe['dwt_predict'] - dataframe['close']) / dataframe[ 'close'] return dataframe ################################### # Williams %R def williams_r(self, dataframe: DataFrame, period: int = 14) -> Series: """Williams %R, or just %R, is a technical analysis oscillator showing the current closing price in relation to the high and low of the past N days (for a given N). It was developed by a publisher and promoter of trading materials, Larry Williams. Its purpose is to tell whether a stock or commodity market is trading near the high or the low, or somewhere in between, of its recent trading range. The oscillator is on a negative scale, from −100 (lowest) up to 0 (highest). """ highest_high = dataframe["high"].rolling(center=False, window=period).max() lowest_low = dataframe["low"].rolling(center=False, window=period).min() WR = Series( (highest_high - dataframe["close"]) / (highest_high - lowest_low), name=f"{period} Williams %R", ) return WR * -100 def madev(self, d, axis=None): """ Mean absolute deviation of a signal """ return np.mean(np.absolute(d - np.mean(d, axis)), axis) def dwtModel(self, data): # the choice of wavelet makes a big difference # for an overview, check out: https://www.kaggle.com/theoviel/denoising-with-direct-wavelet-transform # wavelet = 'db1' # wavelet = 'bior1.1' wavelet = 'haar' # deals well with harsh transitions level = 2 wmode = "smooth" length = len(data) coeff = pywt.wavedec(data, wavelet, mode=wmode) # remove higher harmonics sigma = (1 / 0.6745) * self.madev(coeff[-level]) uthresh = sigma * np.sqrt(2 * np.log(length)) coeff[1:] = (pywt.threshold(i, value=uthresh, mode='hard') for i in coeff[1:]) # inverse transform model = pywt.waverec(coeff, wavelet, mode=wmode) return model def model(self, a: np.ndarray) -> np.float: # must return scalar, so just calculate prediction and take last value # model = self.dwtModel(np.array(a)) # de-trend the data w_mean = a.mean() w_std = a.std() x_notrend = (a - w_mean) / w_std # get DWT model of data restored_sig = self.dwtModel(x_notrend) # re-trend model = (restored_sig * w_std) + w_mean length = len(model) return model[length - 1] # adds coefficients to dataframe row def add_coeffs(self, a: np.ndarray) -> np.float: # get the DWT coefficients wavelet = 'haar' level = 2 length = len(a) coeffs = pywt.wavedec(a, wavelet, level=level, mode='smooth') # remove higher harmonics sigma = (1 / 0.6745) * self.madev(coeffs[-level]) uthresh = sigma * np.sqrt(2 * np.log(length)) coeffs[1:] = (pywt.threshold(i, value=uthresh, mode='hard') for i in coeffs[1:]) features = np.concatenate(np.array(coeffs, dtype=object)) # #print("") # #print(f"features: {np.shape(features)}") # #print(features) # #print("") if self.df_coeffs is None: # cols = np.arange(len(features)) cols = [] for i in range(len(features)): col = "coeff_" + str(i) cols.append(col) self.df_coeffs = pd.DataFrame(columns=cols) # add rows of zeros to account fpr the rolling window startup zeros = [] for i in range(self.dwt_window-1): zeros.append([0] * len(self.df_coeffs.columns)) # Add the rows of zeros to the dataframe # self.df_coeffs = self.df_coeffs.append(pd.DataFrame(zeros, columns=self.df_coeffs.columns), # ignore_index=True) self.df_coeffs = pd.concat([self.df_coeffs, pd.DataFrame(zeros, columns=self.df_coeffs.columns)]) self.df_coeffs.loc[len(self.df_coeffs)] = features return 1.0 def merge_data(self, dataframe: DataFrame) -> DataFrame: # merge df_coeffs into the main dataframe l1 = dataframe.shape[0] l2 = self.df_coeffs.shape[0] if l1 != l2: #print(f" **** size mismatch. len(dataframe)={l1} len(self.df_coeffs)={l2}") dataframe = pd.concat([dataframe, self.df_coeffs], axis=1, ignore_index=False) return dataframe def convert_dataframe(self, dataframe: DataFrame) -> DataFrame: df = dataframe.copy() # convert date column so that it can be scaled. if 'date' in df.columns: dates = pd.to_datetime(df['date'], utc=True) df['date'] = dates.astype('int64') df.fillna(0.0, inplace=True) df.set_index('date') df.reindex() # scale the dataframe self.scaler.fit(df) df = pd.DataFrame(self.scaler.transform(df), columns=df.columns) return df def train_model(self, dataframe: DataFrame): df = self.convert_dataframe(dataframe) # need to exclude the startup period at the front, and the lookahead period at the end df = df.iloc[self.startup_candle_count:-self.lookahead] y = dataframe['close'].iloc[self.startup_candle_count+self.lookahead:].to_numpy() # #print(f"df: {df.shape} y:{y.shape}") # self.coeff_model = SVR(kernel='rbf', C=1.0, epsilon=0.1) params = {'n_estimators': 100, 'max_depth': 4, 'min_samples_split': 2, 'learning_rate': 0.1, 'loss': 'squared_error'} self.coeff_model = GradientBoostingRegressor(**params) self.coeff_model.fit(df, y) def predict(self, a: np.ndarray) -> np.float: y_pred = self.coeff_model.predict(a) return y_pred def add_predictions(self, dataframe: DataFrame) -> DataFrame: df = self.convert_dataframe(dataframe) dataframe['dwt_predict'] = self.coeff_model.predict(df) return dataframe def add_rolling_predictions(self, dataframe: DataFrame) -> DataFrame: df = self.convert_dataframe(dataframe) nrows = df.shape[0] start = 0 dataframe['dwt_predict'] = dataframe['close'] for i in range(nrows): end = start + self.dwt_window - 1 slice = df.iloc[start:end] dataframe['dwt_predict'][start:end] = self.coeff_model.predict(slice) start = start + 1 return dataframe ################################### """ entry Signal """ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] dataframe.loc[:, 'enter_tag'] = '' # some trading volume conditions.append(dataframe['volume'] > 0) # Fisher/Williams in buy region conditions.append(dataframe['fisher_wr'] <= -0.5) # DWT triggers dwt_cond = ( qtpylib.crossed_above(dataframe['model_diff'], self.entry_dwt_diff.value) ) conditions.append(dwt_cond) # DWTs will spike on big gains, so try to constrain spike_cond = ( dataframe['model_diff'] < 2.0 * self.entry_dwt_diff.value ) conditions.append(spike_cond) # set entry tags dataframe.loc[dwt_cond, 'enter_tag'] += 'dwt_entry ' if conditions: dataframe.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1 return dataframe ################################### """ exit Signal """ def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] dataframe.loc[:, 'exit_tag'] = '' # some volume conditions.append(dataframe['volume'] > 0) # Fisher/Williams in sell region conditions.append(dataframe['fisher_wr'] >= 0.5) # DWT triggers dwt_cond = ( qtpylib.crossed_below(dataframe['model_diff'], self.exit_dwt_diff.value) ) conditions.append(dwt_cond) # DWTs will spike on big gains, so try to constrain spike_cond = ( dataframe['model_diff'] > 2.0 * self.exit_dwt_diff.value ) conditions.append(spike_cond) # set exit tags dataframe.loc[dwt_cond, 'exit_tag'] += 'dwt_exit ' if conditions: dataframe.loc[reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1 return dataframe ###################################