commit c91af5dfb8cba57ac83d2806123f80dfd0d77da5 Author: mzaxd Date: Sat Nov 30 22:18:11 2024 +0800 上传文件至 Strategies diff --git a/Strategies/HarmonicDivergenceFv3.py b/Strategies/HarmonicDivergenceFv3.py new file mode 100644 index 0000000..93a9079 --- /dev/null +++ b/Strategies/HarmonicDivergenceFv3.py @@ -0,0 +1,775 @@ +# 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 HarmonicDivergenceFv3(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, + "20": 0.25, + "15": 0.15, + "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.1 + trailing_stop_positive_offset = 0.11 # 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') + + HarmonicDivergenceFv3.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