souti/Freqtrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Synchronise

Browse Source
This commit is contained in:
Jérôme Delacotte
2025-03-09 16:53:02 +01:00
parent 9449d25384
commit 8a1c69e53a
4 changed files with 486 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

340
HammerReversalStrategy.py
View File

@@ -18,6 +18,15 @@ class HammerReversalStrategy(IStrategy):
"main_plot": {
"enter_tag": {
"color": "#197260"
},
'sma5_1d': {
'color': 'green'
},
'bb_upperband_1d': {
'color': 'blue'
},
'bb_lowerband_1d': {
'color': 'red'
}
},
"subplots": {
@@ -25,10 +34,40 @@ class HammerReversalStrategy(IStrategy):
"hammer": {
"color": "blue"
},
"inv_hammer": {
"loose_hammer": {
"color": "#c1b255"
},
"hammer_1h": {
"color": "blue"
},
"loose_hammer_1h": {
"color": "#c1b255"
},
"hammer_1d": {
"color": "blue"
},
"loose_hammer_1d": {
"color": "#c1b255"
}
},
'Percent': {
'percent3_1d': {
"color": 'pink'
},
'percent3': {
"color": 'red'
},
'percent5': {
"color": 'green'
},
'percent12': {
"color": 'blue'
},
'percent48': {
"color": 'yellow'
}
}
}
}
minimal_roi = {
@@ -51,58 +90,67 @@ class HammerReversalStrategy(IStrategy):
position_adjustment_enable = True
columns_logged = False
max_entry_position_adjustment = 20
startup_candle_count = 288
def new_adjust_trade_position(self, trade, current_time, current_rate, current_profit, min_stake, max_stake,
**kwargs) -> float:
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
count_of_buys, hours, days, first_price, last_price = self.getTradeInfos(current_time, trade)
# Initialisation des user_data (backtest compatible)
if 'dynamic_stoploss' not in trade.user_data:
trade.user_data['dynamic_stoploss'] = first_price * 0.98 # Stoploss initial à -2%
if hours < 1 or trade.stake_amount >= max_stake:
return 0
# Ajustement en cas de perte : renfort à la baisse
if (last_candle['close'] < first_price) and (last_candle['touch_support']) and (current_profit < -0.015 * count_of_buys):
additional_stake = self.config['stake_amount']
print(f"Adjust Loss - {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys}")
return max(additional_stake, 0)
# Ajustement en cas de gain : renfort à la hausse
if (last_candle['close'] > first_price) and (current_profit > 0.01 * count_of_buys):
additional_stake = self.config['stake_amount']
# Mise à jour du stoploss dynamique (on lock un profit partiel par exemple)
new_stoploss = current_rate * 0.99 # Stoploss dynamique à -1% sous le prix actuel
trade.user_data['dynamic_stoploss'] = max(trade.user_data['dynamic_stoploss'], new_stoploss)
print(f"Adjust Gain + {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys}")
return max(additional_stake, 0)
return 0
# def new_adjust_trade_position(self, trade, current_time, current_rate, current_profit, min_stake, max_stake,
# **kwargs) -> float:
# dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
# last_candle = dataframe.iloc[-1].squeeze()
#
# count_of_buys, hours, days, first_price, last_price = self.getTradeInfos(current_time, trade)
#
# # Initialisation des user_data (backtest compatible)
# if 'dynamic_stoploss' not in trade.user_data:
# trade.user_data['dynamic_stoploss'] = first_price * 0.98 # Stoploss initial à -2%
#
# if hours < 1 or trade.stake_amount >= max_stake:
# return 0
#
# # Ajustement en cas de perte : renfort à la baisse
# if (last_candle['close'] < first_price) and (last_candle['touch_support']) and (current_profit < -0.015 * count_of_buys):
# additional_stake = self.config['stake_amount']
# print(f"Adjust Loss - {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys}")
# return max(additional_stake, 0)
#
# # Ajustement en cas de gain : renfort à la hausse
# if (last_candle['close'] > first_price) and (current_profit > 0.01 * count_of_buys):
# additional_stake = self.config['stake_amount']
#
# # Mise à jour du stoploss dynamique (on lock un profit partiel par exemple)
# new_stoploss = current_rate * 0.99 # Stoploss dynamique à -1% sous le prix actuel
# trade.user_data['dynamic_stoploss'] = max(trade.user_data['dynamic_stoploss'], new_stoploss)
#
# print(f"Adjust Gain + {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys}")
# return max(additional_stake, 0)
#
# return 0
def adjust_trade_position(self, trade, current_time, current_rate, current_profit, min_stake, max_stake,
**kwargs) -> float:
if trade.has_open_orders:
return None
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
last_candle_288 = dataframe.iloc[-288].squeeze()
"""
Ajuste la position suite à un signal de sortie partielle.
"""
count_of_buys, hours, days, first_price, last_price = self.getTradeInfos(current_time, trade)
if hours < 1 or trade.stake_amount >= max_stake:
# (hours < 1) or
if (self.wallets.get_available_stake_amount() < 50): # or trade.stake_amount >= max_stake:
return 0
dispo = round(self.wallets.get_available_stake_amount())
factor = 1
if (count_of_buys > 4):
factor = count_of_buys / 4
if (count_of_buys > 1) \
and (current_profit > 0.01) \
and (last_candle['close'] < self.pairs[trade.pair]['max_touch'] * 0.99) \
and (last_candle['percent5'] < 0):
and (last_candle['haclose'] < self.pairs[trade.pair]['max_touch'] * 0.99) \
and (last_candle['percent5'] < 0) and (last_candle['percent12'] < 0):
# print(f"Adjust Sell all {current_time} rate={current_rate:.3f} stake={trade.stake_amount} count={count_of_buys} profit={profit:.1f}")
self.log_trade(
@@ -113,14 +161,22 @@ class HammerReversalStrategy(IStrategy):
pair=trade.pair,
rate=current_rate,
trade_type='Sell',
profit=round(trade.calc_profit(current_rate, trade.amount), 2), # round(current_profit * trade.stake_amount, 2),
profit=round(current_profit, 4), # round(current_profit * trade.stake_amount, 2),
buys=trade.nr_of_successful_entries,
stake=round(- trade.stake_amount, 2)
)
self.pairs[trade.pair]['last_max'] = max(last_candle['haclose'], self.pairs[trade.pair]['last_max'])
self.pairs[trade.pair]['max_touch'] = last_candle['haclose']
self.pairs[trade.pair]['last_buy'] = last_candle['haclose']
self.pairs[trade.pair]['last_sell'] = 0
return - trade.stake_amount
if (last_candle['close'] < first_price) and (last_candle['touch_support']) and (current_profit < -0.015 * count_of_buys):
additional_stake = self.calculate_stake(trade.pair, last_candle)
if (last_candle['close'] < first_price) \
and (last_candle['touch_support']) \
and ((count_of_buys <= 4 and last_candle_288['sma5_1h'] <= last_candle['sma5_1h'])
or (count_of_buys > 4 and last_candle_288['sma5_1d'] <= last_candle['sma5_1d'])) \
and (current_profit < -0.015 * count_of_buys * factor):
additional_stake = self.calculate_stake(trade.pair, last_candle, factor)
# print(f"Adjust Loss - {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys} profit={profit:.1f}")
self.log_trade(
@@ -135,11 +191,14 @@ class HammerReversalStrategy(IStrategy):
buys=trade.nr_of_successful_entries,
stake=round(additional_stake, 2)
)
self.pairs[trade.pair]['last_max'] = last_candle['haclose']
self.pairs[trade.pair]['max_touch'] = last_candle['haclose']
self.pairs[trade.pair]['last_buy'] = 0
self.pairs[trade.pair]['last_sell'] = last_candle['haclose']
return max(additional_stake, 0)
if (last_candle['close'] > first_price) and (current_profit > 0.01 * count_of_buys):
additional_stake = self.calculate_stake(trade.pair, last_candle)
additional_stake = self.calculate_stake(trade.pair, last_candle, 1)
self.log_trade(
last_candle=last_candle,
date=current_time,
@@ -152,7 +211,10 @@ class HammerReversalStrategy(IStrategy):
buys=count_of_buys,
stake=round(additional_stake, 2)
)
self.pairs[trade.pair]['last_max'] = last_candle['haclose']
self.pairs[trade.pair]['max_touch'] = last_candle['haclose']
self.pairs[trade.pair]['last_buy'] = last_candle['haclose']
self.pairs[trade.pair]['last_sell'] = 0
# print(f"Adjust Gain + {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys} profit={profit:.1f}")
return max(additional_stake, 0)
@@ -160,17 +222,17 @@ class HammerReversalStrategy(IStrategy):
use_custom_stoploss = True
def new_custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
current_profit: float, **kwargs) -> float:
if 'dynamic_stoploss' in trade.user_data:
stoploss_price = trade.user_data['dynamic_stoploss']
if current_rate < stoploss_price:
print(f"Stoploss touché ! Vente forcée {pair} à {current_rate}")
return 0.001 # on force une sortie immédiate (stop très proche)
# Sinon on reste sur le stoploss standard de la stratégie
return -1 # Exemple: 5% de perte max
# def new_custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
# current_profit: float, **kwargs) -> float:
#
# if 'dynamic_stoploss' in trade.user_data:
# stoploss_price = trade.user_data['dynamic_stoploss']
# if current_rate < stoploss_price:
# print(f"Stoploss touché ! Vente forcée {pair} à {current_rate}")
# return 0.001 # on force une sortie immédiate (stop très proche)
#
# # Sinon on reste sur le stoploss standard de la stratégie
# return -1 # Exemple: 5% de perte max
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
proposed_stake: float, min_stake: float, max_stake: float,
@@ -179,24 +241,26 @@ class HammerReversalStrategy(IStrategy):
# Obtenir les données actuelles pour cette paire
last_candle = dataframe.iloc[-1].squeeze()
return self.calculate_stake(pair, last_candle)
return self.calculate_stake(pair, last_candle, 1)
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)
# Obtenir les données actuelles pour cette paire
last_candle = dataframe.iloc[-1].squeeze()
# self.getTradeInfos(current_time, trade)
# print(f"current_profit={current_profit} mises=" + str(round(self.pairs[pair]['trade_info']['mises'], 4)))
limit_sell = (last_candle['close'] - self.pairs[trade.pair]['max_touch']) / self.pairs[trade.pair]['max_touch']
if (current_profit > 0.01) & (limit_sell < -0.01) & (last_candle['percent12'] < 0): # & (limit_sell < -0.01) & (last_candle['DI+_1h'] < 10):
sl_profit = 0.85 * current_profit # n% du profit en cours
else:
sl_profit = -1 # Hard stop-loss
stoploss = stoploss_from_open(sl_profit, current_profit)
return stoploss
# 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)
#
# # Obtenir les données actuelles pour cette paire
# last_candle = dataframe.iloc[-1].squeeze()
# # self.getTradeInfos(current_time, trade)
# # print(f"current_profit={current_profit} mises=" + str(round(self.pairs[pair]['trade_info']['mises'], 4)))
# limit_sell = (last_candle['close'] - self.pairs[trade.pair]['max_touch']) / self.pairs[trade.pair]['max_touch']
#
# if (current_profit > 0.01) & (limit_sell < -0.01) & (last_candle['percent12'] < 0): # & (limit_sell < -0.01) & (last_candle['DI+_1h'] < 10):
# sl_profit = 0.85 * current_profit # n% du profit en cours
# print(f"Stoploss {current_time} {current_rate} set to {sl_profit} / {limit_sell} / {self.pairs[trade.pair]['max_touch']}")
#
# else:
# sl_profit = -1 # Hard stop-loss
# stoploss = stoploss_from_open(sl_profit, current_profit)
# return stoploss
def custom_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
current_profit: float, **kwargs) -> 'Optional[Union[str, bool]]':
@@ -204,16 +268,20 @@ class HammerReversalStrategy(IStrategy):
Custom exit function for dynamic trade exits.
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
self.pairs[pair]['max_touch'] = max(current_rate, self.pairs[pair]['max_touch'])
last_candle = dataframe.iloc[-1].squeeze()
# Calcul de la "distance de sécurité" avant stockage dans max_touch
limit_sell = (last_candle['haclose'] - self.pairs[pair]['max_touch']) / self.pairs[pair]['max_touch']
# # Obtenir les données actuelles pour cette paire
# last_candle = dataframe.iloc[-1].squeeze()
# previous_last_candle = dataframe.iloc[-2].squeeze()
# if (last_candle['percent'] > 0) | (last_candle['percent3'] > 0.0) | (last_candle['percent5'] > 0.0):
# return None
#
# if current_profit > 0 and last_candle['inv_hammer'] > 0:
# return 'Sell_Hammer'
self.pairs[pair]['max_touch'] = max(last_candle['haclose'], self.pairs[pair]['max_touch'])
# On ne déclenche le trailing stop que si un profit mini a déjà été atteint
# and (limit_sell < -0.01)
if (current_profit > 0.01) and (last_candle['percent12'] < 0) and (last_candle['percent5'] < 0):
print(f"Custom Exit Triggered - {current_time} - Price: {current_rate:.2f} - Profit: {current_profit:.2%}")
print(f"Max touch: {self.pairs[pair]['max_touch']:.2f}, Limit sell: {limit_sell:.2%}")
return 'trailing_stop_exit'
return None
@@ -222,12 +290,15 @@ class HammerReversalStrategy(IStrategy):
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
dispo = round(self.wallets.get_available_stake_amount())
stake_amount = self.calculate_stake(pair, last_candle)
stake_amount = self.calculate_stake(pair, last_candle, 1)
self.pairs[pair]['last_max'] = max(rate, self.pairs[pair]['last_max'])
self.pairs[pair]['max_touch'] = rate
self.pairs[pair]['last_buy'] = rate
# if (self.pairs[pair]['last_sell'] > 0) and \
# (self.pairs[pair]['last_sell'] - last_candle['close']) / self.pairs[pair]['last_sell'] < 0.012:
# return False
self.pairs[pair]['last_max'] = max(last_candle['haclose'], self.pairs[pair]['last_max'])
self.pairs[pair]['max_touch'] = last_candle['haclose']
self.pairs[pair]['last_buy'] = last_candle['haclose']
self.pairs[pair]['last_sell'] = 0
#print(f"Buy {current_time} {entry_tag} rate={rate:.3f} amount={amount}")
self.log_trade(
last_candle=last_candle,
@@ -255,6 +326,7 @@ class HammerReversalStrategy(IStrategy):
# self.pairs[pair]['last_max'] = 0
# self.pairs[pair]['max_touch'] = 0
self.pairs[pair]['last_buy'] = 0
self.pairs[pair]['last_sell'] = rate
self.log_trade(
last_candle=last_candle,
date=current_time,
@@ -270,10 +342,16 @@ class HammerReversalStrategy(IStrategy):
return ok
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
heikinashi = qtpylib.heikinashi(dataframe)
dataframe['haopen'] = heikinashi['open']
dataframe['haclose'] = heikinashi['close']
dataframe['hapercent'] = dataframe['haclose'].pct_change()
dataframe['hammer'] = ta.CDLHAMMER(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'])
dataframe['inv_hammer'] = ta.CDLINVERTEDHAMMER(dataframe['open'], dataframe['high'], dataframe['low'],
dataframe['close'])
# Volume
dataframe['volume_mean'] = ta.SMA(dataframe['volume'], timeperiod=20)
dataframe['volume_above_avg'] = dataframe['volume'] > 1.2 * dataframe['volume_mean']
@@ -320,22 +398,41 @@ class HammerReversalStrategy(IStrategy):
dataframe["percent3"] = dataframe['close'].pct_change(3)
dataframe["percent5"] = dataframe['close'].pct_change(5)
dataframe["percent12"] = dataframe['close'].pct_change(12)
dataframe["percent48"] = dataframe['close'].pct_change(48)
dataframe = self.pattern_hammer(dataframe)
dataframe = self.detect_hammer_with_context(dataframe)
dataframe = self.detect_loose_hammer(dataframe)
#dataframe = self.detect_squeeze_pump(dataframe)
# ======================================================================================
################### INFORMATIVE 1h
informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe="1h")
# informative['hammer'] = ta.CDLHAMMER(informative['open'], informative['high'], informative['low'], informative['close'])
informative = self.detect_loose_hammer(informative)
informative = self.detect_hammer_with_context(informative)
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(informative), window=20, stds=2)
informative['bb_lowerband'] = bollinger['lower']
informative['bb_middleband'] = bollinger['mid']
informative['bb_upperband'] = bollinger['upper']
informative['sma5'] = ta.SMA(informative, timeperiod=5)
informative["percent3"] = informative['close'].pct_change(3)
dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1h", ffill=True)
# ======================================================================================
################### INFORMATIVE 1d
informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe="1d")
# informative['hammer'] = ta.CDLHAMMER(informative['open'], informative['high'], informative['low'], informative['close'])
informative = self.detect_loose_hammer(informative)
informative['max7'] = ta.MAX(informative['close'], timeperiod=7)
informative = self.detect_hammer_with_context(informative)
informative['sma5'] = ta.SMA(informative, timeperiod=5)
informative["percent3"] = informative['sma5'].pct_change(3)
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(informative), window=20, stds=2)
informative['bb_lowerband'] = bollinger['lower']
informative['bb_middleband'] = bollinger['mid']
informative['bb_upperband'] = bollinger['upper']
informative['bb_upperband'], informative['bb_middleband'], informative['bb_lowerband'] = ta.BBANDS(
informative['close'], timeperiod=20
)
dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1d", ffill=True)
dataframe['hammer_marker'] = np.where(dataframe['hammer_signal'], dataframe['low'] * 0.99, np.nan)
@@ -355,7 +452,7 @@ class HammerReversalStrategy(IStrategy):
, # Divergence possible
['enter_long', 'enter_tag']] = [1, 'buy_hammer']
# dataframe.loc[
# d ataframe.loc[
# (dataframe['hammer2'] > 0)
# # & (dataframe['close'] < dataframe['bb_middleband'])
# # (dataframe['volume_above_avg']) &
@@ -367,7 +464,8 @@ class HammerReversalStrategy(IStrategy):
# , # Divergence possible
# ['enter_long', 'enter_tag']] = [1, 'buy_hammer2']
dataframe.loc[
(dataframe['loose_hammer'] > 0)
(dataframe['percent3'] < - 0.005)
& (dataframe['percent48'] < 0.02)
, # Divergence possible
['enter_long', 'enter_tag']] = [1, 'buy_loose_hammer']
@@ -585,17 +683,61 @@ class HammerReversalStrategy(IStrategy):
sma5 = str(sma5_1d) + ' ' + str(sma5_1h)
first_rate = self.pairs[pair]['last_max']
if action != 'Sell':
profit = round((last_candle['close'] - self.pairs[pair]['last_max']) / self.pairs[pair]['last_max'], 2)
# if action != 'Sell':
# profit = round((last_candle['close'] - self.pairs[pair]['last_max']) / self.pairs[pair]['last_max'], 2)
limit_sell = rsi_pct # round((last_candle['close'] - self.pairs[pair]['last_max']) / self.pairs[pair]['last_max'], 4)
max7_1d = last_candle['max7_1d'] #round(100 * (last_candle['close'] - self.pairs[pair]['last_max']) / self.pairs[pair]['last_max'], 1)
max7_1d = round(self.pairs[pair]['max_touch'], 1) #last_candle['max7_1d'] #round(100 * (last_candle['close'] - self.pairs[pair]['last_max']) / self.pairs[pair]['last_max'], 1)
pct_max = round(100 * (last_candle['close'] - max7_1d) / max7_1d, 1)
print(
f"| {date:<16} | {action:<10} | {pair:<10} | {trade_type or '-':<18} | {rate or '-':>12} | {dispo or '-':>6} | {profit or '-':>8} | {pct_max or '-':>5} | {max7_1d or '-':>11} | {self.pairs[pair]['max_touch'] or '-':>12} | {self.pairs[pair]['last_max'] or '-':>12} | {buys or '-':>5} | {stake or '-':>10} |"
f"| {date:<16} | {action:<10} | {pair:<10} | {trade_type or '-':<18} | {rate or '-':>12} | {dispo or '-':>6} | {profit or '-':>8} | {pct_max or '-':>5} | {max7_1d or '-':>11} | {round(self.pairs[pair]['max_touch'], 2) or '-':>12} | {round(self.pairs[pair]['last_max'],2) or '-':>12} | {buys or '-':>5} | {stake or '-':>10} |"
)
def calculate_stake(self, pair, last_candle):
factor = 1 - 2 * (last_candle['close'] - last_candle['max7_1d']) / last_candle['max7_1d']
def calculate_stake(self, pair, last_candle, factor=1):
amount = self.config['stake_amount'] * factor #1000 / self.first_stack_factor.value self.protection_stake_amount.value #
return amount
@property
def protections(self):
return [
{
"method": "CooldownPeriod",
"stop_duration_candles": 12
}
# {
# "method": "MaxDrawdown",
# "lookback_period_candles": self.lookback.value,
# "trade_limit": self.trade_limit.value,
# "stop_duration_candles": self.protection_stop.value,
# "max_allowed_drawdown": self.protection_max_allowed_dd.value,
# "only_per_pair": False
# },
# {
# "method": "StoplossGuard",
# "lookback_period_candles": 24,
# "trade_limit": 4,
# "stop_duration_candles": self.protection_stoploss_stop.value,
# "only_per_pair": False
# },
# {
# "method": "StoplossGuard",
# "lookback_period_candles": 24,
# "trade_limit": 4,
# "stop_duration_candles": 2,
# "only_per_pair": False
# },
# {
# "method": "LowProfitPairs",
# "lookback_period_candles": 6,
# "trade_limit": 2,
# "stop_duration_candles": 60,
# "required_profit": 0.02
# },
# {
# "method": "LowProfitPairs",
# "lookback_period_candles": 24,
# "trade_limit": 4,
# "stop_duration_candles": 2,
# "required_profit": 0.01
# }
]