# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # isort: skip_file # --- Do not remove these libs --- import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame from freqtrade.strategy import IStrategy, merge_informative_pair # -------------------------------- # Add your lib to import here import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from technical.util import resample_to_interval, resampled_merge from freqtrade.exchange import timeframe_to_minutes from functools import reduce from datetime import datetime, timedelta from freqtrade.persistence import Trade def calc_streaks(series: pd.Series): # logic tables geq = series >= series.shift(1) # True if rising eq = series == series.shift(1) # True if equal logic_table = pd.concat([geq, eq], axis=1) streaks = [0] # holds the streak duration, starts with 0 for row in logic_table.iloc[1:].itertuples(): # iterate through logic table if row[2]: # same value as before streaks.append(0) continue last_value = streaks[-1] if row[1]: # higher value than before streaks.append(last_value + 1 if last_value >= 0 else 1) # increase or reset to +1 else: # lower value than before streaks.append(last_value - 1 if last_value < 0 else -1) # decrease or reset to -1 return streaks bb_enter = 0.08 pov_enter = -20 # This class is a sample. Feel free to customize it. def easeInCubic(t): return t * t * t def clamp(num, min_value, max_value): return max(min(num, max_value), min_value) def clamp01(num): return clamp(num, 0, 1) class crsiv9(IStrategy): timeframe = "5m" """ This is a sample strategy to inspire you. 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 prototype for the methods: minimal_roi, stoploss, populate_indicators, populate_buy_trend, populate_sell_trend, hyperopt_space, buy_strategy_generator """ # Strategy interface version - allow new iterations of the strategy interface. # Check the documentation or the Sample strategy to get the latest version. INTERFACE_VERSION = 2 # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 100 # ROI table: # ROI table: minimal_roi = { "0": 0.15, "60": 0.05, "90": 0.025, "150": 0.016, } # Stoploss: stoploss = -0.30 use_custom_stoploss = True custom_stop_ramp_minutes = 110 trailing_stop = True trailing_stop_positive = 0.002 trailing_stop_positive_offset = 0.04 trailing_only_offset_is_reached = False order_time_in_force = { 'buy': 'gtc', 'sell': 'gtc' } plot_config = { # Main plot indicators (Moving averages, ...) 'main_plot': { 'sma10': {}, }, 'subplots': { # Subplots - each dict defines one additional plot "stoch": { 'slowd': {'color': 'blue'}, 'slowk': {'color': 'orange'}, }, "stochrsi": { 'fastd_rsi': {'color': 'red'}, 'fastk_rsi': {'color': 'black'}, }, "bb_width": { 'bb_width': {}, }, "up": { 'bull': {}, }, "adx": { 'adx': {}, 'crsi': {}, } } } def get_ticker_indicator(self): return int(self.timeframe[:-1]) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi'] = ta.RSI(dataframe, 5) dataframe['rsi2'] = ta.RSI(dataframe) dataframe['adx'] = ta.ADX(dataframe, timeperiod=10) dataframe['streak'] = calc_streaks(dataframe["close"]) dataframe['srsi'] = ta.RSI(dataframe['streak'], 2) dataframe['roc'] = ta.ROC(dataframe, 5) dataframe['sma10'] = ta.SMA(dataframe, timeperiod=10) dataframe['crsi'] = (dataframe['rsi'] + dataframe['srsi'] + dataframe['roc']) / 3 bollinger = qtpylib.bollinger_bands( qtpylib.typical_price(dataframe), window=20, stds=2) dataframe['bb_lowerband'] = bollinger['lower'] dataframe['bb_middleband'] = bollinger['mid'] dataframe['bb_upperband'] = bollinger['upper'] dataframe['cci'] = ta.CCI(dataframe, 7) # dataframe['ema155'] = ta.SMA(dataframe, timeperiod=50) dataframe['bb_width'] = ( (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / dataframe['bb_middleband']) resample_rsi_interval = timeframe_to_minutes(self.timeframe) * 6 resample_rsi_key = 'resample_{}_rsi'.format(resample_rsi_interval) dataframe_long = resample_to_interval(dataframe, resample_rsi_interval) dataframe_long['rsi'] = ta.RSI(dataframe_long, timeperiod=14) dataframe = resampled_merge(dataframe, dataframe_long) dataframe[resample_rsi_key].fillna(method='ffill', inplace=True) # bull used to select between two different sets of buy/sell threshold values # based on long RSI dataframe['bull'] = dataframe[resample_rsi_key] dataframe['sar'] = ta.SAR(dataframe) def SSLChannels(dataframe, length=7, mode='sma'): df = dataframe.copy() df['ATR'] = ta.ATR(df, timeperiod=14) df['smaHigh'] = df['high'].rolling(length).mean() + df['ATR'] df['smaLow'] = df['low'].rolling(length).mean() - df['ATR'] df['hlv'] = np.where(df['close'] > df['smaHigh'], 1, np.where( df['close'] < df['smaLow'], -1, np.NAN)) df['hlv'] = df['hlv'].ffill() df['sslDown'] = np.where( df['hlv'] < 0, df['smaHigh'], df['smaLow']) df['sslUp'] = np.where(df['hlv'] < 0, df['smaLow'], df['smaHigh']) return df['sslDown'], df['sslUp'] ssl = SSLChannels(dataframe, 10) dataframe['sslDown'] = ssl[0] dataframe['sslUp'] = ssl[1] dataframe['sslTrend'] = dataframe['sslUp'].gt(dataframe['sslDown']) # first check if dataprovider is available if self.dp: if self.dp.runmode in ('live', 'dry_run'): ob = self.dp.orderbook(metadata['pair'], 1) dataframe['best_bid'] = ob['bids'][0][0] dataframe['best_ask'] = ob['asks'][0][0] return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['close'] < dataframe['sar']) & (dataframe['streak'] < -1) & (dataframe['bull'] < 55) & (dataframe['sslTrend'] == False) & (dataframe['crsi'] < 11) & (dataframe['cci'] <= -100) & (dataframe['adx'] > 22) & (dataframe['bb_width'] > 0.035) ) | ( # (dataframe['close'] < dataframe['sar']) & # (dataframe['streak'] < -1) & (dataframe['bull'] > 65) & # (dataframe['sslTrend'] == True)& (dataframe['crsi'] < 25) & # (dataframe['cci'] <= 0) & (dataframe['adx'] > 25) & (dataframe['bb_width'] > dataframe['bb_width'].shift(4)) & (dataframe['bb_width'] > 0.045) ) , 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( qtpylib.crossed_above(dataframe['crsi'], 55) ) , 'sell'] = 1 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: since_open = int((current_time.timestamp() - trade.open_date_utc.timestamp()) // 60) sl_pct = 1 - easeInCubic(clamp01(since_open / self.custom_stop_ramp_minutes)) sl_ramp = abs(self.stoploss) * sl_pct return sl_ramp + 0.001 # we can't go all the way to zero