# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement import pandas_ta as pta import numpy as np from pandas import DataFrame from freqtrade.strategy import IStrategy, merge_informative_pair, CategoricalParameter, IntParameter, DecimalParameter from typing import Optional from datetime import datetime from freqtrade.persistence import Trade from functools import reduce from datetime import timedelta import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib import logging log = logging.getLogger(__name__) # Originally based in the ideas of TheForce, particularly using crosses on EMA5 close and open as triggers # along with Stochastic fast. # https://github.com/StephaneTurquay/freqtrade-strategies-crypto-trading-bot # # ---- # Mods made by @hextropian (Twitter), a.k.a. as DrWho?#8511 (Discord) # Use at your own risk - no warranties of success whatsoever. class TheForceMod_3(IStrategy): INTERFACE_VERSION = 3 minimal_roi = { "0": 100 # Effectively disabled # Hyperopted on top10 pairs # "0": 0.276, # "36": 0.077, # "90": 0.031, # "145": 0 } custom_info = { # You definitely want to adjust this to better reflect your risk and portfolio 'risk_reward_ratio': 1.5, 'sl_multiplier': 3.5, 'set_to_break_even_at_profit': 1.01, } stoploss = -0.99 # Effectively disabled # Hyperopted on top10 pairs # stoploss = -0.217 use_custom_stoploss = True # We use a custom function for fixed risk/reward. # Hyperopted numbers # Trailing stop: trailing_stop = False trailing_stop_positive = 0.341 trailing_stop_positive_offset = 0.359 trailing_only_offset_is_reached = False # Optimal timeframe for the strategy. timeframe = '5m' inf_timeframe='4h' # Run "populate_indicators()" only for new candle. process_only_new_candles = True # These values can be overridden in the "ask_strategy" section in the config. use_exit_signal = True # exit_profit_only = True is dangerous. You need to keep a close eye in case of a strong # downtrend and control your exits manually. Only recommended for testing. exit_profit_only = False ignore_roi_if_entry_signal = False # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 200 # Protection parameters from Apollo11 strategy @property def protections(self): return [ { # Don't enter a trade right after selling a trade. "method": "CooldownPeriod", "stop_duration": to_minutes(minutes=5), }, { # Stop trading if max-drawdown is reached. "method": "MaxDrawdown", "lookback_period": to_minutes(hours=12), "trade_limit": 20, # Considering all pairs that have a minimum of 20 trades "stop_duration": to_minutes(hours=1), "max_allowed_drawdown": 0.2, # If max-drawdown is > 20% this will activate }, { # Stop trading if a certain amount of stoploss occurred within a certain time window. "method": "StoplossGuard", "lookback_period": to_minutes(hours=6), "trade_limit": 4, # Considering all pairs that have a minimum of 4 trades "stop_duration": to_minutes(minutes=30), "only_per_pair": False, # Looks at all pairs }, { # Lock pairs with low profits "method": "LowProfitPairs", "lookback_period": to_minutes(hours=1, minutes=30), "trade_limit": 2, # Considering all pairs that have a minimum of 2 trades "stop_duration": to_minutes(hours=2), "required_profit": 0.02, # If profit < 2% this will activate for a pair }, { # Lock pairs with low profits "method": "LowProfitPairs", "lookback_period": to_minutes(hours=6), "trade_limit": 4, # Considering all pairs that have a minimum of 4 trades "stop_duration": to_minutes(minutes=30), "required_profit": 0.01, # If profit < 1% this will activate for a pair }, ] def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ custom_stoploss using a fixed risk/reward ratio. Based on: https://github.com/freqtrade/freqtrade-strategies/blob/main/user_data/strategies/FixedRiskRewardLoss.py """ result = break_even_sl = takeprofit_sl = -1 custom_info_pair = self.custom_info.get(pair) if custom_info_pair is not None: # using current_time/open_date directly via custom_info_pair[trade.open_daten] # would only work in backtesting/hyperopt. # in live/dry-run, we have to search for nearest row before it open_date_mask = custom_info_pair.index.unique().get_loc(trade.open_date_utc, method='ffill') open_df = custom_info_pair.iloc[open_date_mask] # trade might be open too long for us to find opening candle if(len(open_df) != 1): return -1 # won't update current stoploss initial_sl_abs = open_df['stoploss_rate'] # calculate initial stoploss at open_date initial_sl = initial_sl_abs/current_rate-1 # calculate take profit treshold # by using the initial risk and multiplying it risk_distance = trade.open_rate-initial_sl_abs reward_distance = risk_distance*self.custom_info['risk_reward_ratio'] # take_profit tries to lock in profit once price gets over # risk/reward ratio treshold take_profit_price_abs = trade.open_rate+reward_distance # take_profit gets triggerd at this profit take_profit_pct = take_profit_price_abs/trade.open_rate-1 # break_even tries to set sl at open_rate+fees (0 loss) break_even_profit_distance = risk_distance*self.custom_info['set_to_break_even_at_profit'] # break_even gets triggerd at this profit break_even_profit_pct = (break_even_profit_distance+current_rate)/current_rate-1 result = initial_sl if(current_profit >= break_even_profit_pct): break_even_sl = (trade.open_rate*(1+trade.fee_open+trade.fee_close) / current_rate)-1 result = break_even_sl if(current_profit >= take_profit_pct): takeprofit_sl = take_profit_price_abs/current_rate-1 result = takeprofit_sl return result def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ From NostalgiaForInfinityX by iterativ https://github.com/iterativv/NostalgiaForInfinity allow force entries and protects against slippage. """ if (entry_tag == 'force_entry'): return True dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if(len(dataframe) < 1): return True dataframe = dataframe.iloc[-1].squeeze() if ((rate > dataframe['close'])): slippage = ((rate / dataframe['close']) - 1.0) if slippage < 0.044: return True else: log.warning( "Cancelling buy for %s due to slippage %s", pair, slippage ) return False return True def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs_4h = [(pair, self.inf_timeframe) for pair in pairs] # Explicitly adding BTC and ETH in case they're not already in the whitelist. informative_pairs_4h += [("ETH/USDT", self.inf_timeframe), ("BTC/USDT", self.inf_timeframe), ] return informative_pairs_4h def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # assert self.dp, 'DataProvider is required for multiple timeframes.' # Get the informative pairs informative_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_timeframe) # Get informative BTC and ETH informative_BTC = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe=self.inf_timeframe) informative_ETH = self.dp.get_pair_dataframe(pair="ETH/USDT", timeframe=self.inf_timeframe) btc_t3 = ta.T3(informative_BTC, timeperiod=11, vfactor=0.7) eth_t3 = ta.T3(informative_ETH, timeperiod=11, vfactor=0.7) current_t3 = ta.T3(informative_4h, timeperiod=11, vfactor=0.7) # Trend directions for BTC, ETH and the current pair, used as a reference. dataframe['macro_ok'] = ( # Candles are green and above T3 # Hoghly correlated assets will likely follow the trend. (informative_BTC['close'] >= informative_BTC['open']) & (informative_BTC['close'] >= btc_t3) & (informative_ETH['close'] >= informative_ETH['open']) & (informative_ETH['close'] >= eth_t3) & (informative_4h['close'] >= informative_4h['open']) & (informative_4h['close'] >= current_t3) ) # Decreasing size candles and trading below T3. dataframe['macro_downtrend'] = ( (informative_BTC['close'] < btc_t3) & (informative_BTC['close'] < informative_BTC['close'].shift(1)) & (informative_ETH['close'] < eth_t3) & (informative_ETH['close'] < informative_ETH['close'].shift(1)) & (informative_4h['close'] < current_t3) & (informative_4h['close'] < informative_4h['close'].shift(1)) ) # Stochastic Fast stoch_fast = ta.STOCHF(dataframe,5,3,3) dataframe['fastd'] = stoch_fast['fastd'] dataframe['fastk'] = stoch_fast['fastk'] # MACD macd = ta.MACD(dataframe,12,26,1) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # EMA 5 for close and open dataframe['ema5c'] = ta.EMA(dataframe['close'], timeperiod=5) dataframe['ema5o'] = ta.EMA(dataframe['open'], timeperiod=5) # ADX - great to validate actual trends dataframe['adx'] = ta.ADX(dataframe) # t3 dataframe['t3'] = ta.T3(dataframe, timeperiod=11, vfactor=0.7) # Stop loss dataframe['stoploss_rate'] = dataframe['close'] - (ta.ATR(dataframe, timeperiod=32) * self.custom_info['sl_multiplier']) self.custom_info[metadata['pair']] = dataframe[['date', 'stoploss_rate']].copy().set_index('date') return dataframe ################################################################################# ## ## ## BUY (Enter) conditions ## ## ## ################################################################################# def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dont_buy_conditions = [] dont_buy_conditions.append( ( # don't buy if there isn't 2% profit to be made (dataframe['close'].rolling(48).max() < (dataframe['close'] * 1.008)) ) ) dont_buy_conditions.append( ( # don't buy if we're in a macro downtrend (dataframe['macro_downtrend']) ) ) dataframe.loc[ ( ( (qtpylib.crossed_above(dataframe['ema5c'], dataframe['ema5o'])) & (dataframe['macro_ok']) & (dataframe['fastk'] >= 20) & (dataframe['fastk'] <= 80) & (dataframe['fastd'] >= 20) & (dataframe['fastd'] <= 80) & (dataframe['fastk'] >= dataframe['fastd']) & (dataframe['macd'] > dataframe['macd'].shift(1)) & (dataframe['macdsignal'] > dataframe['macdsignal'].shift(1)) & (dataframe['close'] > dataframe['close'].shift(1)) & (dataframe['close'] >= dataframe['t3']) & (dataframe['volume'] > 0) ) ), 'enter_long'] = 1 if dont_buy_conditions: for condition in dont_buy_conditions: dataframe.loc[condition, 'enter_long'] = 0 return dataframe ################################################################################# ## ## ## SELL (Exit) conditions ## ## ## ################################################################################# def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( ( (qtpylib.crossed_below(dataframe['ema5c'], dataframe['ema5o'])) & (dataframe['fastk'] <= 80) & (dataframe['fastd'] <= 80) & (dataframe['fastk'] < dataframe['fastd']) & (dataframe['macd'] < dataframe['macd'].shift(1)) & (dataframe['macdsignal'] < dataframe['macdsignal'].shift(1)) & (dataframe['adx'] >= 18) & (dataframe['volume'] > 0) ) | ( # Sudden strong downturn in BTC and ETH and we're also trading below T3 (dataframe['macro_downtrend']) & (dataframe['close'] < dataframe['t3']) ) ), 'exit_long'] = 1 return dataframe # Helper function def to_minutes(**timdelta_kwargs): return int(timedelta(**timdelta_kwargs).total_seconds() / 60)