# source: https://raw.githubusercontent.com/DerSalvador/freqtrade-helm-chart/a669dc11b640b0eb63aa8f8b51e9f181fd7ee43c/chart/deployed_strategies/binance-michael-k8s-namespace/InverseV2.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # --- Do not remove these libs --- import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame, Series from freqtrade.strategy import IStrategy from freqtrade.strategy import merge_informative_pair, timeframe_to_minutes from freqtrade.strategy import CategoricalParameter, DecimalParameter, IntParameter # -------------------------------- # Add your lib to import here import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib import numpy # noqa from freqtrade.persistence import Trade from datetime import datetime, timedelta class Github_DerSalvador_freqtrade_helm_chart__InverseV2__20260115_122204(IStrategy): INTERFACE_VERSION = 3 # Buy hyperspace params: entry_params = {'entry_fisher_cci_1': -0.42, 'entry_fisher_cci_2': 0.41, 'entry_fisher_length': 31} # Sell hyperspace params: exit_params = {'exit_fisher_cci_1': 0.42, 'exit_fisher_cci_2': -0.34} # ROI table: minimal_roi = {'0': 100} # Stoploss: stoploss = -0.2 # Trailing stop: trailing_stop = True trailing_stop_positive = 0.078 trailing_stop_positive_offset = 0.174 trailing_only_offset_is_reached = False # Optimal timeframe for the strategy. timeframe = '1h' info_timeframe = '4h' # Run "populate_indicators()" only for new candle. process_only_new_candles = False # These values can be overridden in the "ask_strategy" section in the config. use_exit_signal = True 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 # Optional order type mapping. order_types = {'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': False} # Optional order time in force. order_time_in_force = {'entry': 'gtc', 'exit': 'gtc'} plot_config = {'main_plot': {}, 'subplots': {'fisher': {'fisher_stoch': {'color': 'blue'}, 'fisher_cci': {'color': 'red'}, 'fisher_rsi': {'color': 'black'}, 'fisher_mfi': {'color': 'purple'}}}} # Hyperoptable parameters entry_fisher_length = IntParameter(low=13, high=55, default=34, space='entry', optimize=True, load=True) entry_fisher_cci_1 = DecimalParameter(low=-0.6, high=-0.3, decimals=2, default=-0.5, space='entry', optimize=True, load=True) entry_fisher_cci_2 = DecimalParameter(low=0.3, high=0.6, decimals=2, default=0.5, space='entry', optimize=True, load=True) exit_fisher_cci_1 = DecimalParameter(low=0.3, high=0.6, decimals=2, default=0.5, space='exit', optimize=True, load=True) exit_fisher_cci_2 = DecimalParameter(low=-0.6, high=-0.3, decimals=2, default=-0.5, space='exit', optimize=True, load=True) def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] previous_candle_1 = dataframe.iloc[-2] if last_candle is not None: # if (exit_reason in ['roi','exit_signal','trailing_stop_loss']): if exit_reason in ['exit_signal']: if last_candle['di_up'] and last_candle['adx'] > previous_candle_1['adx']: return False return True def informative_pairs(self): # get access to all pairs available in whitelist. pairs = self.dp.current_whitelist() # Assign tf to each pair so they can be downloaded and cached for strategy. informative_pairs = [(pair, self.info_timeframe) for pair in pairs] informative_pairs.append(('BTC/USDT', self.info_timeframe)) return informative_pairs def informative_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, 'DataProvider is required for multiple timeframes.' # Get the informative pair informative_p = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.info_timeframe) # EMA informative_p['ema_50'] = ta.EMA(informative_p, timeperiod=50) informative_p['ema_100'] = ta.EMA(informative_p, timeperiod=100) informative_p['ema_200'] = ta.EMA(informative_p, timeperiod=200) # SSL Channels ssl_down, ssl_up = self.SSLChannels(informative_p, 20) informative_p['ssl_down'] = ssl_down informative_p['ssl_up'] = ssl_up return informative_p def informative_btc_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, 'DataProvider is required for multiple timeframes.' informative_btc = self.dp.get_pair_dataframe(pair='BTC/USDT', timeframe=self.info_timeframe) informative_btc['btc_cci'] = ta.CCI(informative_btc) return informative_btc def normal_tf_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # RSI # dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.entry_fisher_length.value) # # Inverse Fisher transform on RSI, values [-1.0, 1.0] (https://goo.gl/2JGGoy) # rsi = 0.1 * (dataframe['rsi'] - 50) # wmarsi = ta.WMA(rsi, timeperiod = 9) # dataframe['fisher_rsi'] = (numpy.exp(2 * wmarsi) - 1) / (numpy.exp(2 * wmarsi) + 1) # # MFI - Money Flow Index # dataframe['mfi'] = ta.MFI(dataframe, timeperiod=self.entry_fisher_length.value) # # Inverse Fisher transform on MFI # mfi = 0.1 * (dataframe['mfi'] - 50) # wmamfi = ta.WMA(mfi, timeperiod = 9) # dataframe['fisher_mfi'] = (numpy.exp(2 * wmamfi) - 1) / (numpy.exp(2 * wmamfi) + 1) # # Stochastic # stoch_fast = ta.STOCHF(dataframe, fastk_period=self.entry_fisher_length.value) # dataframe['fastk'] = stoch_fast['fastk'] # # Inverse Fisher transform on Stochastic # stoch = 0.1 * (dataframe['fastk'] - 50) # wmastoch = ta.WMA(stoch, timeperiod = 9) # dataframe['fisher_stoch'] = (numpy.exp(2 * wmastoch) - 1) / (numpy.exp(2 * wmastoch) + 1) # Commodity Channel Index: values [Oversold:-100, Overbought:100] for cci_length in self.entry_fisher_length.range: dataframe[f'cci'] = ta.CCI(dataframe, timeperiod=cci_length) # Inverse Fisher transform on CCI cci = 0.1 * (dataframe[f'cci'] / 4) wmacci = ta.WMA(cci, timeperiod=9) dataframe[f'fisher_cci_{cci_length}'] = (numpy.exp(2 * wmacci) - 1) / (numpy.exp(2 * wmacci) + 1) # dataframe['fisher_average'] = ( # (dataframe['fisher_rsi'] + # dataframe['fisher_cci'] + # dataframe['fisher_mfi'] + # dataframe['fisher_stoch'] # ) / 4).astype(float) dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200) # confirm_trade_exit dataframe['adx'] = ta.ADX(dataframe, timeperiod=3) dataframe['di_up'] = ta.PLUS_DI(dataframe, timeperiod=3) > ta.MINUS_DI(dataframe, timeperiod=3) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ --> Informative timeframe ___________________________________________________________________________________________ """ if self.info_timeframe != 'none': informative_p = self.informative_indicators(dataframe, metadata) dataframe = merge_informative_pair(dataframe, informative_p, self.timeframe, self.info_timeframe, ffill=True) drop_columns = [s + '_' + self.info_timeframe for s in ['date']] dataframe.drop(columns=dataframe.columns.intersection(drop_columns), inplace=True) '\n --> Informative btc timeframe\n ___________________________________________________________________________________________\n ' if self.info_timeframe != 'none': informative_btc = self.informative_btc_indicators(dataframe, metadata) dataframe = merge_informative_pair(dataframe, informative_btc, self.timeframe, self.info_timeframe, ffill=True) drop_columns = [s + '_' + self.info_timeframe for s in ['date']] dataframe.drop(columns=dataframe.columns.intersection(drop_columns), inplace=True) '\n --> The indicators for the normal timeframe\n ___________________________________________________________________________________________\n ' dataframe = self.normal_tf_indicators(dataframe, metadata) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Make sure Volume is not 0 dataframe.loc[(qtpylib.crossed_above(dataframe[f'fisher_cci_{self.entry_fisher_length.value}'], self.entry_fisher_cci_1.value) | (qtpylib.crossed_below(dataframe[f'fisher_cci_{self.entry_fisher_length.value}'], self.entry_fisher_cci_2.value).rolling(8).max() == 1) & qtpylib.crossed_above(dataframe[f'fisher_cci_{self.entry_fisher_length.value}'], self.entry_fisher_cci_2.value)) & (dataframe[f'ssl_up_{self.info_timeframe}'] > dataframe[f'ssl_down_{self.info_timeframe}']) & (dataframe['ema_50'] > dataframe['ema_200']) & (dataframe[f'ema_50_{self.info_timeframe}'] > dataframe[f'ema_100_{self.info_timeframe}']) & (dataframe[f'ema_50_{self.info_timeframe}'] > dataframe[f'ema_200_{self.info_timeframe}']) & (dataframe[f'btc_cci_{self.info_timeframe}'] < 0) & (dataframe['volume'] > 0), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Make sure Volume is not 0 dataframe.loc[(qtpylib.crossed_below(dataframe[f'fisher_cci_{self.entry_fisher_length.value}'], self.exit_fisher_cci_1.value) | qtpylib.crossed_below(dataframe[f'fisher_cci_{self.entry_fisher_length.value}'], self.exit_fisher_cci_2.value)) & (dataframe['volume'] > 0), 'exit_long'] = 1 return dataframe # SSL Channels def SSLChannels(self, dataframe, length=7): 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'])