diff --git a/Strategies/HarmonicDivergence.py b/Strategies/HarmonicDivergence.py
new file mode 100644
index 0000000..543d140
--- /dev/null
+++ b/Strategies/HarmonicDivergence.py
@@ -0,0 +1,755 @@
+# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
+# flake8: noqa: F401
+
+# --- Do not remove these libs ---
+import datetime
+from typing import List, Tuple
+import numpy as np # noqa
+import pandas as pd # noqa
+pd.options.mode.chained_assignment = None
+from pandas import DataFrame, Series
+from technical.util import resample_to_interval, resampled_merge
+from freqtrade.strategy import IStrategy, merge_informative_pair
+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
+from collections import deque
+
+
+class PlotConfig():
+
+ def __init__(self):
+ self.config = {
+ 'main_plot': {
+ resample('bollinger_upperband') : {'color': 'rgba(4,137,122,0.7)'},
+ resample('kc_upperband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('kc_middleband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('kc_lowerband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('bollinger_lowerband') : {
+ 'color': 'rgba(4,137,122,0.7)',
+ 'fill_to': resample('bollinger_upperband'),
+ 'fill_color': 'rgba(4,137,122,0.07)'
+ },
+ resample('ema9') : {'color': 'purple'},
+ resample('ema20') : {'color': 'yellow'},
+ resample('ema50') : {'color': 'red'},
+ resample('ema200') : {'color': 'white'},
+ },
+ 'subplots': {
+ "ATR" : {
+ resample('atr'):{'color':'firebrick'}
+ }
+ }
+ }
+
+ def add_pivots_in_config(self):
+ self.config['main_plot']["pivot_lows"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'olive'
+ }
+ }
+ }
+ self.config['main_plot']["pivot_highs"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'violet'
+ }
+ }
+ }
+ self.config['main_plot']["pivot_highs"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'violet'
+ }
+ }
+ }
+ return self
+
+ def add_divergence_in_config(self, indicator:str):
+ # self.config['main_plot']["bullish_divergence_" + indicator + "_occurence"] = {
+ # "plotly": {
+ # 'mode': 'markers',
+ # 'marker': {
+ # 'symbol': 'diamond',
+ # 'size': 11,
+ # 'line': {
+ # 'width': 2
+ # },
+ # 'color': 'orange'
+ # }
+ # }
+ # }
+ # self.config['main_plot']["bearish_divergence_" + indicator + "_occurence"] = {
+ # "plotly": {
+ # 'mode': 'markers',
+ # 'marker': {
+ # 'symbol': 'diamond',
+ # 'size': 11,
+ # 'line': {
+ # 'width': 2
+ # },
+ # 'color': 'purple'
+ # }
+ # }
+ # }
+ for i in range(3):
+ self.config['main_plot']["bullish_divergence_" + indicator + "_line_" + str(i)] = {
+ "plotly": {
+ 'mode': 'lines',
+ 'line' : {
+ 'color': 'green',
+ 'dash' :'dash'
+ }
+ }
+ }
+ self.config['main_plot']["bearish_divergence_" + indicator + "_line_" + str(i)] = {
+ "plotly": {
+ 'mode': 'lines',
+ 'line' : {
+ "color":'crimson',
+ 'dash' :'dash'
+ }
+ }
+ }
+ return self
+
+ def add_total_divergences_in_config(self, dataframe):
+ total_bullish_divergences_count = dataframe[resample("total_bullish_divergences_count")]
+ total_bullish_divergences_names = dataframe[resample("total_bullish_divergences_names")]
+ self.config['main_plot'][resample("total_bullish_divergences")] = {
+ "plotly": {
+ 'mode': 'markers+text',
+ 'text': total_bullish_divergences_count,
+ 'hovertext': total_bullish_divergences_names,
+ 'textfont':{'size': 11, 'color':'green'},
+ 'textposition':'bottom center',
+ 'marker': {
+ 'symbol': 'diamond',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'green'
+ }
+ }
+ }
+ total_bearish_divergences_count = dataframe[resample("total_bearish_divergences_count")]
+ total_bearish_divergences_names = dataframe[resample("total_bearish_divergences_names")]
+ self.config['main_plot'][resample("total_bearish_divergences")] = {
+ "plotly": {
+ 'mode': 'markers+text',
+ 'text': total_bearish_divergences_count,
+ 'hovertext': total_bearish_divergences_names,
+ 'textfont':{'size': 11, 'color':'crimson'},
+ 'textposition':'top center',
+ 'marker': {
+ 'symbol': 'diamond',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'crimson'
+ }
+ }
+ }
+ return self
+
+class HarmonicDivergence(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_*
+ """
+ # 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
+
+ # Minimal ROI designed for the strategy.
+ # This attribute will be overridden if the config file contains "minimal_roi".
+ minimal_roi = {
+ "300" : 0.02,
+ "60": 0.04,
+ "30": 0.06,
+ "0": 0.05,
+ # "420" : 0.005,
+ # "300" : 0.007,
+ # "240" : 0.009,
+ #"0": 0.018
+ #"0": 0.007
+ }
+
+ # Optimal stoploss designed for the strategy.
+ # This attribute will be overridden if the config file contains "stoploss".
+ stoploss = -0.5
+
+ use_custom_stoploss = True
+
+ # Trailing stoploss
+ trailing_stop = True
+ trailing_stop_positive = 0.02
+ trailing_stop_positive_offset = 0.03 # Disabled / not configured
+ trailing_only_offset_is_reached = True
+
+ # Optimal timeframe for the strategy.
+ timeframe = '15m'
+
+ # 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_sell_signal = True
+ sell_profit_only = False
+ ignore_roi_if_buy_signal = False
+
+ # Number of candles the strategy requires before producing valid signals
+ startup_candle_count: int = 30
+
+ # Optional order type mapping.
+ order_types = {
+ 'buy': 'market',
+ 'sell': 'market',
+ 'stoploss': 'limit',
+ 'stoploss_on_exchange': False
+ }
+
+ # Optional order time in force.
+ order_time_in_force = {
+ 'buy': 'gtc',
+ 'sell': 'gtc'
+ }
+
+ plot_config = None
+
+ def get_ticker_indicator(self):
+ return int(self.timeframe[:-1])
+
+ 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
+ """
+
+ # Get the informative pair
+ # informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='15m')
+ # informative = resample_to_interval(dataframe, self.get_ticker_indicator() * 15)
+ informative = dataframe
+ # Momentum Indicators
+ # ------------------------------------
+
+ # RSI
+ informative['rsi'] = ta.RSI(informative)
+ # Stochastic Slow
+ informative['stoch'] = ta.STOCH(informative)['slowk']
+ # ROC
+ informative['roc'] = ta.ROC(informative)
+ # Ultimate Oscillator
+ informative['uo'] = ta.ULTOSC(informative)
+ # Awesome Oscillator
+ informative['ao'] = qtpylib.awesome_oscillator(informative)
+ # MACD
+ informative['macd'] = ta.MACD(informative)['macd']
+ # Commodity Channel Index
+ informative['cci'] = ta.CCI(informative)
+ # CMF
+ informative['cmf'] = chaikin_money_flow(informative, 20)
+ # OBV
+ informative['obv'] = ta.OBV(informative)
+ # MFI
+ informative['mfi'] = ta.MFI(informative)
+ # ADX
+ informative['adx'] = ta.ADX(informative)
+
+ # ATR
+ informative['atr'] = qtpylib.atr(informative, window=14, exp=False)
+
+ # Keltner Channel
+ # keltner = qtpylib.keltner_channel(dataframe, window=20, atrs=1)
+ keltner = emaKeltner(informative)
+ informative["kc_upperband"] = keltner["upper"]
+ informative["kc_middleband"] = keltner["mid"]
+ informative["kc_lowerband"] = keltner["lower"]
+
+ # Bollinger Bands
+ bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(informative), window=20, stds=2)
+ informative['bollinger_upperband'] = bollinger['upper']
+ informative['bollinger_lowerband'] = bollinger['lower']
+
+ # EMA - Exponential Moving Average
+ informative['ema9'] = ta.EMA(informative, timeperiod=9)
+ informative['ema20'] = ta.EMA(informative, timeperiod=20)
+ informative['ema50'] = ta.EMA(informative, timeperiod=50)
+ informative['ema200'] = ta.EMA(informative, timeperiod=200)
+
+ pivots = pivot_points(informative)
+ informative['pivot_lows'] = pivots['pivot_lows']
+ informative['pivot_highs'] = pivots['pivot_highs']
+
+ # Use the helper function merge_informative_pair to safely merge the pair
+ # Automatically renames the columns and merges a shorter timeframe dataframe and a longer timeframe informative pair
+ # use ffill to have the 1d value available in every row throughout the day.
+ # Without this, comparisons between columns of the original and the informative pair would only work once per day.
+ # Full documentation of this method, see below
+
+
+ initialize_divergences_lists(informative)
+ add_divergences(informative, 'rsi')
+ add_divergences(informative, 'stoch')
+ add_divergences(informative, 'roc')
+ add_divergences(informative, 'uo')
+ add_divergences(informative, 'ao')
+ add_divergences(informative, 'macd')
+ add_divergences(informative, 'cci')
+ add_divergences(informative, 'cmf')
+ add_divergences(informative, 'obv')
+ add_divergences(informative, 'mfi')
+ add_divergences(informative, 'adx')
+
+ # print("-------------------informative-------------------")
+ # print(informative)
+ # print("-------------------dataframe-------------------")
+ # print(dataframe)
+ # dataframe = merge_informative_pair(dataframe, informative, self.timeframe, '15m', ffill=True)
+
+ # dataframe = resampled_merge(dataframe, informative)
+ # print(dataframe[resample("total_bullish_divergences_count")])
+ # for index, value in enumerate(dataframe[resample("total_bullish_divergences_count")]):
+ # if value < 0.5:
+ # dataframe[resample("total_bullish_divergences_count")][index] = None
+ # dataframe[resample("total_bullish_divergences")][index] = None
+ # dataframe[resample("total_bullish_divergences_names")][index] = None
+ # else:
+ # print(value)
+ # print(dataframe[resample("total_bullish_divergences")][index])
+ # print(dataframe[resample("total_bullish_divergences_names")][index])
+ HarmonicDivergence.plot_config = (
+ PlotConfig()
+ # .add_pivots_in_config()
+ # .add_divergence_in_config('rsi')
+ # .add_divergence_in_config('stoch')
+ # .add_divergence_in_config('roc')
+ # .add_divergence_in_config('uo')
+ # .add_divergence_in_config('ao')
+ # .add_divergence_in_config('macd')
+ # .add_divergence_in_config('cci')
+ # .add_divergence_in_config('cmf')
+ # .add_divergence_in_config('obv')
+ # .add_divergence_in_config('mfi')
+ # .add_divergence_in_config('adx')
+ .add_total_divergences_in_config(dataframe)
+ .config)
+
+ 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
+ """
+ dataframe.loc[
+ (
+ (dataframe[resample('total_bullish_divergences')].shift() > 0)
+ # # & (dataframe['high'] > dataframe['high'].shift())
+ # & (
+ # (keltner_middleband_check(dataframe) & (ema_check(dataframe)) & (green_candle(dataframe)))
+ # # (keltner_middleband_check(dataframe) & (green_candle(dataframe)))
+ # | (keltner_lowerband_check(dataframe) & (ema_check(dataframe)))
+ # # | keltner_lowerband_check(dataframe)
+ # # | (keltner_lowerband_check(dataframe) & (green_candle(dataframe)))
+ # | (bollinger_lowerband_check(dataframe) & (ema_check(dataframe)))
+ # )
+ & two_bands_check(dataframe)
+ # # & bollinger_keltner_check(dataframe)
+ # & ema_cross_check(dataframe)
+ & (dataframe['volume'] > 0) # Make sure Volume is not 0
+ ),
+ '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
+ """
+ dataframe.loc[
+ (
+ (dataframe['volume'] > 0) # Make sure Volume is not 0
+ ),
+ 'sell'] = 0
+ return dataframe
+
+ def custom_sell(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
+ current_profit: float, **kwargs):
+
+ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+ takeprofit = 999999
+ # self.trailing_stop = False
+
+ for i in range(1,len(dataframe['close'])):
+ if dataframe.iloc[-i]['date'].to_pydatetime().replace(tzinfo=datetime.timezone.utc) == trade.open_date_utc:
+ buy_candle = dataframe.iloc[-i-1].squeeze()
+ takeprofit = buy_candle[resample('high')] + buy_candle[resample('atr')]
+ break
+
+ # if takeprofit < current_rate:
+ # self.trailing_stop = True
+ # return True
+
+ def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
+ current_rate: float, current_profit: float, **kwargs) -> float:
+
+ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+ stoploss = 999999
+
+ for i in range(1,len(dataframe['close'])):
+ if dataframe.iloc[-i]['date'].to_pydatetime().replace(tzinfo=datetime.timezone.utc) == trade.open_date_utc:
+ buy_candle = dataframe.iloc[-i-1].squeeze()
+ stoploss = buy_candle[resample('low')] - buy_candle[resample('atr')]
+ # stoploss = buy_candle[resample('high')] - buy_candle[resample('atr')]
+ break
+
+ # Convert absolute price to percentage relative to current_rate
+ if stoploss < current_rate:
+ return (stoploss / current_rate) - 1
+
+ # return maximum stoploss value, keeping current stoploss price unchanged
+ return 1
+
+def resample(indicator):
+ # return "resample_15_" + indicator
+ return indicator
+
+def two_bands_check(dataframe):
+ check = (
+ # ((dataframe['low'] < dataframe['bollinger_lowerband']) & (dataframe['high'] > dataframe['kc_lowerband'])) |
+ ((dataframe[resample('low')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('kc_upperband')])) # 1
+ # ((dataframe['low'] < dataframe['kc_lowerband']) & (dataframe['high'] > dataframe['kc_middleband'])) # 2
+ # | ((dataframe['low'] < dataframe['kc_middleband']) & (dataframe['high'] > dataframe['kc_upperband'])) # 2
+ )
+ return ~check
+
+def ema_cross_check(dataframe):
+ dataframe['ema20_50_cross'] = qtpylib.crossed_below(dataframe[resample('ema20')],dataframe[resample('ema50')])
+ dataframe['ema20_200_cross'] = qtpylib.crossed_below(dataframe[resample('ema20')],dataframe[resample('ema200')])
+ dataframe['ema50_200_cross'] = qtpylib.crossed_below(dataframe[resample('ema50')],dataframe[resample('ema200')])
+ return ~(
+ dataframe['ema20_50_cross']
+ | dataframe['ema20_200_cross']
+ | dataframe['ema50_200_cross']
+ )
+
+def green_candle(dataframe):
+ return dataframe[resample('open')] < dataframe[resample('close')]
+
+def keltner_middleband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('kc_middleband')]) & (dataframe[resample('high')] > dataframe[resample('kc_middleband')])
+
+def keltner_lowerband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('kc_lowerband')])
+
+def bollinger_lowerband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('bollinger_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('bollinger_lowerband')])
+
+def bollinger_keltner_check(dataframe):
+ return (dataframe[resample('bollinger_lowerband')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('bollinger_upperband')] > dataframe[resample('kc_upperband')])
+
+def ema_check(dataframe):
+ check = (
+ (dataframe[resample('ema9')] < dataframe[resample('ema20')])
+ & (dataframe[resample('ema20')] < dataframe[resample('ema50')])
+ & (dataframe[resample('ema50')] < dataframe[resample('ema200')]))
+ return ~check
+
+def initialize_divergences_lists(dataframe: DataFrame):
+ # 使用 `.loc` 单步赋值来替代链式赋值
+ dataframe.loc[:, "total_bullish_divergences"] = np.nan
+ dataframe.loc[:, "total_bullish_divergences_count"] = 0
+ dataframe.loc[:, "total_bullish_divergences_names"] = ''
+ dataframe.loc[:, "total_bearish_divergences"] = np.nan
+ dataframe.loc[:, "total_bearish_divergences_count"] = 0
+ dataframe.loc[:, "total_bearish_divergences_names"] = ''
+
+def add_divergences(dataframe: DataFrame, indicator: str):
+ bearish_divergences, bearish_lines, bullish_divergences, bullish_lines = divergence_finder_dataframe(dataframe, indicator)
+ dataframe.loc[:, 'bearish_divergence_' + indicator + '_occurence'] = bearish_divergences
+ dataframe.loc[:, 'bullish_divergence_' + indicator + '_occurence'] = bullish_divergences
+
+def divergence_finder_dataframe(dataframe: DataFrame, indicator_source: str) -> Tuple[pd.Series, pd.Series]:
+ bearish_lines = [np.empty(len(dataframe['close'])) * np.nan]
+ bearish_divergences = np.empty(len(dataframe['close'])) * np.nan
+ bullish_lines = [np.empty(len(dataframe['close'])) * np.nan]
+ bullish_divergences = np.empty(len(dataframe['close'])) * np.nan
+ low_iterator = []
+ high_iterator = []
+
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+ if np.isnan(row.pivot_lows):
+ low_iterator.append(0 if len(low_iterator) == 0 else low_iterator[-1])
+ else:
+ low_iterator.append(index)
+ if np.isnan(row.pivot_highs):
+ high_iterator.append(0 if len(high_iterator) == 0 else high_iterator[-1])
+ else:
+ high_iterator.append(index)
+
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+
+ bearish_occurence = bearish_divergence_finder(dataframe,
+ dataframe[indicator_source],
+ high_iterator,
+ index)
+
+ if bearish_occurence != None:
+ (prev_pivot , current_pivot) = bearish_occurence
+ bearish_prev_pivot = dataframe['close'][prev_pivot]
+ bearish_current_pivot = dataframe['close'][current_pivot]
+ bearish_ind_prev_pivot = dataframe[indicator_source][prev_pivot]
+ bearish_ind_current_pivot = dataframe[indicator_source][current_pivot]
+ length = current_pivot - prev_pivot
+ bearish_lines_index = 0
+ can_exist = True
+ while(True):
+ can_draw = True
+ if bearish_lines_index <= len(bearish_lines):
+ bearish_lines.append(np.empty(len(dataframe['close'])) * np.nan)
+ actual_bearish_lines = bearish_lines[bearish_lines_index]
+ for i in range(length + 1):
+ point = bearish_prev_pivot + (bearish_current_pivot - bearish_prev_pivot) * i / length
+ indicator_point = bearish_ind_prev_pivot + (bearish_ind_current_pivot - bearish_ind_prev_pivot) * i / length
+ if i != 0 and i != length:
+ if (point <= dataframe['close'][prev_pivot + i]
+ or indicator_point <= dataframe[indicator_source][prev_pivot + i]):
+ can_exist = False
+ if not np.isnan(actual_bearish_lines[prev_pivot + i]):
+ can_draw = False
+ if not can_exist:
+ break
+ if can_draw:
+ for i in range(length + 1):
+ actual_bearish_lines[prev_pivot + i] = bearish_prev_pivot + (bearish_current_pivot - bearish_prev_pivot) * i / length
+ break
+ bearish_lines_index = bearish_lines_index + 1
+ if can_exist:
+ bearish_divergences[index] = row.close
+ dataframe.loc[index, "total_bearish_divergences"] = row.close
+ if index > 30:
+ dataframe.loc[index - 30, "total_bearish_divergences_count"] += 1
+ dataframe.loc[index - 30, "total_bearish_divergences_names"] += indicator_source.upper() + '
'
+
+ bullish_occurence = bullish_divergence_finder(dataframe,
+ dataframe[indicator_source],
+ low_iterator,
+ index)
+
+ if bullish_occurence != None:
+ (prev_pivot , current_pivot) = bullish_occurence
+ bullish_prev_pivot = dataframe['close'][prev_pivot]
+ bullish_current_pivot = dataframe['close'][current_pivot]
+ bullish_ind_prev_pivot = dataframe[indicator_source][prev_pivot]
+ bullish_ind_current_pivot = dataframe[indicator_source][current_pivot]
+ length = current_pivot - prev_pivot
+ bullish_lines_index = 0
+ can_exist = True
+ while(True):
+ can_draw = True
+ if bullish_lines_index <= len(bullish_lines):
+ bullish_lines.append(np.empty(len(dataframe['close'])) * np.nan)
+ actual_bullish_lines = bullish_lines[bullish_lines_index]
+ for i in range(length + 1):
+ point = bullish_prev_pivot + (bullish_current_pivot - bullish_prev_pivot) * i / length
+ indicator_point = bullish_ind_prev_pivot + (bullish_ind_current_pivot - bullish_ind_prev_pivot) * i / length
+ if i != 0 and i != length:
+ if (point >= dataframe['close'][prev_pivot + i]
+ or indicator_point >= dataframe[indicator_source][prev_pivot + i]):
+ can_exist = False
+ if not np.isnan(actual_bullish_lines[prev_pivot + i]):
+ can_draw = False
+ if not can_exist:
+ break
+ if can_draw:
+ for i in range(length + 1):
+ actual_bullish_lines[prev_pivot + i] = bullish_prev_pivot + (bullish_current_pivot - bullish_prev_pivot) * i / length
+ break
+ bullish_lines_index = bullish_lines_index + 1
+ if can_exist:
+ bullish_divergences[index] = row.close
+ dataframe.loc[index, "total_bullish_divergences"] = row.close
+ if index > 30:
+ dataframe.loc[index - 30, "total_bullish_divergences_count"] += 1
+ dataframe.loc[index - 30, "total_bullish_divergences_names"] += indicator_source.upper() + '
'
+
+ return (bearish_divergences, bearish_lines, bullish_divergences, bullish_lines)
+
+def bearish_divergence_finder(dataframe, indicator, high_iterator, index):
+ if high_iterator[index] == index:
+ current_pivot = high_iterator[index]
+ occurences = list(dict.fromkeys(high_iterator))
+ current_index = occurences.index(high_iterator[index])
+ for i in range(current_index-1,current_index-6,-1):
+ prev_pivot = occurences[i]
+ if np.isnan(prev_pivot):
+ return
+ if ((dataframe['pivot_highs'][current_pivot] < dataframe['pivot_highs'][prev_pivot] and indicator[current_pivot] > indicator[prev_pivot])
+ or (dataframe['pivot_highs'][current_pivot] > dataframe['pivot_highs'][prev_pivot] and indicator[current_pivot] < indicator[prev_pivot])):
+ return (prev_pivot , current_pivot)
+ return None
+
+def bullish_divergence_finder(dataframe, indicator, low_iterator, index):
+ if low_iterator[index] == index:
+ current_pivot = low_iterator[index]
+ occurences = list(dict.fromkeys(low_iterator))
+ current_index = occurences.index(low_iterator[index])
+ for i in range(current_index-1,current_index-6,-1):
+ prev_pivot = occurences[i]
+ if np.isnan(prev_pivot):
+ return
+ if ((dataframe['pivot_lows'][current_pivot] < dataframe['pivot_lows'][prev_pivot] and indicator[current_pivot] > indicator[prev_pivot])
+ or (dataframe['pivot_lows'][current_pivot] > dataframe['pivot_lows'][prev_pivot] and indicator[current_pivot] < indicator[prev_pivot])):
+ return (prev_pivot, current_pivot)
+ return None
+
+from enum import Enum
+class PivotSource(Enum):
+ HighLow = 0
+ Close = 1
+
+def pivot_points(dataframe: DataFrame, window: int = 5, pivot_source: PivotSource = PivotSource.Close) -> DataFrame:
+ high_source = None
+ low_source = None
+
+ if pivot_source == PivotSource.Close:
+ high_source = 'close'
+ low_source = 'close'
+ elif pivot_source == PivotSource.HighLow:
+ high_source = 'high'
+ low_source = 'low'
+
+ pivot_points_lows = np.empty(len(dataframe['close'])) * np.nan
+ pivot_points_highs = np.empty(len(dataframe['close'])) * np.nan
+ last_values = deque()
+
+ # find pivot points
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+ last_values.append(row)
+ if len(last_values) >= window * 2 + 1:
+ current_value = last_values[window]
+ is_greater = True
+ is_less = True
+ for window_index in range(0, window):
+ left = last_values[window_index]
+ right = last_values[2 * window - window_index]
+ local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right)
+ is_greater &= local_is_greater
+ is_less &= local_is_less
+ if is_greater:
+ pivot_points_highs[index - window] = getattr(current_value, high_source)
+ if is_less:
+ pivot_points_lows[index - window] = getattr(current_value, low_source)
+ last_values.popleft()
+
+ # find last one
+ if len(last_values) >= window + 2:
+ current_value = last_values[-2]
+ is_greater = True
+ is_less = True
+ for window_index in range(0, window):
+ left = last_values[-2 - window_index - 1]
+ right = last_values[-1]
+ local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right)
+ is_greater &= local_is_greater
+ is_less &= local_is_less
+ if is_greater:
+ pivot_points_highs[index - 1] = getattr(current_value, high_source)
+ if is_less:
+ pivot_points_lows[index - 1] = getattr(current_value, low_source)
+
+ return pd.DataFrame(index=dataframe.index, data={
+ 'pivot_lows': pivot_points_lows,
+ 'pivot_highs': pivot_points_highs
+ })
+
+def check_if_pivot_is_greater_or_less(current_value, high_source: str, low_source: str, left, right) -> Tuple[bool, bool]:
+ is_greater = True
+ is_less = True
+ if (getattr(current_value, high_source) < getattr(left, high_source) or
+ getattr(current_value, high_source) < getattr(right, high_source)):
+ is_greater = False
+
+ if (getattr(current_value, low_source) > getattr(left, low_source) or
+ getattr(current_value, low_source) > getattr(right, low_source)):
+ is_less = False
+ return (is_greater, is_less)
+
+def emaKeltner(dataframe):
+ keltner = {}
+ atr = qtpylib.atr(dataframe, window=10)
+ ema20 = ta.EMA(dataframe, timeperiod=20)
+ keltner['upper'] = ema20 + atr
+ keltner['mid'] = ema20
+ keltner['lower'] = ema20 - atr
+ return keltner
+
+def chaikin_money_flow(dataframe, n=20, fillna=False) -> Series:
+ """Chaikin Money Flow (CMF)
+ It measures the amount of Money Flow Volume over a specific period.
+ http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:chaikin_money_flow_cmf
+ Args:
+ dataframe(pandas.Dataframe): dataframe containing ohlcv
+ n(int): n period.
+ fillna(bool): if True, fill nan values.
+ Returns:
+ pandas.Series: New feature generated.
+ """
+ df = dataframe.copy()
+ mfv = ((df['close'] - df['low']) - (df['high'] - df['close'])) / (df['high'] - df['low'])
+ mfv = mfv.fillna(0.0) # float division by zero
+ mfv *= df['volume']
+ cmf = (mfv.rolling(n, min_periods=0).sum()
+ / df['volume'].rolling(n, min_periods=0).sum())
+ if fillna:
+ cmf = cmf.replace([np.inf, -np.inf], np.nan).fillna(0)
+ return Series(cmf, name='cmf')
\ No newline at end of file
diff --git a/Strategies/HarmonicDivergenceFv2.py b/Strategies/HarmonicDivergenceFv2.py
new file mode 100644
index 0000000..3d5df80
--- /dev/null
+++ b/Strategies/HarmonicDivergenceFv2.py
@@ -0,0 +1,773 @@
+# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
+# flake8: noqa: F401
+
+# --- Do not remove these libs ---
+import datetime
+from typing import List, Tuple
+import numpy as np # noqa
+import pandas as pd # noqa
+pd.options.mode.chained_assignment = None
+from pandas import DataFrame, Series
+from technical.util import resample_to_interval, resampled_merge
+from freqtrade.strategy import IStrategy, merge_informative_pair
+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
+from collections import deque
+from typing import Optional, Dict, Any
+
+
+
+class PlotConfig():
+
+ def __init__(self):
+ self.config = {
+ 'main_plot': {
+ resample('bollinger_upperband') : {'color': 'rgba(4,137,122,0.7)'},
+ resample('kc_upperband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('kc_middleband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('kc_lowerband') : {'color': 'rgba(4,146,250,0.7)'},
+ resample('bollinger_lowerband') : {
+ 'color': 'rgba(4,137,122,0.7)',
+ 'fill_to': resample('bollinger_upperband'),
+ 'fill_color': 'rgba(4,137,122,0.07)'
+ },
+ resample('ema9') : {'color': 'purple'},
+ resample('ema20') : {'color': 'yellow'},
+ resample('ema50') : {'color': 'red'},
+ resample('ema200') : {'color': 'white'},
+ },
+ 'subplots': {
+ "ATR" : {
+ resample('atr'):{'color':'firebrick'}
+ }
+ }
+ }
+
+ def add_pivots_in_config(self):
+ self.config['main_plot']["pivot_lows"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'olive'
+ }
+ }
+ }
+ self.config['main_plot']["pivot_highs"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'violet'
+ }
+ }
+ }
+ self.config['main_plot']["pivot_highs"] = {
+ "plotly": {
+ 'mode': 'markers',
+ 'marker': {
+ 'symbol': 'diamond-open',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'violet'
+ }
+ }
+ }
+ return self
+
+ def add_divergence_in_config(self, indicator:str):
+ for i in range(3):
+ self.config['main_plot']["bullish_divergence_" + indicator + "_line_" + str(i)] = {
+ "plotly": {
+ 'mode': 'lines',
+ 'line' : {
+ 'color': 'green',
+ 'dash' :'dash'
+ }
+ }
+ }
+ self.config['main_plot']["bearish_divergence_" + indicator + "_line_" + str(i)] = {
+ "plotly": {
+ 'mode': 'lines',
+ 'line' : {
+ "color":'crimson',
+ 'dash' :'dash'
+ }
+ }
+ }
+ return self
+
+ def add_total_divergences_in_config(self, dataframe):
+ total_bullish_divergences_count = dataframe[resample("total_bullish_divergences_count")]
+ total_bullish_divergences_names = dataframe[resample("total_bullish_divergences_names")]
+ self.config['main_plot'][resample("total_bullish_divergences")] = {
+ "plotly": {
+ 'mode': 'markers+text',
+ 'text': total_bullish_divergences_count,
+ 'hovertext': total_bullish_divergences_names,
+ 'textfont':{'size': 11, 'color':'green'},
+ 'textposition':'bottom center',
+ 'marker': {
+ 'symbol': 'diamond',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'green'
+ }
+ }
+ }
+ total_bearish_divergences_count = dataframe[resample("total_bearish_divergences_count")]
+ total_bearish_divergences_names = dataframe[resample("total_bearish_divergences_names")]
+ self.config['main_plot'][resample("total_bearish_divergences")] = {
+ "plotly": {
+ 'mode': 'markers+text',
+ 'text': total_bearish_divergences_count,
+ 'hovertext': total_bearish_divergences_names,
+ 'textfont':{'size': 11, 'color':'crimson'},
+ 'textposition':'top center',
+ 'marker': {
+ 'symbol': 'diamond',
+ 'size': 11,
+ 'line': {
+ 'width': 2
+ },
+ 'color': 'crimson'
+ }
+ }
+ }
+ return self
+
+class HarmonicDivergenceF(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_*
+ """
+ # 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
+
+ # Minimal ROI designed for the strategy.
+ # This attribute will be overridden if the config file contains "minimal_roi".
+ minimal_roi = {
+ "300" : 0.25,
+ "60": 0.35,
+ "30": 0.45,
+ "0": 0.08,
+ }
+
+ strategy_direction = "long" # 只做多
+ position_adjustment_enable = True # 允许杠杆调整
+ max_leverage = 50 # 设置最大杠杆倍数
+
+ def leverage(self, pair: str, current_time: datetime, current_rate: float,
+ proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str,
+ **kwargs) -> float:
+ """
+ Customize leverage for each new trade. This method is only called in futures mode.
+
+ :param pair: Pair that's currently analyzed
+ :param current_time: datetime object, containing the current datetime
+ :param current_rate: Rate, calculated based on pricing settings in exit_pricing.
+ :param proposed_leverage: A leverage proposed by the bot.
+ :param max_leverage: Max leverage allowed on this pair
+ :param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal.
+ :param side: "long" or "short" - indicating the direction of the proposed trade
+ :return: A leverage amount, which is between 1.0 and max_leverage.
+ """
+ return 10.0
+
+ def adjust_position_size(self, pair: str, current_rate: float) -> float:
+ """
+ 动态调整仓位大小和杠杆
+ :param pair: 当前交易对
+ :param current_rate: 当前价格
+ :return: 动态调整后的仓位大小
+ """
+ # 获取账户余额
+ account_balance = self.wallets.get_total()
+
+ # 定义最大账户风险比例（如 2%）
+ max_risk_percentage = min(0.05, 0.02 + (pair_leverage - 1) * 0.005) # 根据杠杆调整风险比例
+ max_risk_amount = account_balance * max_risk_percentage
+
+
+ # 动态计算可用杠杆，假设波动越大杠杆越低
+ pair_leverage = self.get_pair_leverage(pair)
+
+ # ATR 动态调整：波动越大，杠杆越低（仅示例，可结合实际需求调整）
+ atr = self.get_atr(pair) # 自定义函数获取 ATR
+ dynamic_leverage = min(pair_leverage, max(5, 50 / atr)) # ATR 越大，杠杆越低
+
+ # 按风险计算最大可投入金额
+ max_risk_amount = account_balance * max_risk_percentage
+ max_position_size = max_risk_amount / abs(current_rate) # 根据当前价格计算仓位
+
+ # 动态调整的实际仓位
+ adjusted_position_size = max_position_size * dynamic_leverage
+
+ # 确保仓位不低于最小交易量限制
+ minimum_order_size = 10 # 例如 10 USDT
+ return max(adjusted_position_size, minimum_order_size)
+
+
+ # Optimal stoploss designed for the strategy.
+ # This attribute will be overridden if the config file contains "stoploss".
+ stoploss = -0.2 #没用
+
+ use_custom_stoploss = True
+
+ # Trailing stoploss
+ trailing_stop = True
+ trailing_stop_positive = 0.02
+ trailing_stop_positive_offset = 0.03 # Disabled / not configured
+ trailing_only_offset_is_reached = True
+
+ # Optimal timeframe for the strategy.
+ timeframe = '15m'
+
+ # 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 = 30
+
+ # Optional order type mapping.
+ order_types = {
+ 'entry': 'market',
+ 'exit': 'market',
+ 'stoploss': 'limit',
+ 'stoploss_on_exchange': False
+ }
+
+ order_time_in_force = {
+ 'entry': 'gtc', # 开仓订单的时间有效性（Good-Till-Cancelled）
+ 'exit': 'gtc' # 平仓订单的时间有效性
+ }
+
+ plot_config = None
+
+ def get_ticker_indicator(self):
+ return int(self.timeframe[:-1])
+
+ 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
+ """
+
+ # Get the informative pair
+ # informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='15m')
+ # informative = resample_to_interval(dataframe, self.get_ticker_indicator() * 15)
+ informative = dataframe
+ # Momentum Indicators
+ # ------------------------------------
+
+ # RSI
+ informative['rsi'] = ta.RSI(informative)
+ # Stochastic Slow
+ informative['stoch'] = ta.STOCH(informative)['slowk']
+ # ROC
+ informative['roc'] = ta.ROC(informative)
+ # Ultimate Oscillator
+ informative['uo'] = ta.ULTOSC(informative)
+ # Awesome Oscillator
+ informative['ao'] = qtpylib.awesome_oscillator(informative)
+ # MACD
+ informative['macd'] = ta.MACD(informative)['macd']
+ # Commodity Channel Index
+ informative['cci'] = ta.CCI(informative)
+ # CMF
+ informative['cmf'] = chaikin_money_flow(informative, 20)
+ # OBV
+ informative['obv'] = ta.OBV(informative)
+ # MFI
+ informative['mfi'] = ta.MFI(informative)
+ # ADX
+ informative['adx'] = ta.ADX(informative)
+
+ # ATR
+ informative['atr'] = qtpylib.atr(informative, window=14, exp=False)
+
+ # Keltner Channel
+ # keltner = qtpylib.keltner_channel(dataframe, window=20, atrs=1)
+ keltner = emaKeltner(informative)
+ informative["kc_upperband"] = keltner["upper"]
+ informative["kc_middleband"] = keltner["mid"]
+ informative["kc_lowerband"] = keltner["lower"]
+
+ # Bollinger Bands
+ bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(informative), window=20, stds=2)
+ informative['bollinger_upperband'] = bollinger['upper']
+ informative['bollinger_lowerband'] = bollinger['lower']
+
+ # EMA - Exponential Moving Average
+ informative['ema9'] = ta.EMA(informative, timeperiod=9)
+ informative['ema20'] = ta.EMA(informative, timeperiod=20)
+ informative['ema50'] = ta.EMA(informative, timeperiod=50)
+ informative['ema200'] = ta.EMA(informative, timeperiod=200)
+
+ pivots = pivot_points(informative)
+ informative['pivot_lows'] = pivots['pivot_lows']
+ informative['pivot_highs'] = pivots['pivot_highs']
+
+ initialize_divergences_lists(informative)
+ add_divergences(informative, 'rsi')
+ add_divergences(informative, 'stoch')
+ add_divergences(informative, 'roc')
+ add_divergences(informative, 'uo')
+ add_divergences(informative, 'ao')
+ add_divergences(informative, 'macd')
+ add_divergences(informative, 'cci')
+ add_divergences(informative, 'cmf')
+ add_divergences(informative, 'obv')
+ add_divergences(informative, 'mfi')
+ add_divergences(informative, 'adx')
+
+ HarmonicDivergenceF.plot_config = (
+ PlotConfig()
+ .add_total_divergences_in_config(dataframe)
+ .config)
+
+ return dataframe
+
+ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+ """
+ 定义合约策略的开多逻辑
+ """
+ dataframe.loc[
+ (
+ (dataframe[resample('total_bullish_divergences')].shift() > 0) # 存在看涨背离信号
+ & two_bands_check(dataframe) # 检查价格是否在布林带和肯特纳通道区间
+ & (dataframe['volume'] > 0) # 确保成交量大于 0
+ ),
+ 'buy'
+ ] = 1 # 生成开多信号
+
+ return dataframe
+
+
+ def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+ """
+ 定义合约策略的平多逻辑
+ """
+ dataframe.loc[
+ (
+ (dataframe['volume'] > 0) # 成交量条件
+ & (dataframe['rsi'] > 70) # RSI 超买时平多
+ # 或其他条件，如达到目标价位
+ ),
+ 'sell'
+ ] = 1 # 生成平多信号
+
+ return dataframe
+
+
+ def custom_exit(
+ self,
+ pair: str,
+ trade: 'Trade',
+ current_time: datetime,
+ current_rate: float,
+ current_profit: float,
+ **kwargs
+ ) -> Optional[float]:
+ """
+ 自定义退出逻辑，考虑杠杆的动态止盈方法。
+ 返回想要卖出的价格，或者 None 继续持有。
+
+ :param pair: 交易对
+ :param trade: 当前交易对象
+ :param current_time: 当前时间
+ :param current_rate: 当前价格
+ :param current_profit: 当前利润
+ :return: 退出价格或 None
+ """
+ try:
+ # 获取当前的数据集
+ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+
+ # 示例：根据 ATR 和杠杆动态设置止盈
+ for i in range(1, len(dataframe['close'])):
+ if dataframe.iloc[-i]['date'].to_pydatetime().replace(tzinfo=datetime.timezone.utc) == trade.open_date_utc:
+ buy_candle = dataframe.iloc[-i - 1].squeeze()
+
+ # 获取 ATR 值
+ atr = buy_candle['atr']
+
+ # 使用现有的 leverage 方法获取杠杆倍数
+ # 由于 leverage 方法需要多个参数，我们提供所需的参数
+ leverage = self.leverage(
+ pair=pair,
+ current_time=current_time,
+ current_rate=current_rate,
+ proposed_leverage=20.0, # 使用你在 leverage 方法中设置的默认值
+ max_leverage=20.0, # 最大杠杆值
+ entry_tag=None,
+ side=trade.trade_direction
+ )
+
+ # 动态计算止盈价格，根据 ATR 和杠杆调整
+ # 可以根据杠杆大小调整 ATR 的影响
+ atr_multiplier = (leverage / 20.0) # 根据最大杠杆归一化
+ takeprofit = buy_candle['high'] + (atr * atr_multiplier)
+
+ # 如果当前价格达到或超过目标价格，返回平仓价格
+ if current_rate >= takeprofit:
+ return takeprofit # 直接返回期望的平仓价格
+
+ return None # 使用默认退出逻辑
+
+ except Exception as e:
+ self.log.error(f"Error in custom_exit: {str(e)}")
+ return None
+
+
+def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
+ current_rate: float, current_profit: float, **kwargs) -> float:
+ """
+ 动态止损逻辑，根据开仓时的 ATR 设置止损
+ """
+ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+ stoploss = 999999 # 初始止损值
+
+ for i in range(1, len(dataframe['close'])):
+ if dataframe.iloc[-i]['date'].to_pydatetime().replace(tzinfo=datetime.timezone.utc) == trade.open_date_utc:
+ buy_candle = dataframe.iloc[-i-1].squeeze()
+ stoploss = buy_candle[resample('low')] - buy_candle[resample('atr')] # 使用 ATR 计算止损
+ break
+
+ if stoploss < current_rate:
+ return (stoploss / current_rate) - 1
+
+ return 1 # 未触发止损
+
+def resample(indicator):
+ # return "resample_15_" + indicator
+ return indicator
+
+def two_bands_check(dataframe):
+ check = (
+ ((dataframe[resample('low')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('kc_upperband')])) # 1
+ )
+ return ~check
+
+def ema_cross_check(dataframe):
+ dataframe['ema20_50_cross'] = qtpylib.crossed_below(dataframe[resample('ema20')],dataframe[resample('ema50')])
+ dataframe['ema20_200_cross'] = qtpylib.crossed_below(dataframe[resample('ema20')],dataframe[resample('ema200')])
+ dataframe['ema50_200_cross'] = qtpylib.crossed_below(dataframe[resample('ema50')],dataframe[resample('ema200')])
+ return ~(
+ dataframe['ema20_50_cross']
+ | dataframe['ema20_200_cross']
+ | dataframe['ema50_200_cross']
+ )
+
+def green_candle(dataframe):
+ return dataframe[resample('open')] < dataframe[resample('close')]
+
+def keltner_middleband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('kc_middleband')]) & (dataframe[resample('high')] > dataframe[resample('kc_middleband')])
+
+def keltner_lowerband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('kc_lowerband')])
+
+def bollinger_lowerband_check(dataframe):
+ return (dataframe[resample('low')] < dataframe[resample('bollinger_lowerband')]) & (dataframe[resample('high')] > dataframe[resample('bollinger_lowerband')])
+
+def bollinger_keltner_check(dataframe):
+ return (dataframe[resample('bollinger_lowerband')] < dataframe[resample('kc_lowerband')]) & (dataframe[resample('bollinger_upperband')] > dataframe[resample('kc_upperband')])
+
+def ema_check(dataframe):
+ check = (
+ (dataframe[resample('ema9')] < dataframe[resample('ema20')])
+ & (dataframe[resample('ema20')] < dataframe[resample('ema50')])
+ & (dataframe[resample('ema50')] < dataframe[resample('ema200')]))
+ return ~check
+
+def initialize_divergences_lists(dataframe: DataFrame):
+ # 使用 `.loc` 单步赋值来替代链式赋值
+ dataframe.loc[:, "total_bullish_divergences"] = np.nan
+ dataframe.loc[:, "total_bullish_divergences_count"] = 0
+ dataframe.loc[:, "total_bullish_divergences_names"] = ''
+ dataframe.loc[:, "total_bearish_divergences"] = np.nan
+ dataframe.loc[:, "total_bearish_divergences_count"] = 0
+ dataframe.loc[:, "total_bearish_divergences_names"] = ''
+
+def add_divergences(dataframe: DataFrame, indicator: str):
+ bearish_divergences, bearish_lines, bullish_divergences, bullish_lines = divergence_finder_dataframe(dataframe, indicator)
+ dataframe.loc[:, 'bearish_divergence_' + indicator + '_occurence'] = bearish_divergences
+ dataframe.loc[:, 'bullish_divergence_' + indicator + '_occurence'] = bullish_divergences
+
+def divergence_finder_dataframe(dataframe: DataFrame, indicator_source: str) -> Tuple[pd.Series, pd.Series]:
+ bearish_lines = [np.empty(len(dataframe['close'])) * np.nan]
+ bearish_divergences = np.empty(len(dataframe['close'])) * np.nan
+ bullish_lines = [np.empty(len(dataframe['close'])) * np.nan]
+ bullish_divergences = np.empty(len(dataframe['close'])) * np.nan
+ low_iterator = []
+ high_iterator = []
+
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+ if np.isnan(row.pivot_lows):
+ low_iterator.append(0 if len(low_iterator) == 0 else low_iterator[-1])
+ else:
+ low_iterator.append(index)
+ if np.isnan(row.pivot_highs):
+ high_iterator.append(0 if len(high_iterator) == 0 else high_iterator[-1])
+ else:
+ high_iterator.append(index)
+
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+
+ bearish_occurence = bearish_divergence_finder(dataframe,
+ dataframe[indicator_source],
+ high_iterator,
+ index)
+
+ if bearish_occurence != None:
+ (prev_pivot , current_pivot) = bearish_occurence
+ bearish_prev_pivot = dataframe['close'][prev_pivot]
+ bearish_current_pivot = dataframe['close'][current_pivot]
+ bearish_ind_prev_pivot = dataframe[indicator_source][prev_pivot]
+ bearish_ind_current_pivot = dataframe[indicator_source][current_pivot]
+ length = current_pivot - prev_pivot
+ bearish_lines_index = 0
+ can_exist = True
+ while(True):
+ can_draw = True
+ if bearish_lines_index <= len(bearish_lines):
+ bearish_lines.append(np.empty(len(dataframe['close'])) * np.nan)
+ actual_bearish_lines = bearish_lines[bearish_lines_index]
+ for i in range(length + 1):
+ point = bearish_prev_pivot + (bearish_current_pivot - bearish_prev_pivot) * i / length
+ indicator_point = bearish_ind_prev_pivot + (bearish_ind_current_pivot - bearish_ind_prev_pivot) * i / length
+ if i != 0 and i != length:
+ if (point <= dataframe['close'][prev_pivot + i]
+ or indicator_point <= dataframe[indicator_source][prev_pivot + i]):
+ can_exist = False
+ if not np.isnan(actual_bearish_lines[prev_pivot + i]):
+ can_draw = False
+ if not can_exist:
+ break
+ if can_draw:
+ for i in range(length + 1):
+ actual_bearish_lines[prev_pivot + i] = bearish_prev_pivot + (bearish_current_pivot - bearish_prev_pivot) * i / length
+ break
+ bearish_lines_index = bearish_lines_index + 1
+ if can_exist:
+ bearish_divergences[index] = row.close
+ dataframe.loc[index, "total_bearish_divergences"] = row.close
+ if index > 30:
+ dataframe.loc[index - 30, "total_bearish_divergences_count"] += 1
+ dataframe.loc[index - 30, "total_bearish_divergences_names"] += indicator_source.upper() + '
'
+
+ bullish_occurence = bullish_divergence_finder(dataframe,
+ dataframe[indicator_source],
+ low_iterator,
+ index)
+
+ if bullish_occurence != None:
+ (prev_pivot , current_pivot) = bullish_occurence
+ bullish_prev_pivot = dataframe['close'][prev_pivot]
+ bullish_current_pivot = dataframe['close'][current_pivot]
+ bullish_ind_prev_pivot = dataframe[indicator_source][prev_pivot]
+ bullish_ind_current_pivot = dataframe[indicator_source][current_pivot]
+ length = current_pivot - prev_pivot
+ bullish_lines_index = 0
+ can_exist = True
+ while(True):
+ can_draw = True
+ if bullish_lines_index <= len(bullish_lines):
+ bullish_lines.append(np.empty(len(dataframe['close'])) * np.nan)
+ actual_bullish_lines = bullish_lines[bullish_lines_index]
+ for i in range(length + 1):
+ point = bullish_prev_pivot + (bullish_current_pivot - bullish_prev_pivot) * i / length
+ indicator_point = bullish_ind_prev_pivot + (bullish_ind_current_pivot - bullish_ind_prev_pivot) * i / length
+ if i != 0 and i != length:
+ if (point >= dataframe['close'][prev_pivot + i]
+ or indicator_point >= dataframe[indicator_source][prev_pivot + i]):
+ can_exist = False
+ if not np.isnan(actual_bullish_lines[prev_pivot + i]):
+ can_draw = False
+ if not can_exist:
+ break
+ if can_draw:
+ for i in range(length + 1):
+ actual_bullish_lines[prev_pivot + i] = bullish_prev_pivot + (bullish_current_pivot - bullish_prev_pivot) * i / length
+ break
+ bullish_lines_index = bullish_lines_index + 1
+ if can_exist:
+ bullish_divergences[index] = row.close
+ dataframe.loc[index, "total_bullish_divergences"] = row.close
+ if index > 30:
+ dataframe.loc[index - 30, "total_bullish_divergences_count"] += 1
+ dataframe.loc[index - 30, "total_bullish_divergences_names"] += indicator_source.upper() + '
'
+
+ return (bearish_divergences, bearish_lines, bullish_divergences, bullish_lines)
+
+def bearish_divergence_finder(dataframe, indicator, high_iterator, index):
+ if high_iterator[index] == index:
+ current_pivot = high_iterator[index]
+ occurences = list(dict.fromkeys(high_iterator))
+ current_index = occurences.index(high_iterator[index])
+ for i in range(current_index-1,current_index-6,-1):
+ prev_pivot = occurences[i]
+ if np.isnan(prev_pivot):
+ return
+ if ((dataframe['pivot_highs'][current_pivot] < dataframe['pivot_highs'][prev_pivot] and indicator[current_pivot] > indicator[prev_pivot])
+ or (dataframe['pivot_highs'][current_pivot] > dataframe['pivot_highs'][prev_pivot] and indicator[current_pivot] < indicator[prev_pivot])):
+ return (prev_pivot , current_pivot)
+ return None
+
+def bullish_divergence_finder(dataframe, indicator, low_iterator, index):
+ if low_iterator[index] == index:
+ current_pivot = low_iterator[index]
+ occurences = list(dict.fromkeys(low_iterator))
+ current_index = occurences.index(low_iterator[index])
+ for i in range(current_index-1,current_index-6,-1):
+ prev_pivot = occurences[i]
+ if np.isnan(prev_pivot):
+ return
+ if ((dataframe['pivot_lows'][current_pivot] < dataframe['pivot_lows'][prev_pivot] and indicator[current_pivot] > indicator[prev_pivot])
+ or (dataframe['pivot_lows'][current_pivot] > dataframe['pivot_lows'][prev_pivot] and indicator[current_pivot] < indicator[prev_pivot])):
+ return (prev_pivot, current_pivot)
+ return None
+
+from enum import Enum
+class PivotSource(Enum):
+ HighLow = 0
+ Close = 1
+
+def pivot_points(dataframe: DataFrame, window: int = 5, pivot_source: PivotSource = PivotSource.Close) -> DataFrame:
+ high_source = None
+ low_source = None
+
+ if pivot_source == PivotSource.Close:
+ high_source = 'close'
+ low_source = 'close'
+ elif pivot_source == PivotSource.HighLow:
+ high_source = 'high'
+ low_source = 'low'
+
+ pivot_points_lows = np.empty(len(dataframe['close'])) * np.nan
+ pivot_points_highs = np.empty(len(dataframe['close'])) * np.nan
+ last_values = deque()
+
+ # find pivot points
+ for index, row in enumerate(dataframe.itertuples(index=True, name='Pandas')):
+ last_values.append(row)
+ if len(last_values) >= window * 2 + 1:
+ current_value = last_values[window]
+ is_greater = True
+ is_less = True
+ for window_index in range(0, window):
+ left = last_values[window_index]
+ right = last_values[2 * window - window_index]
+ local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right)
+ is_greater &= local_is_greater
+ is_less &= local_is_less
+ if is_greater:
+ pivot_points_highs[index - window] = getattr(current_value, high_source)
+ if is_less:
+ pivot_points_lows[index - window] = getattr(current_value, low_source)
+ last_values.popleft()
+
+ # find last one
+ if len(last_values) >= window + 2:
+ current_value = last_values[-2]
+ is_greater = True
+ is_less = True
+ for window_index in range(0, window):
+ left = last_values[-2 - window_index - 1]
+ right = last_values[-1]
+ local_is_greater, local_is_less = check_if_pivot_is_greater_or_less(current_value, high_source, low_source, left, right)
+ is_greater &= local_is_greater
+ is_less &= local_is_less
+ if is_greater:
+ pivot_points_highs[index - 1] = getattr(current_value, high_source)
+ if is_less:
+ pivot_points_lows[index - 1] = getattr(current_value, low_source)
+
+ return pd.DataFrame(index=dataframe.index, data={
+ 'pivot_lows': pivot_points_lows,
+ 'pivot_highs': pivot_points_highs
+ })
+
+def check_if_pivot_is_greater_or_less(current_value, high_source: str, low_source: str, left, right) -> Tuple[bool, bool]:
+ is_greater = True
+ is_less = True
+ if (getattr(current_value, high_source) < getattr(left, high_source) or
+ getattr(current_value, high_source) < getattr(right, high_source)):
+ is_greater = False
+
+ if (getattr(current_value, low_source) > getattr(left, low_source) or
+ getattr(current_value, low_source) > getattr(right, low_source)):
+ is_less = False
+ return (is_greater, is_less)
+
+def emaKeltner(dataframe):
+ keltner = {}
+ atr = qtpylib.atr(dataframe, window=10)
+ ema20 = ta.EMA(dataframe, timeperiod=20)
+ keltner['upper'] = ema20 + atr
+ keltner['mid'] = ema20
+ keltner['lower'] = ema20 - atr
+ return keltner
+
+def chaikin_money_flow(dataframe, n=20, fillna=False) -> Series:
+ """Chaikin Money Flow (CMF)
+ It measures the amount of Money Flow Volume over a specific period.
+ http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:chaikin_money_flow_cmf
+ Args:
+ dataframe(pandas.Dataframe): dataframe containing ohlcv
+ n(int): n period.
+ fillna(bool): if True, fill nan values.
+ Returns:
+ pandas.Series: New feature generated.
+ """
+ df = dataframe.copy()
+ mfv = ((df['close'] - df['low']) - (df['high'] - df['close'])) / (df['high'] - df['low'])
+ mfv = mfv.fillna(0.0) # float division by zero
+ mfv *= df['volume']
+ cmf = (mfv.rolling(n, min_periods=0).sum()
+ / df['volume'].rolling(n, min_periods=0).sum())
+ if fillna:
+ cmf = cmf.replace([np.inf, -np.inf], np.nan).fillna(0)
+ return Series(cmf, name='cmf')
\ No newline at end of file