328 lines
15 KiB
Python
328 lines
15 KiB
Python
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
|
|
# flake8: noqa: F401
|
|
# isort: skip_file
|
|
# --- Do not remove these libs ---
|
|
import logging
|
|
import numpy as np
|
|
import pandas as pd
|
|
from pandas import DataFrame
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Union, Tuple
|
|
from freqtrade.exchange import timeframe_to_prev_date
|
|
from freqtrade.persistence import Trade
|
|
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, stoploss_from_open,
|
|
IntParameter, IStrategy, merge_informative_pair, informative, stoploss_from_absolute)
|
|
|
|
# --------------------------------
|
|
# Add your lib to import here
|
|
import talib.abstract as ta
|
|
import pandas_ta as pta
|
|
from technical import qtpylib
|
|
|
|
class FractalAtr(IStrategy):
|
|
|
|
# Strategy interface version - allow new iterations of the strategy interface.
|
|
# Check the documentation or the Sample strategy to get the latest version.
|
|
INTERFACE_VERSION = 3
|
|
|
|
# Optimal timeframe for the strategy.
|
|
timeframe = '1m'
|
|
|
|
# Can this strategy go short?
|
|
can_short: bool = False
|
|
|
|
# Minimal ROI designed for the strategy.
|
|
# This attribute will be overridden if the config file contains "minimal_roi".
|
|
minimal_roi = {
|
|
"0": 0.5
|
|
}
|
|
|
|
# Optimal stoploss designed for the strategy.
|
|
# This attribute will be overridden if the config file contains "stoploss".
|
|
stoploss = -1
|
|
|
|
position_adjustment_enable = True
|
|
use_custom_stoploss = True
|
|
|
|
# Trailing stoploss
|
|
trailing_stop = False
|
|
|
|
# Run "populate_indicators()" only for new candle.
|
|
process_only_new_candles = True
|
|
|
|
# These values can be overridden 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
|
|
|
|
# Strategy parameters
|
|
buy_rsi = IntParameter(10, 40, default=30, space="buy")
|
|
sell_rsi = IntParameter(60, 90, default=70, space="sell")
|
|
|
|
# trailing stoploss hyperopt parameters
|
|
# hard stoploss profit
|
|
|
|
pHSL = DecimalParameter(-0.200, -0.040, default=-0.99, decimals=3, space='sell', optimize=False, load=True)
|
|
# profit threshold 1, trigger point, SL_1 is used
|
|
pPF_1 = DecimalParameter(0.008, 0.020, default=0.022, decimals=3, space='sell', optimize=True, load=True)
|
|
pSL_1 = DecimalParameter(0.008, 0.020, default=0.021, decimals=3, space='sell', optimize=True, load=True)
|
|
|
|
# profit threshold 2, SL_2 is used
|
|
pPF_2 = DecimalParameter(0.040, 0.100, default=0.080, decimals=3, space='sell', optimize=True, load=True)
|
|
pSL_2 = DecimalParameter(0.020, 0.070, default=0.040, decimals=3, space='sell', optimize=True, load=True)
|
|
|
|
# Optional order type mapping.
|
|
order_types = {
|
|
'entry': 'limit',
|
|
'exit': 'limit',
|
|
'stoploss': 'market',
|
|
'stoploss_on_exchange': False
|
|
}
|
|
|
|
# Optional order time in force.
|
|
order_time_in_force = {
|
|
'entry': 'GTC',
|
|
'exit': 'GTC'
|
|
}
|
|
locked_pairs = {}
|
|
|
|
def adjust_trade_position(self, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float, min_stake: float,
|
|
max_stake: float, **kwargs):
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
|
if True | (len(dataframe) < 1):
|
|
return None
|
|
last_candle = dataframe.iloc[-1].squeeze()
|
|
filled_buys = trade.select_filled_orders('buy')
|
|
count_of_buys = len(filled_buys)
|
|
|
|
condition = (last_candle['enter_long'] == 1) # & (last_candle['close'] <= last_candle['close_1h'])
|
|
|
|
# self.protection_nb_buy_lost.value
|
|
if (0 < count_of_buys <= 2) & (current_profit < - 0.1) & (condition):
|
|
try:
|
|
stake_amount = self.config['stake_amount']
|
|
print("Adjust " + trade.pair + " " + str(current_time) + " " + str(current_profit) + " " + str(count_of_buys) + " "
|
|
+ str(stake_amount))
|
|
return stake_amount
|
|
except Exception as exception:
|
|
print(exception)
|
|
return None
|
|
return None
|
|
|
|
def calculateScore(self, last_candle):
|
|
score = 1 #- (last_candle['percent10'] / 0.01
|
|
return score
|
|
|
|
@informative('1h')
|
|
def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
self.will_frac(dataframe)
|
|
dataframe['bear'] = dataframe['high'].where(dataframe['fractal_sup'] == True, 0)
|
|
dataframe['bear'] = dataframe['bear'].replace(to_replace=0, method="ffill")
|
|
dataframe['bull'] = dataframe['low'].where(dataframe['fractal_inf'] == True, 0)
|
|
dataframe['bull'] = dataframe['bull'].replace(to_replace=0, method="ffill")
|
|
dataframe['max72'] = ta.MAX(dataframe['close'], timeperiod=72)
|
|
dataframe['min72'] = ta.MIN(dataframe['close'], timeperiod=72)
|
|
dataframe['mid72'] = (dataframe['min72'] + (dataframe['max72'] - dataframe['min72']) / 2)
|
|
dataframe['width_72'] = (dataframe['max72'] - dataframe['min72']) / dataframe['max72']
|
|
|
|
dataframe['percent'] = dataframe['close'].pct_change()
|
|
dataframe['percent5'] = dataframe['close'].pct_change(5)
|
|
return dataframe
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
self.will_frac(dataframe)
|
|
dataframe['rsi'] = ta.RSI(dataframe)
|
|
dataframe['atr'] = ta.ATR(dataframe)
|
|
dataframe['bear'] = dataframe['high'].where(dataframe['fractal_sup'] == True, 0)
|
|
dataframe['bear'] = dataframe['bear'].replace(to_replace=0, method="ffill")
|
|
dataframe['bull'] = dataframe['low'].where(dataframe['fractal_inf'] == True, 0)
|
|
dataframe['bull'] = dataframe['bull'].replace(to_replace=0, method="ffill")
|
|
dataframe["percent"] = dataframe["close"].pct_change()
|
|
dataframe["percent3"] = dataframe["close"].pct_change(3)
|
|
dataframe["percent10"] = dataframe["close"].pct_change(10)
|
|
|
|
dataframe['min200'] = ta.MIN(dataframe['close'], timeperiod=200)
|
|
dataframe['max200'] = ta.MAX(dataframe['close'], timeperiod=200)
|
|
dataframe['atr_relatif'] = ta.ATR(dataframe) / dataframe["min200"]
|
|
dataframe["close50"] = dataframe["close"].rolling(50).mean()
|
|
dataframe["close3"] = dataframe["close"].rolling(3).mean()
|
|
dataframe['rsi_mean'] = dataframe['rsi'].rolling(3).mean()
|
|
dataframe['atr_relatif_mean'] = dataframe['atr_relatif'].rolling(3).mean()
|
|
dataframe['width_200'] = (dataframe['max200'] - dataframe['min200']) / dataframe['max200']
|
|
dataframe['width_close'] = (dataframe['close'] - dataframe['min200']) / dataframe['close']
|
|
dataframe['200_close'] = dataframe['width_close'] / dataframe['width_200'] * 100
|
|
dataframe['72h_close'] = (dataframe['close'] - dataframe['min72_1h']) / dataframe['min72_1h'] * 100
|
|
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
dataframe.loc[
|
|
(
|
|
(qtpylib.crossed_above(dataframe['close'], dataframe['bull']) |
|
|
qtpylib.crossed_above(dataframe['close'].shift(1), dataframe['bull'].shift(1)) |
|
|
qtpylib.crossed_above(dataframe['close'].shift(2), dataframe['bull'].shift(2))
|
|
) &
|
|
((dataframe['rsi'] < 20) | (dataframe['atr_relatif'] > 0.008)) &
|
|
# ((dataframe['200_close'] < 25) | (dataframe['width_200'] < 0.10)) &
|
|
(dataframe['close3'] < dataframe['close50']) & # Make sure Volume is not 0
|
|
(dataframe['volume'] > 0) # Make sure Volume is not 0
|
|
|
|
), ['buy', 'enter_tag']] = (1, 'buy_fractal')
|
|
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
# dataframe.loc[
|
|
# (
|
|
# #(dataframe['volume'] > 0) # Make sure Volume is not 0
|
|
# ),
|
|
# 'exit_long'] = 1
|
|
return dataframe
|
|
|
|
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str,
|
|
current_time: datetime, entry_tag: Optional[str], **kwargs) -> bool:
|
|
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
last_candle = dataframe.iloc[-1].squeeze()
|
|
last_candle_12 = dataframe.iloc[-13].squeeze()
|
|
|
|
allow_to_buy = True #(not self.stop_all) #& (not self.all_down)
|
|
|
|
if pair in self.locked_pairs:
|
|
trade = self.locked_pairs[pair]
|
|
if (last_candle['close3'] >= last_candle['mid72_1h']):
|
|
print(pair + ' is locked ' + str(current_time) + " rate=" + str(rate) + " locked_rate=" + str(
|
|
trade.open_rate) + ' close3=' + str(last_candle['close3']) + ' mid72=' + str(last_candle['mid72_1h'])
|
|
+ str(trade))
|
|
allow_to_buy = False
|
|
# else:
|
|
# print(pair + ' unlocked ' + str(current_time) + " rate=" + str(rate) + " locked_rate=" + str(
|
|
# self.locked_pairs[pair]))
|
|
# del self.locked_pairs[pair]
|
|
|
|
if allow_to_buy:
|
|
print('Buy ' + entry_tag + ' ' + str(current_time) + ' ' + pair + " dispo=" + str(
|
|
round(self.wallets.get_available_stake_amount())) + " score=" + str(self.calculateScore(last_candle)))
|
|
return allow_to_buy
|
|
|
|
def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float,
|
|
time_in_force: str,
|
|
exit_reason: str, current_time, **kwargs, ) -> bool:
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
last_candle = dataframe.iloc[-1].squeeze()
|
|
|
|
string = ""
|
|
buy_signal = self.getTradeCandle(dataframe, trade)
|
|
if not buy_signal.empty:
|
|
buy_signal_candle = buy_signal.iloc[-1]
|
|
string = str(buy_signal_candle['date']) + ' open=' + str(buy_signal_candle['open']) \
|
|
+ " score=" + str(self.calculateScore(last_candle))
|
|
|
|
print('Sell trade ' + exit_reason + ' ' + str(current_time) + ' ' + pair + " dispo=" + str(
|
|
round(self.wallets.get_available_stake_amount())) #"+ str(amount) + ' ' + str(rate)
|
|
+ " open_rate=" + str(trade.open_rate) + " rate=" + str(rate) + " profit" + str(trade.calc_profit(rate, amount))
|
|
+ " " + string)
|
|
|
|
return True
|
|
|
|
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
|
|
current_profit: float, **kwargs) -> 'Optional[Union[str, bool]]':
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
last_candle = dataframe.iloc[-1].squeeze()
|
|
if (last_candle['percent'] > 0) | (last_candle['percent3'] > -0.002):
|
|
return None
|
|
|
|
days = (current_time - trade.open_date_utc).days
|
|
hours = (current_time - trade.open_date_utc).seconds / 3600
|
|
minutes = (current_time - trade.open_date_utc).seconds / 60
|
|
|
|
# if (days >= 1) & (current_profit < -0.02):
|
|
# return 'too_old'
|
|
|
|
self.testLockedTrade(trade, current_profit, current_rate, current_time, pair)
|
|
# if pair in self.locked_pairs:
|
|
# print(pair + " stoploss")
|
|
# return "force_stoploss"
|
|
factor = 10
|
|
exit_atr = (trade.open_rate + (last_candle['atr'] * factor))
|
|
if (current_rate > exit_atr):
|
|
return "exit_atr"
|
|
|
|
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)
|
|
last_candle = dataframe.iloc[-1].squeeze()
|
|
if (last_candle['percent'] > 0) | (last_candle['percent3'] > -0.002):
|
|
return -1
|
|
|
|
# print(pair + " " + str(current_time) + " rate=" + str(current_profit))
|
|
self.testLockedTrade(trade, current_profit, current_rate, current_time, pair)
|
|
|
|
# hard stoploss profit
|
|
HSL = self.pHSL.value
|
|
PF_1 = self.pPF_1.value
|
|
SL_1 = self.pSL_1.value
|
|
PF_2 = self.pPF_2.value
|
|
SL_2 = self.pSL_2.value
|
|
|
|
# For profits between PF_1 and PF_2 the stoploss (sl_profit) used is linearly interpolated
|
|
# between the values of SL_1 and SL_2. For all profits above PL_2 the sl_profit value
|
|
# rises linearly with current profit, for profits below PF_1 the hard stoploss profit is used.
|
|
|
|
if current_profit > PF_2:
|
|
sl_profit = SL_2 + (current_profit - PF_2)
|
|
elif (current_profit > PF_1):
|
|
sl_profit = SL_1 + ((current_profit - PF_1) * (SL_2 - SL_1) / (PF_2 - PF_1))
|
|
else:
|
|
sl_profit = HSL
|
|
|
|
return stoploss_from_open(sl_profit, current_profit)
|
|
|
|
# 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)
|
|
# candle = dataframe.iloc[-1].squeeze()
|
|
# factor = 10
|
|
# return stoploss_from_absolute(current_rate - (candle['atr'] * factor), current_rate, is_short=trade.is_short)
|
|
def getTradeCandle(self, dataframe, trade: 'Trade'):
|
|
trade_open_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
|
|
buy_signal = dataframe.loc[dataframe['date'] <= trade_open_date]
|
|
return buy_signal
|
|
|
|
def testLockedTrade(self, trade: 'Trade', current_profit, current_rate, current_time, pair):
|
|
if current_profit <= -0.1:
|
|
if pair in self.locked_pairs:
|
|
trade = self.locked_pairs[pair]
|
|
if current_rate > trade.open_rate:
|
|
print(pair + ' unlocked ' + str(current_time) + " rate=" + str(current_rate) + " locked_rate=" + str(
|
|
trade.open_rate) + ' profit=' + str(current_profit))
|
|
del self.locked_pairs[pair]
|
|
# else:
|
|
# self.locked_pairs[pair] = current_rate
|
|
else:
|
|
self.locked_pairs[pair] = trade
|
|
# self.lock_pair(pair, until=current_time + timedelta(hours=24))
|
|
|
|
print(pair + " locked at " + str(current_time) + " rate=" + str(current_rate) + " locked_rate=" + str(
|
|
trade.open_rate) + ' profit=' + str(current_profit))
|
|
|
|
def will_frac(self, df: pd.DataFrame, period: int = 2) -> Tuple[pd.Series, pd.Series]:
|
|
"""
|
|
Indicate bearish and bullish fractal patterns using shifted Series.
|
|
:param df: OHLC data
|
|
:param period: number of lower (or higher) points on each side of a high (or low)
|
|
:return: tuple of boolean Series (bearish, bullish) where True marks a fractal pattern
|
|
"""
|
|
periods = [p for p in range(-period, period + 1) if p != 0] # default [-2, -1, 1, 2]
|
|
highs = [df['high'] > df['high'].shift(p) for p in periods]
|
|
bears = pd.Series(np.logical_and.reduce(highs), index=df.index)
|
|
lows = [df['low'] < df['low'].shift(p) for p in periods]
|
|
bulls = pd.Series(np.logical_and.reduce(lows), index=df.index)
|
|
df['fractal_sup'] = bears
|
|
df['fractal_inf'] = bulls
|
|
return df |