602 lines
27 KiB
Python
602 lines
27 KiB
Python
from datetime import timedelta, datetime
|
|
from freqtrade.strategy.interface import IStrategy
|
|
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, stoploss_from_open,
|
|
IntParameter, IStrategy, merge_informative_pair, informative, stoploss_from_absolute)
|
|
from pandas import DataFrame
|
|
from freqtrade.persistence import Trade
|
|
from sklearn.tree import DecisionTreeClassifier
|
|
from sklearn.preprocessing import StandardScaler
|
|
import numpy as np
|
|
import talib.abstract as ta
|
|
import pandas_ta as pdta
|
|
import freqtrade.vendor.qtpylib.indicators as qtpylib
|
|
from typing import Optional, Union, Tuple
|
|
|
|
|
|
class HammerReversalStrategy(IStrategy):
|
|
plot_config = {
|
|
"main_plot": {
|
|
"enter_tag": {
|
|
"color": "#197260"
|
|
}
|
|
},
|
|
"subplots": {
|
|
"Hammer": {
|
|
"hammer": {
|
|
"color": "blue"
|
|
},
|
|
"inv_hammer": {
|
|
"color": "#c1b255"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
minimal_roi = {
|
|
"0": 5
|
|
}
|
|
# Regrouper toutes les informations dans un seul dictionnaire
|
|
pairs = {
|
|
pair: {
|
|
"last_max": 0,
|
|
"trade_info": {},
|
|
"max_touch": 0.0,
|
|
"last_sell": 0.0,
|
|
"last_buy": 0.0
|
|
}
|
|
for pair in ["BTC/USDT", "ETH/USDT", "DOGE/USDT", "DASH/USDT", "XRP/USDT", "SOL/USDT"]
|
|
}
|
|
|
|
stoploss = -1
|
|
timeframe = '1h'
|
|
position_adjustment_enable = True
|
|
columns_logged = False
|
|
max_entry_position_adjustment = 20
|
|
|
|
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:
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
|
last_candle = dataframe.iloc[-1].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:
|
|
return 0
|
|
|
|
dispo = round(self.wallets.get_available_stake_amount())
|
|
|
|
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):
|
|
# 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(
|
|
last_candle=last_candle,
|
|
date=current_time,
|
|
action="Sell All",
|
|
dispo=dispo,
|
|
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),
|
|
buys=trade.nr_of_successful_entries,
|
|
stake=round(- trade.stake_amount, 2)
|
|
)
|
|
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)
|
|
|
|
# print(f"Adjust Loss - {current_time} rate={current_rate:.3f} stake={additional_stake} count={count_of_buys} profit={profit:.1f}")
|
|
self.log_trade(
|
|
last_candle=last_candle,
|
|
date=current_time,
|
|
action="Loss -",
|
|
dispo=dispo,
|
|
pair=trade.pair,
|
|
rate=current_rate,
|
|
trade_type='Decrease',
|
|
profit=round(current_profit, 4), # round(current_profit * trade.stake_amount, 2),
|
|
buys=trade.nr_of_successful_entries,
|
|
stake=round(additional_stake, 2)
|
|
)
|
|
|
|
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)
|
|
self.log_trade(
|
|
last_candle=last_candle,
|
|
date=current_time,
|
|
dispo=dispo,
|
|
action="Gain +",
|
|
rate=current_rate,
|
|
pair=trade.pair,
|
|
trade_type='Increase',
|
|
profit=round(current_profit, 2),
|
|
buys=count_of_buys,
|
|
stake=round(additional_stake, 2)
|
|
)
|
|
|
|
# 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)
|
|
|
|
return 0
|
|
|
|
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 custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
|
|
proposed_stake: float, min_stake: float, max_stake: 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()
|
|
return self.calculate_stake(pair, last_candle)
|
|
|
|
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_exit(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
|
|
current_profit: float, **kwargs) -> 'Optional[Union[str, bool]]':
|
|
"""
|
|
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'])
|
|
|
|
# # 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'
|
|
|
|
return None
|
|
|
|
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()
|
|
dispo = round(self.wallets.get_available_stake_amount())
|
|
stake_amount = self.calculate_stake(pair, last_candle)
|
|
|
|
self.pairs[pair]['last_max'] = max(rate, self.pairs[pair]['last_max'])
|
|
self.pairs[pair]['max_touch'] = rate
|
|
self.pairs[pair]['last_buy'] = rate
|
|
|
|
#print(f"Buy {current_time} {entry_tag} rate={rate:.3f} amount={amount}")
|
|
self.log_trade(
|
|
last_candle=last_candle,
|
|
date=current_time,
|
|
action="START BUY",
|
|
pair=pair,
|
|
rate=rate,
|
|
dispo=dispo,
|
|
profit=0,
|
|
stake=round(stake_amount, 2)
|
|
)
|
|
return True
|
|
|
|
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()
|
|
dispo = round(self.wallets.get_available_stake_amount())
|
|
|
|
allow_to_sell = (last_candle['percent5'] < -0.00)
|
|
ok = (allow_to_sell) | (exit_reason == 'force_exit')
|
|
if ok:
|
|
# self.pairs[pair]['last_max'] = 0
|
|
# self.pairs[pair]['max_touch'] = 0
|
|
self.pairs[pair]['last_buy'] = 0
|
|
self.log_trade(
|
|
last_candle=last_candle,
|
|
date=current_time,
|
|
action="Sell",
|
|
pair=pair,
|
|
trade_type=exit_reason,
|
|
rate=last_candle['close'],
|
|
dispo=dispo,
|
|
profit=round(trade.calc_profit(rate, amount), 2)
|
|
)
|
|
#print(f"Sell {current_time} {exit_reason} rate={rate:.3f} amount={amount} profit={amount * rate:.3f}")
|
|
|
|
return ok
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
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']
|
|
|
|
# RSI
|
|
dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14)
|
|
dataframe['rsi_low'] = dataframe['rsi'] < 30
|
|
dataframe['rsi_high'] = dataframe['rsi'] > 70
|
|
|
|
# Support / Résistance
|
|
dataframe['lowest_20'] = dataframe['low'].rolling(window=20).min()
|
|
dataframe['highest_20'] = dataframe['high'].rolling(window=20).max()
|
|
dataframe['touch_support'] = dataframe['low'] <= dataframe['lowest_20']
|
|
dataframe['touch_resistance'] = dataframe['high'] >= dataframe['highest_20']
|
|
|
|
# MACD
|
|
macd = pdta.macd(dataframe['close'])
|
|
# dataframe['macd'] = macd['macd']
|
|
# dataframe['macdsignal'] = macd['macdsignal']
|
|
# dataframe['macdhist'] = macd['macdhist']
|
|
dataframe['macd'] = macd['MACD_12_26_9']
|
|
dataframe['macdsignal'] = macd['MACDs_12_26_9']
|
|
dataframe['macdhist'] = macd['MACDh_12_26_9']
|
|
|
|
# Bollinger Bands
|
|
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
|
|
dataframe['bb_lowerband'] = bollinger['lower']
|
|
dataframe['bb_middleband'] = bollinger['mid']
|
|
dataframe['bb_upperband'] = bollinger['upper']
|
|
|
|
dataframe['touch_bb_lower'] = dataframe['low'] <= dataframe['bb_lowerband']
|
|
|
|
# ADX (Trend Force)
|
|
dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14)
|
|
# ATR
|
|
dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14)
|
|
|
|
# Ratio mèche/corps (manche)
|
|
dataframe['candle_length'] = dataframe['high'] - dataframe['low']
|
|
dataframe['candle_body'] = abs(dataframe['close'] - dataframe['open'])
|
|
dataframe['wick_ratio'] = dataframe['candle_length'] / dataframe['candle_body']
|
|
|
|
dataframe["percent"] = dataframe['close'].pct_change()
|
|
dataframe["percent3"] = dataframe['close'].pct_change(3)
|
|
dataframe["percent5"] = dataframe['close'].pct_change(5)
|
|
dataframe["percent12"] = dataframe['close'].pct_change(12)
|
|
|
|
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 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['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)
|
|
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
dataframe.loc[
|
|
(dataframe['hammer'] > 0) & False
|
|
# & (dataframe['close'] < dataframe['bb_middleband'])
|
|
# & (dataframe['volume_above_avg'])
|
|
# & (dataframe['rsi_low'])
|
|
# & (dataframe['touch_support'] | dataframe['touch_bb_lower']) # Support ou BB inférieure
|
|
# & (dataframe['wick_ratio'] > 2) # Manche >= 2x corps
|
|
# (dataframe['adx'] < 30) & # Éviter les tendances trop fortes
|
|
# (dataframe['macd'] > dataframe['macdsignal'])
|
|
, # Divergence possible
|
|
['enter_long', 'enter_tag']] = [1, 'buy_hammer']
|
|
|
|
# dataframe.loc[
|
|
# (dataframe['hammer2'] > 0)
|
|
# # & (dataframe['close'] < dataframe['bb_middleband'])
|
|
# # (dataframe['volume_above_avg']) &
|
|
# # (dataframe['rsi_low']) &
|
|
# # & (dataframe['touch_support'] | dataframe['touch_bb_lower']) # Support ou BB inférieure
|
|
# # (dataframe['wick_ratio'] > 2) & # Manche >= 2x corps
|
|
# # (dataframe['adx'] < 30) & # Éviter les tendances trop fortes
|
|
# # (dataframe['macd'] > dataframe['macdsignal'])
|
|
# , # Divergence possible
|
|
# ['enter_long', 'enter_tag']] = [1, 'buy_hammer2']
|
|
dataframe.loc[
|
|
(dataframe['loose_hammer'] > 0)
|
|
, # Divergence possible
|
|
['enter_long', 'enter_tag']] = [1, 'buy_loose_hammer']
|
|
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
# dataframe.loc[
|
|
# (dataframe['inv_hammer'] > 0)
|
|
# # (dataframe['volume_above_avg']) &
|
|
# # (dataframe['rsi_high']) &
|
|
# # (dataframe['touch_resistance'] | (dataframe['high'] >= dataframe['bb_upperband'])) &
|
|
# # (dataframe['wick_ratio'] > 2) &
|
|
# # (dataframe['adx'] < 30) &
|
|
# # (dataframe['macd'] < dataframe['macdsignal'])
|
|
# ,
|
|
# ['exit_long', 'exit_tag']] = [1, 'sell_hammer']
|
|
return dataframe
|
|
|
|
def getTradeInfos(self, current_time, trade):
|
|
filled_buys = trade.select_filled_orders('buy')
|
|
count_of_buys = len(filled_buys)
|
|
first_price = filled_buys[0].price
|
|
days = 0
|
|
minutes = 0
|
|
hours = 0
|
|
last_price = first_price
|
|
mises=0
|
|
for buy in filled_buys:
|
|
minutes = (current_time - buy.order_date_utc).seconds / 60
|
|
hours = round(minutes / 60, 0)
|
|
days = (current_time - buy.order_date_utc).days
|
|
last_price = buy.price
|
|
mises += buy.amount * buy.price
|
|
# self.pairs[trade.pair]['trade_info'] = {
|
|
# "count_of_buys": count_of_buys,
|
|
# "hours": hours,
|
|
# "days": days,
|
|
# "minutes": minutes,
|
|
# "first_price": first_price,
|
|
# "last_price": last_price,
|
|
# "mises": mises
|
|
# }
|
|
return count_of_buys, hours, days, first_price, last_price
|
|
|
|
def informative_pairs(self):
|
|
# get access to all pairs available in whitelist.
|
|
pairs = self.dp.current_whitelist()
|
|
informative_pairs = [(pair, '1d') for pair in pairs]
|
|
informative_pairs += [(pair, '1h') for pair in pairs]
|
|
|
|
return informative_pairs
|
|
|
|
# def pattern_hammer(self, df: DataFrame) -> DataFrame:
|
|
# """
|
|
# Expected df contains Open, High, Low, Close,
|
|
# """
|
|
# # Compute percentile
|
|
# for level in [50, 90]:
|
|
# df[f'{level}_percentile'] = df[['high', 'low']].apply(lambda x: np.percentile(x, q=level),
|
|
# axis=1)
|
|
#
|
|
# condition = ((df['open'].values >= df[
|
|
# '50_percentile'].values) # open larger then 50 percentile, i.e. at the upper half
|
|
# & (df['close'].values >= df['90_percentile'].values) # close larger then 90 percentile, i.e. at the top of candlestick
|
|
# & (df['close'].values >= df['open'].values) # bullish candlestick
|
|
# )
|
|
#
|
|
# df['hammer2'] = np.where(condition, 1, 0)
|
|
# return df
|
|
|
|
def pattern_hammer(self, df: DataFrame) -> DataFrame:
|
|
lower_shadow = np.minimum(df['open'], df['close']) - df['low']
|
|
upper_shadow = df['high'] - np.maximum(df['open'], df['close'])
|
|
body = abs(df['close'] - df['open'])
|
|
|
|
df['hammer2'] = (
|
|
(lower_shadow > 2 * body) & # Longue mèche basse
|
|
(upper_shadow < body) & # Faible mèche haute
|
|
(df['close'] > df['open']) & # Bougie verte
|
|
((df['close'] - df['low']) / (df['high'] - df['low']) > 0.75) # Clôture dans le quart supérieur
|
|
).astype(int)
|
|
|
|
return df
|
|
|
|
def detect_hammer_with_context(self, df: DataFrame) -> DataFrame:
|
|
"""
|
|
Détection d'un marteau validé par :
|
|
- Structure de la bougie (marteau classique)
|
|
- Volume anormalement haut (signale l'intérêt du marché)
|
|
- Divergence RSI (momentum qui se retourne)
|
|
|
|
"""
|
|
|
|
# === Détection du marteau ===
|
|
lower_shadow = np.minimum(df['open'], df['close']) - df['low']
|
|
upper_shadow = df['high'] - np.maximum(df['open'], df['close'])
|
|
body = abs(df['close'] - df['open'])
|
|
|
|
df['hammer'] = (
|
|
(lower_shadow > 2 * body) & # Longue mèche basse
|
|
(upper_shadow < body) & # Faible mèche haute
|
|
(df['close'] > df['open']) & # Bougie verte
|
|
((df['close'] - df['low']) / (df['high'] - df['low']) > 0.75) # Clôture en haut de la bougie
|
|
).astype(int)
|
|
|
|
# === Filtre sur le volume ===
|
|
df['volume_mean'] = df['volume'].rolling(window=20).mean()
|
|
df['high_volume'] = df['volume'] > 1.5 * df['volume_mean']
|
|
|
|
# === RSI pour la divergence ===
|
|
df['rsi'] = ta.RSI(df['close'], timeperiod=14)
|
|
|
|
df['rsi_lowest'] = df['rsi'].rolling(window=5).min() # Cherche un creux récent de RSI
|
|
df['price_lowest'] = df['close'].rolling(window=5).min()
|
|
|
|
# Divergence haussière = prix fait un nouveau plus bas, mais RSI remonte
|
|
df['bullish_divergence'] = (
|
|
(df['low'] < df['low'].shift(1)) &
|
|
(df['rsi'] > df['rsi'].shift(1)) &
|
|
(df['rsi'] < 30) # Survendu
|
|
)
|
|
|
|
# === Condition finale : marteau + contexte favorable ===
|
|
df['hammer_signal'] = (
|
|
df['hammer'] &
|
|
df['high_volume'] &
|
|
df['bullish_divergence']
|
|
).astype(int)
|
|
|
|
return df
|
|
|
|
def detect_loose_hammer(self, df: DataFrame) -> DataFrame:
|
|
"""
|
|
Détection large de marteaux : accepte des corps plus gros, ne vérifie pas le volume,
|
|
ne demande pas de divergence, juste un pattern visuel simple.
|
|
"""
|
|
|
|
body = abs(df['close'] - df['open'])
|
|
upper_shadow = abs(df['high'] - np.maximum(df['close'], df['open']))
|
|
lower_shadow = abs(np.minimum(df['close'], df['open']) - df['low'])
|
|
|
|
# Critères simplifiés :
|
|
df['loose_hammer'] = (
|
|
(lower_shadow > body * 2.5) & # mèche basse > 1.5x corps
|
|
(upper_shadow < body) # petite mèche haute
|
|
# (df['close'] > df['open']) # bougie verte (optionnel, on peut prendre aussi les rouges)
|
|
).astype(int)
|
|
df['won_hammer'] = (
|
|
(upper_shadow > body * 2.5) & # mèche basse > 1.5x corps
|
|
(lower_shadow < body) # petite mèche haute
|
|
# (df['close'] > df['open']) # bougie verte (optionnel, on peut prendre aussi les rouges)
|
|
).astype(int)
|
|
|
|
return df
|
|
|
|
def detect_squeeze_pump(self, dataframe: DataFrame) -> DataFrame:
|
|
"""
|
|
Détecte un pump vertical violent, pour éviter d'acheter dans une phase de distribution ultra risquée.
|
|
"""
|
|
# Ratio volume par rapport à la moyenne mobile
|
|
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume'].rolling(20).mean()
|
|
|
|
dataframe['bb_upper_dist'] = (dataframe['close'] - dataframe['bb_upperband']) / dataframe['bb_upperband']
|
|
|
|
# Bougie ultra verticale
|
|
dataframe['candle_pct_change'] = (dataframe['close'] - dataframe['open']) / dataframe['open']
|
|
|
|
# ATR pour détecter la volatilité excessive
|
|
dataframe['atr_ratio'] = dataframe['atr'] / dataframe['close']
|
|
|
|
# Condition de détection (à ajuster selon la pair et le marché)
|
|
dataframe['squeeze_alert'] = (
|
|
(dataframe['volume_ratio'] > 5) & # volume X5 ou plus
|
|
(dataframe['candle_pct_change'] > 0.05) & # bougie verte de +5% ou plus
|
|
(dataframe['bb_upper_dist'] > 0.03) # ferme largement au-dessus de la BB supérieure
|
|
)
|
|
|
|
return dataframe
|
|
|
|
def log_trade(self, action, pair, date, trade_type=None, rate=None, dispo=None, profit=None, buys=None, stake=None, last_candle=None):
|
|
# Afficher les colonnes une seule fois
|
|
if self.config.get('runmode') == 'hyperopt':
|
|
return
|
|
if not self.columns_logged:
|
|
print(
|
|
f"| {'Date':<16} | {'Action':<10} | {'Pair':<10} | {'Trade Type':<18} | {'Rate':>12} | {'Dispo':>6} | {'Profit':>8} | {'Pct':>5} | {'max7_1d':>11} | {'max_touch':>12} | {'last_max':>12} | {'Buys':>5} | {'Stake':>10} |"
|
|
)
|
|
print(
|
|
f"|{'-' * 18}|{'-' * 12}|{'-' * 12}|{'-' * 20}|{'-' * 14}|{'-' * 8}|{'-' * 10}|{'-' * 7}|{'-' * 13}|{'-' * 14}|{'-' * 14}|{'-' * 7}|{'-' * 12}|"
|
|
)
|
|
self.columns_logged = True
|
|
date = str(date)[:16] if date else "-"
|
|
limit = None
|
|
# if buys is not None:
|
|
# limit = round(last_rate * (1 - self.fibo[buys] / 100), 4)
|
|
|
|
rsi = ''
|
|
rsi_pct = ''
|
|
# if last_candle is not None:
|
|
# if (not np.isnan(last_candle['rsi_1d'])) and (not np.isnan(last_candle['rsi_1h'])):
|
|
# rsi = str(int(last_candle['rsi_1d'])) + " " + str(int(last_candle['rsi_1h']))
|
|
# if (not np.isnan(last_candle['rsi_pct_1d'])) and (not np.isnan(last_candle['rsi_pct_1h'])):
|
|
# rsi_pct = str(int(10000 * last_candle['bb_mid_pct_1d'])) + " " + str(
|
|
# int(last_candle['rsi_pct_1d'])) + " " + str(int(last_candle['rsi_pct_1h']))
|
|
|
|
# first_rate = self.percent_threshold.value
|
|
# last_rate = self.threshold.value
|
|
# action = self.color_line(action, action)
|
|
sma5_1d = ''
|
|
sma5_1h = ''
|
|
# if last_candle['sma5_pct_1d'] is not None:
|
|
# sma5_1d = round(last_candle['sma5_pct_1d'] * 100, 2)
|
|
# if last_candle['sma5_pct_1h'] is not None:
|
|
# sma5_1h = round(last_candle['sma5_pct_1h'] * 100, 2)
|
|
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)
|
|
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)
|
|
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} |"
|
|
)
|
|
|
|
def calculate_stake(self, pair, last_candle):
|
|
factor = 1 - 2 * (last_candle['close'] - last_candle['max7_1d']) / last_candle['max7_1d']
|
|
|
|
amount = self.config['stake_amount'] * factor #1000 / self.first_stack_factor.value self.protection_stake_amount.value #
|
|
return amount
|