# --- Do not remove these libs --- from freqtrade.strategy.interface import IStrategy from typing import Dict, List from functools import reduce from pandas import DataFrame # -------------------------------- import talib.abstract as ta import numpy as np import freqtrade.vendor.qtpylib.indicators as qtpylib import datetime from technical.util import resample_to_interval, resampled_merge from datetime import datetime, timedelta from freqtrade.persistence import Trade from freqtrade.strategy import stoploss_from_open, merge_informative_pair, DecimalParameter, IntParameter, CategoricalParameter import technical.indicators as ftt def EWO(dataframe, ema_length=5, ema2_length=35): df = dataframe.copy() ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) emadif = (ema1 - ema2) / df['close'] * 100 return emadif class strat_suck_1a_stash86_20210728(IStrategy): INTERFACE_VERSION = 2 # Buy hyperspace params: buy_params = { "base_nb_candles_buy": 16, "ewo_high": 5.638, "ewo_low": -19.993, "low_offset": 0.978, "rsi_buy": 61, } # Sell hyperspace params: sell_params = { "base_nb_candles_sell": 49, "high_offset": 1.006, } # ROI table: minimal_roi = { "0": 10 } # Stoploss: stoploss = -0.1 # SMAOffset sma_offset_optimize=True base_nb_candles_buy = IntParameter( 5, 80, default=buy_params['base_nb_candles_buy'], space='buy', optimize=sma_offset_optimize) low_offset = DecimalParameter( 0.9, 0.99, default=buy_params['low_offset'], space='buy', optimize=sma_offset_optimize) high_offset = DecimalParameter( 0.99, 1.1, default=sell_params['high_offset'], space='sell', optimize=True) # Protection fast_ewo = 50 slow_ewo = 200 protection_optimize=False ewo_high = DecimalParameter( 2.0, 12.0, default=buy_params['ewo_high'], space='buy', optimize=protection_optimize) rsi_buy = IntParameter(30, 70, default=buy_params['rsi_buy'], space='buy', optimize=protection_optimize) rsi_n_buy = IntParameter(5, 50, default=5) # Trailing stop: trailing_stop = True trailing_stop_positive = 0.001 trailing_stop_positive_offset = 0.015 trailing_only_offset_is_reached = True # Sell signal use_sell_signal = True sell_profit_only = True sell_profit_offset = 0.01 ignore_roi_if_buy_signal = False # Optimal timeframe for the strategy timeframe = '5m' informative_timeframe = '1h' process_only_new_candles = True startup_candle_count: int = 300 plot_config = { 'main_plot': { 'ma_buy': {'color': 'orange'}, 'bb_sell': {'color': 'orange'}, }, } use_custom_stoploss = True def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: if current_profit >= 0.027: return stoploss_from_open(current_profit - 0.025) return -0.99 def custom_sell(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, **kwargs): if current_time - timedelta(minutes=120) > trade.open_date_utc: if current_profit >= 0.005: return 'i cant wait' def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [(pair, self.informative_timeframe) for pair in pairs] return informative_pairs def get_informative_indicators(self, metadata: dict): dataframe = self.dp.get_pair_dataframe( pair=metadata['pair'], timeframe=self.informative_timeframe) return dataframe def top_percent_change(self, dataframe: DataFrame, length: int) -> float: """ Percentage change of the current close from the range maximum Open price :param dataframe: DataFrame The original OHLC dataframe :param length: int The length to look back """ df = dataframe.copy() if length == 0: return ((df['open'] - df['close']) / df['close']) else: return ((df['open'].rolling(length).max() - df['close']) / df['close']) def dip_from_bb_mid_to_bottom(self, dataframe: DataFrame, length: int) -> int: df = dataframe.copy() if length == 0: return (df['high'] >= df['bb_middleband']) & (df['low'] <= df['bb_lowerband']) else: return (df['high'].rolling(length).max() >= df['bb_middleband'].rolling(length).max()) & (df['low'] <= df['bb_lowerband']) def informative_1h_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, "DataProvider is required for multiple timeframes." # Get the informative pair informative_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.informative_timeframe) # EMA informative_1h['ema_200'] = ta.EMA(informative_1h, timeperiod=200) return informative_1h def normal_tf_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # MACD macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # The indicators for the 1h informative timeframe informative_1h = self.informative_1h_indicators(dataframe, metadata) dataframe = merge_informative_pair(dataframe, informative_1h, self.timeframe, self.informative_timeframe, ffill=True) # The indicators for the normal (5m) timeframe dataframe = self.normal_tf_indicators(dataframe, metadata) return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions =[] for val in self.rsi_n_buy.range: conditions.append( ( (dataframe['close'] < (dataframe['close'].tail(300).min() + ((dataframe['close'].tail(300).max() - dataframe['close'].tail(300).min()) / 3)) ) & (dataframe['close'].tail(10).min() >= dataframe['close'] ) & (dataframe['rsi'].shift(val) > dataframe['rsi']) & ((dataframe['rsi'].shift(val) - dataframe['rsi']) >= 15) & (((dataframe['macd'].shift(val) + dataframe['macdsignal'].shift(val)) / 2) > ((dataframe['macd'] + dataframe['macdsignal']) / 2)) & (dataframe['low'] > dataframe['low'].shift(val)) & (((dataframe['low'].shift(val) / 100) * 102) < dataframe['low']) & (dataframe['volume'] > 0) ) ) conditions.append( ( (dataframe['close'] < (dataframe['close'].tail(300).min() + ((dataframe['close'].tail(300).max() - dataframe['close'].tail(300).min()) / 3)) ) & (dataframe['close'].tail(10).min() >= dataframe['close'] ) & (dataframe['rsi'] > dataframe['rsi'].shift(val)) & ((dataframe['rsi'] - dataframe['rsi'].shift(val)) >= 15) & (((dataframe['macd'] + dataframe['macdsignal']) / 2) > ((dataframe['macd'].shift(val) + dataframe['macdsignal'].shift(val)) / 2)) & (dataframe['low'].shift(val) > dataframe['low']) & (((dataframe['low'] / 100) * 102) < dataframe['low'].shift(val)) & (dataframe['volume'] > 0) ) ) if conditions: dataframe.loc[ reduce(lambda x, y: x | y, conditions), 'buy' ] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['volume'] > 0) ), 'sell'] = 0 return dataframe