import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame import os from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, informative) from freqtrade.data.dataprovider import DataProvider from statsmodels.tsa.statespace.sarimax import SARIMAX, SARIMAXResults from pmdarima.arima import auto_arima from datetime import datetime pd.set_option('display.max_columns', None) class AutoArimaTripleV1(IStrategy): """ This is a strategy template to get you started. More information in https://www.freqtrade.io/en/latest/strategy-customization/ You can: :return: a Dataframe with all mandatory indicators for the strategies - Rename the class name (Do not forget to update class_name) - Add any methods you want to build your strategy - Add any lib you need to build your strategy You must keep: - the lib in the section "Do not remove these libs" - the methods: populate_indicators, populate_buy_trend, populate_sell_trend You should keep: - timeframe, minimal_roi, stoploss, trailing_* """ INTERFACE_VERSION = 2 log = open(os.path.join(os.path.dirname(os.path.realpath(__file__)), '../logs/strategy-log.log'), 'w') minimal_roi = { "60": 0.01, "30": 0.02, "0": 0.04 } stoploss = -0.10 trailing_stop = False timeframe = '5m' process_only_new_candles = True use_sell_signal = True sell_profit_only = False ignore_roi_if_buy_signal = False startup_candle_count: int = 30 order_types = { 'buy': 'market', 'sell': 'market', 'stoploss': 'market', 'stoploss_on_exchange': False } order_time_in_force = { 'buy': 'gtc', 'sell': 'gtc' } plot_config = { 'main_plot': { 'tema': {}, 'sar': {'color': 'white'}, }, 'subplots': { "MACD": { 'macd': {'color': 'blue'}, 'macdsignal': {'color': 'orange'}, }, "RSI": { 'rsi': {'color': 'red'}, } } } predictor_30m = None predictor_1h = None predictor_5m = None last_predicted_frame_30m = None, None last_predicted_frame_1h = None, None last_predicted_frame_5m = None, None @informative('30m') def populate_indicators_30m(self, dataframe: DataFrame, metadata: dict) -> DataFrame: current_date = dataframe.loc[len(dataframe) - 1].date last_date, last_prediction = self.last_predicted_frame_30m if last_date == current_date: self.__print('Use stored predicted value as 30m indicator.') dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1) return dataframe if not self.predictor_30m: self.predictor_30m = ArimaPredictor(self.dp, metadata['pair'], '30m', '30min') self.__print('===30m===') self.__print(dataframe.tail(2)) prediction = self.predictor_30m.predict(dataframe, current_date) dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1) self.last_predicted_frame_30m = current_date, prediction self.__print(f'Prediction 30m {current_date}: {prediction:0.2f}') return dataframe def __print(self, message): self.log.write(f'{message}\n') @informative('1h') def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: current_date = dataframe.loc[len(dataframe) - 1].date last_date, last_prediction = self.last_predicted_frame_1h if last_date == current_date: self.__print('Use stored predicted value as 1h indicator.') dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1) return dataframe if not self.predictor_1h: self.predictor_1h = ArimaPredictor(self.dp, metadata['pair'], '1h', 'H') self.__print('===1h===') self.__print(dataframe.tail(2)) prediction = self.predictor_1h.predict(dataframe, current_date) dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1) self.last_predicted_frame_1h = current_date, prediction self.__print(f'Prediction 1h {current_date}: {prediction:0.2f}') return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame Performance Note: For the best performance be frugal on the number of indicators you are using. Let uncomment only the indicator you are using in your strategies or your hyperopt configuration, otherwise you will waste your memory and CPU usage. :param dataframe: Dataframe with data from the exchange :param metadata: Additional information, like the currently traded pair :return: a Dataframe with all mandatory indicators for the strategies """ current_date = dataframe.loc[len(dataframe) - 1].date last_date, last_prediction = self.last_predicted_frame_5m if last_date == current_date: self.__print('Use stored predicted value as 5m indicator.') dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1) return dataframe if not self.predictor_5m: self.predictor_5m = ArimaPredictor(self.dp, metadata['pair'], '5m', '5min') self.__print('===5m===') self.__print(dataframe.tail(2)) prediction = self.predictor_5m.predict(dataframe.loc[:, ['date', 'open', 'high', 'low', 'close', 'volume']], current_date) dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1) self.last_predicted_frame_5m = current_date, prediction self.__print(f'Prediction 5m {current_date}: {prediction:0.2f}') return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the buy signal for the given dataframe :param dataframe: DataFrame populated with indicators :param metadata: Additional information, like the currently traded pair :return: DataFrame with buy column """ self.__print('===buy===') self.__print(dataframe.tail(1)) dataframe.loc[ ( ), 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the sell signal for the given dataframe :param dataframe: DataFrame populated with indicators :param metadata: Additional information, like the currently traded pair :return: DataFrame with buy column """ self.__print('===sell===') self.__print(dataframe.tail(1)) dataframe.loc[ ( ), 'sell'] = 1 return dataframe class ArimaPredictor(object): train_size = 10000 steps = -1 scale = None model: SARIMAXResults = None last_x = None def __init__(self, dp: DataProvider, pair: str, t_frame: str, frequency: str): self.dp = dp self.pair = pair self.t_frame = t_frame self.frequency = frequency self.data = self.__download_data() def predict(self, c_data: DataFrame, current_prediction_date): self.__combine_data(c_data) must_train = self.__must_train() if must_train: self.__train(current_prediction_date) x, y = self.__get_current_data(current_prediction_date) if not must_train: self.model.append(y, exog=self.last_x, refit=False) prediction = self.model.forecast(steps=self.steps*-1, exog=x) self.last_x = x prediction = prediction.tail(1).mean() current = x['mean'].mean() return (prediction-current)/current*100 def __get_current_data(self, current_prediction_date): index = self.data.index.get_loc(current_prediction_date) row = self.data.iloc[index:index+1, :].copy().asfreq(freq=self.frequency) row['mean'] = (row['low'] + row['high']) / 2 self.scale(row) return row, row.rename(columns={'mean': 'step_ahead'})\ .drop('open', axis=1)\ .drop('high', axis=1)\ .drop('low', axis=1)\ .drop('volume', axis=1)\ .drop('close', axis=1) def __must_train(self): return self.model is None def __train(self, current_prediction_date): start = datetime.now() print(f'Started {self.t_frame} training at {start}.') end_index = self.data.index.get_loc(current_prediction_date) x, y = self.__prepare_data(self.data, self.frequency) train_x, train_y, self.scale = self.__get_train_data(x, y, end_index) step_wise = auto_arima(train_y, exogenous=train_x, start_p=1, start_q=1, max_p=7, max_q=7, d=1, max_d=7, trace=False, error_action='ignore', suppress_warnings=True, stepwise=True) print(f'Optimal parameter for {self.t_frame} are {step_wise.order}.') self.model = SARIMAX(train_y, exog=train_x, order=step_wise.order, enforce_invertibility=False, enforce_stationarity=False).fit() print(f'Finished training of {self.t_frame} in {datetime.now()-start}.') def __download_data(self): h_data = self.dp.historic_ohlcv(self.pair, self.t_frame).set_index('date') return h_data def __combine_data(self, c_data: DataFrame): c_data = c_data.copy().set_index('date') self.data = self.data.append(c_data[c_data.index.isin(self.data.index) == False], verify_integrity=True) def __prepare_data(self, df: DataFrame, frequency: str): df = df.copy() df['mean'] = (df['low'] + df['high']) / 2 df['step_ahead'] = df['mean'].shift(self.steps) df = df.dropna() x = df.copy().drop('step_ahead', axis=1).asfreq(freq=frequency).interpolate() y = df.copy().drop('open', axis=1).drop('high', axis=1).drop('low', axis=1).drop('volume', axis=1).drop('mean', axis=1).drop( 'close', axis=1).asfreq(freq=frequency).interpolate() return x, y def __get_train_data(self, x, y, current_prediction_index): if current_prediction_index - self.train_size < 0: raise Exception('Not enough data in history. Please first download enough data') train_x = x.iloc[current_prediction_index - self.train_size:current_prediction_index, :].copy() train_y = y.iloc[current_prediction_index - self.train_size:current_prediction_index, :].copy() fix_scaler = FixedScaler(train_x['mean'].min(), train_x['mean'].max(), train_x['volume'].min(), train_x['volume'].max()) def scale(df): nonlocal fix_scaler df['volume'] = df['volume'].apply(lambda v: fix_scaler.scale(v)) scale(train_x) return train_x, train_y, scale class FixedScaler(object): def __init__(self, min_in, max_in, min_out, max_out): self.min_in = min_in self.max_in = max_in self.min_out = min_out self.max_out = max_out def scale(self, value): value_std = (value - self.min_out) / (self.max_out - self.min_out) return value_std * (self.max_in - self.min_in) + self.min_in