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" }, 'sma5_1d': { 'color': 'green' }, 'bb_upperband_1d': { 'color': 'blue' }, 'bb_lowerband_1d': { 'color': 'red' } }, "subplots": { "Hammer": { "hammer": { "color": "blue" }, "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 = { "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 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 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) # (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['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( last_candle=last_candle, date=current_time, action="Sell All", dispo=dispo, pair=trade.pair, rate=current_rate, trade_type='Sell', 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 ((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( 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) ) 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, 1) 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) ) 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) 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, 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 # 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]]': """ Custom exit function for dynamic trade exits. """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) 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'] 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 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, 1) # 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, 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.pairs[pair]['last_sell'] = rate 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: 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'] # 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["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 = 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'] 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'] # d ataframe.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['percent3'] < - 0.005) & (dataframe['percent48'] < 0.02) , # 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 = 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} | {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): 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 # } ]