773 lines
32 KiB
Python
773 lines
32 KiB
Python
# 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() + '<br>'
|
||
|
||
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() + '<br>'
|
||
|
||
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') |