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

Ajout test_signal_success + changement calcul montant

Browse Source
This commit is contained in:
Jérôme Delacotte
2025-05-06 21:09:24 +02:00
parent 38c36fddaa
commit cbcd88c48b
1 changed files with 72 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
Zeus_8_3_2_B_4_2.py
View File

@@ -90,8 +90,11 @@ class Zeus_8_3_2_B_4_2(IStrategy):
"sma10": {
"color": "blue"
},
"sma5_1h": {
"min12_1d": {
"color": "red"
},
"max12_1d": {
"color": 'red'
}
},
@@ -174,7 +177,8 @@ class Zeus_8_3_2_B_4_2(IStrategy):
"last_trade": None,
"last_count_of_buys": 0,
'base_stake_amount': 0,
'stop_buy': False
'stop_buy': False,
'last_date': 0
}
for pair in ["BTC/USDC", "ETH/USDC", "DOGE/USDC", "XRP/USDC", "SOL/USDC",
"BTC/USDT", "ETH/USDT", "DOGE/USDT", "XRP/USDT", "SOL/USDT"]
@@ -234,10 +238,14 @@ class Zeus_8_3_2_B_4_2(IStrategy):
stake_amount = self.adjust_stake_amount(pair, last_candle)
minutes = 0
if self.pairs[pair]['last_date'] != 0:
minutes = (current_time - self.pairs[pair]['last_date']).total_seconds() / 60.0
self.log_trade(
last_candle=last_candle,
date=current_time,
action="Buy" if allow_to_buy else "Canceled",
action=("Buy" if allow_to_buy else "Canceled") + " " + str(minutes),
pair=pair,
rate=rate,
dispo=dispo,
@@ -258,6 +266,8 @@ class Zeus_8_3_2_B_4_2(IStrategy):
allow_to_sell = (last_candle['percent'] < 0)
minutes = (current_time - trade.date_last_filled_utc).total_seconds() / 60.0
if allow_to_sell:
self.trades = list()
self.pairs[pair]['last_count_of_buys'] = trade.nr_of_successful_entries #self.pairs[pair]['count_of_buys']
@@ -270,7 +280,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
self.log_trade(
last_candle=last_candle,
date=current_time,
action="Sell",
action="Sell " + str(minutes),
pair=pair,
trade_type=exit_reason,
rate=last_candle['close'],
@@ -279,6 +289,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
)
self.pairs[pair]['max_touch'] = 0
self.pairs[pair]['last_buy'] = 0
self.pairs[pair]['last_date'] = current_time
return (allow_to_sell) | (exit_reason == 'force_exit')
@@ -541,20 +552,20 @@ class Zeus_8_3_2_B_4_2(IStrategy):
dataframe['average_line_288'] = talib.MIDPOINT(dataframe['close'], timeperiod=288)
dataframe['average_line_288_098'] = dataframe['average_line_288'] * 0.98
dataframe['average_line_288_099'] = dataframe['average_line_288'] * 0.99
# Sort the close prices to find the 4 lowest values
sorted_close_prices = dataframe['close'].tail(576).sort_values()
lowest_4 = sorted_close_prices.head(20)
dataframe['lowest_4_average'] = lowest_4.mean()
# Propagate this mean value across the entire dataframe
# dataframe['lowest_4_average'] = dataframe['lowest_4_average'].iloc[0]
# # Sort the close prices to find the 4 highest values
sorted_close_prices = dataframe['close'].tail(288).sort_values(ascending=False)
highest_4 = sorted_close_prices.head(20)
# # Calculate the mean of the 4 highest values
dataframe['highest_4_average'] = highest_4.mean()
# # Sort the close prices to find the 4 lowest values
# sorted_close_prices = dataframe['close'].rolling(576).sort_values()
# lowest_4 = sorted_close_prices.head(20)
#
# dataframe['lowest_4_average'] = lowest_4.mean()
# # Propagate this mean value across the entire dataframe
# # dataframe['lowest_4_average'] = dataframe['lowest_4_average'].iloc[0]
#
# # # Sort the close prices to find the 4 highest values
# sorted_close_prices = dataframe['close'].rolling(288).sort_values(ascending=False)
# highest_4 = sorted_close_prices.head(20)
#
# # # Calculate the mean of the 4 highest values
# dataframe['highest_4_average'] = highest_4.mean()
# Compter les baisses consécutives
self.calculateDownAndUp(dataframe, limit=0.0001)
@@ -581,14 +592,24 @@ class Zeus_8_3_2_B_4_2(IStrategy):
informative['rsi_diff_2'] = informative['rsi_diff'].diff()
informative['sma5'] = talib.SMA(informative, timeperiod=5)
informative['sma5_diff'] = 100 * informative['sma5'].diff() / informative['sma5']
informative['sma5_pct'] = 100 * (informative['sma5'] - informative['sma5'].shift(1)) / informative['sma5']
informative['sma5_diff_sum'] = (informative['sma5_pct'].rolling(5).sum()) / 5
informative['sma5_diff2_sum'] = informative['sma5_diff_sum'].diff()
self.calculateDownAndUp(informative, limit=0.0012)
if self.dp.runmode.value in ('backtest'):
self.test_signal_success(informative, percent=0.01, window_size=24)
# if self.dp.runmode.value in ('backtest'):
# self.test_signal_success(informative, percent=0.01, window_size=24)
# if self.dp.runmode.value in ('backtest'):
# condition = (informative['sma5'].shift(2) > informative['sma5'].shift(1)) \
# & (informative['sma5'].shift(1) < informative['sma5']) \
# & (informative['down_pct'].shift(3) < -0.015)
#
# self.test_signal_success(informative, condition, percent=0.01, window_size=3)
# self.test_signal_success(informative, condition, percent=0.01, window_size=5)
# self.test_signal_success(informative, condition, percent=0.01, window_size=10)
dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1h", ffill=True)
@@ -597,6 +618,9 @@ class Zeus_8_3_2_B_4_2(IStrategy):
informative = self.calculateTendency(informative, 3)
informative = self.apply_regression_derivatives(informative, column='mid', window=5, degree=3)
informative['max12'] = talib.MAX(informative['close'], timeperiod=12)
informative['min12'] = talib.MIN(informative['close'], timeperiod=12)
informative['rsi'] = talib.RSI(informative['close']) #, timeperiod=7)
informative['rsi_diff'] = informative['rsi'].diff()
informative['rsi_sum'] = (informative['rsi'].rolling(7).sum() - 350) / 7
@@ -607,26 +631,6 @@ class Zeus_8_3_2_B_4_2(IStrategy):
informative['sma5_diff_sum'] = (informative['sma5_pct'].rolling(5).sum()) / 5
informative['sma5_diff2_sum'] = informative['sma5_diff_sum'].diff()
sorted_close_prices = informative['close'].tail(365).sort_values()
lowest_4 = sorted_close_prices.head(4)
informative['lowest_4'] = lowest_4.mean()
sorted_close_prices = informative['close'].tail(365).sort_values(ascending=False)
highest_4 = sorted_close_prices.head(4)
informative['highest_4'] = highest_4.mean()
last_14_days = informative.tail(14)
# Récupérer le minimum et le maximum de la colonne 'close' des 14 derniers jours
min_14_days = last_14_days['close'].min()
max_14_days = last_14_days['close'].max()
informative['lowest'] = min_14_days
informative['highest'] = max_14_days
informative['pct_min_max'] = (max_14_days - min_14_days) / min_14_days
informative['mid_min_max'] = min_14_days + (max_14_days - min_14_days) / 2
informative['middle'] = informative['lowest_4'] + (informative['highest_4'] - informative['lowest_4']) / 2
informative['mid_min_max_0.98'] = informative['mid_min_max'] * 0.98
dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1d", ffill=True)
dataframe['count_buys'] = 0
@@ -726,14 +730,14 @@ class Zeus_8_3_2_B_4_2(IStrategy):
)
# Calculer l'amplitude en %
drop_stats['amplitude_pct'] = ((drop_stats['end_price'] - drop_stats['start_price']) / drop_stats[
'start_price']) * 100
# drop_stats = drop_stats[drop_stats['amplitude_pct'] < -1]
# Associer les amplitudes calculées à chaque drop_id dans dataframe
dataframe = dataframe.merge(drop_stats[['amplitude_pct']], on='drop_id', how='left')
# Remplir les lignes sans drop_id par 0
dataframe['amplitude_pct'] = dataframe['amplitude_pct'].fillna(0)
dataframe['amplitude_pct_60'] = dataframe['amplitude_pct'].rolling(60).sum()
# drop_stats['amplitude_pct'] = ((drop_stats['end_price'] - drop_stats['start_price']) / drop_stats[
# 'start_price']) * 100
# # drop_stats = drop_stats[drop_stats['amplitude_pct'] < -1]
# # Associer les amplitudes calculées à chaque drop_id dans dataframe
# dataframe = dataframe.merge(drop_stats[['amplitude_pct']], on='drop_id', how='left')
# # Remplir les lignes sans drop_id par 0
# dataframe['amplitude_pct'] = dataframe['amplitude_pct'].fillna(0)
# dataframe['amplitude_pct_60'] = dataframe['amplitude_pct'].rolling(60).sum()
# ----------------------------------------------------------
# self.getBinanceOrderBook(pair, dataframe)
@@ -844,14 +848,14 @@ class Zeus_8_3_2_B_4_2(IStrategy):
& (dataframe['open'] < dataframe['average_line_288'])
& (dataframe['up_count'] > 0)
), ['enter_long', 'enter_tag']] = (1, 'mnmx200')
dataframe.loc[
(
(dataframe['close'].shift(2) <= dataframe['min200'])
& (dataframe['pct_change'] < 0)
& (dataframe['min200'].shift(2) == dataframe['min200'])
& (dataframe['close'] < dataframe['lowest_4_average'])
& (dataframe['up_count'] > 0)
), ['enter_long', 'enter_tag']] = (1, 'min200')
# dataframe.loc[
# (
# (dataframe['close'].shift(2) <= dataframe['min200'])
# & (dataframe['pct_change'] < 0)
# & (dataframe['min200'].shift(2) == dataframe['min200'])
# & (dataframe['close'] < dataframe['lowest_4_average'])
# & (dataframe['up_count'] > 0)
# ), ['enter_long', 'enter_tag']] = (1, 'min200')
dataframe.loc[
(
@@ -985,63 +989,30 @@ class Zeus_8_3_2_B_4_2(IStrategy):
def adjust_stake_amount(self, pair: str, last_candle: DataFrame):
# Calculer le minimum des 14 derniers jours
current_price = last_candle['close']
# trade = self.getTrade(pair)
# if trade:
# current_price = trade.open_rate
base_stake_amount = self.config.get('stake_amount', 100) # Montant de base configuré
# Calculer le max des 14 derniers jours
min_14_days_4 = last_candle['lowest_4_1d']
max_14_days_4 = last_candle['highest_4_1d']
percent_4 = 1 - (current_price - min_14_days_4) / (max_14_days_4 - min_14_days_4)
factor_4 = 1 / ((current_price - min_14_days_4) / (max_14_days_4 - min_14_days_4))
max_min_4 = max_14_days_4 / min_14_days_4
# min_14_days = dataframe['lowest_1d']
# max_14_days = dataframe['highest_1d']
# percent = 1 - (current_price - min_14_days) / (max_14_days - min_14_days)
# factor = 1 / ((current_price - min_14_days) / (max_14_days - min_14_days))
# max_min = max_14_days / min_14_days
# Stack amount ajusté price=2473.47 min_max=0.15058074985054215 percent=0.8379141364642171 amount=20.0
first_price = self.pairs[pair]['first_buy']
if (first_price == 0):
first_price = last_candle['close']
last_max = last_candle['max200']
if self.pairs[pair]['last_max'] > 0:
last_max = self.pairs[pair]['last_max']
last_count = self.pairs[pair]['last_count_of_buys']
# factor = 1
last_max = last_candle['max12_1d']
# if self.pairs[pair]['last_max'] == 0:
# self.pairs[pair]['last_max'] = last_candle['max12_1d']
# print(f"last_max set to {last_max}")
#
# if last_max > 0:
# pct = 100 * (last_max - first_price) / last_max
#
# if pct >= 20:
# factor = 2
# else:
# if pct >= 15:
# factor = 1.5
# if self.pairs[pair]['last_max'] > 0:
# last_max = self.pairs[pair]['last_max']
# print(f"last_max is {last_max}")
pct = 5
if last_max > 0:
pct = 100 * (last_max - first_price) / last_max
thresholds = [2, 10, 20, 30]
factors = [0.5, 1.0, 1.5, 2.0]
thresholds = [2, 5, 10, 20]
factors = [1, 1.25, 1.5, 2.0]
factor = self.multi_step_interpolate(pct, thresholds, factors)
# factor = self.interpolate_factor(pct, start_pct=5, end_pct=50, start_factor=0.8, end_factor=3.0)
adjusted_stake_amount = base_stake_amount * factor #max(base_stake_amount, min(100, base_stake_amount * percent_4))
# if pair in ('BTC/USDT', 'ETH/USDT'):
# if percent_4 > 0.5:
# adjusted_stake_amount = 300
# adjusted_stake_amount_2 = max(base_stake_amount / 2.5, min(75, base_stake_amount * percent))
# print(f"Stack amount ajusté price={current_price} factor={factor} amount={adjusted_stake_amount:.4f}")
# print(f"Stack amount ajusté price={current_price} max_min={max_min:.4f} min_14={min_14_days:.4f} max_14={max_14_days:.4f} factor={factor:.4f} percent={percent:.4f} amount={adjusted_stake_amount_2:.4f}")
return adjusted_stake_amount
@@ -1136,96 +1107,6 @@ class Zeus_8_3_2_B_4_2(IStrategy):
#
# return adjusted_stake_amount
def analyze_conditions(self, pair: str, row: DataFrame):
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return
if row is None or row.empty:
return
# Créer un tableau pour stocker les résultats de l'analyse
results = []
# row = dataframe.iloc[-1].squeeze()
# result = {'triggered': False, 'conditions_failed': []}
try:
buy_level = row['buy_level']
except Exception as exception:
print(exception)
return None
# Première condition : 'buy_fractal'
print('------------------------------------------------')
print('Test buy fractal ' + pair + ' buy_level=' + str(buy_level))
if not (
(row['close'] <= (row['min200'] * 1.002)) and
(row['percent_max_144'] <= -0.012) and
(row['haopen'] < buy_level) and
(row['open'] < row['average_line_288']) and
(dataframe['min50'].shift(3).iloc[-1] == row['min50'])
):
failed_conditions = []
if row['close'] > (row['min200'] * 1.002):
print('close > min200 * 1.002')
if row['percent_max_144'] > -0.012:
print('percent_max_144 > -0.012')
if row['haopen'] >= buy_level:
print('haopen >= buy_level')
if row['open'] >= row['average_line_288']:
print('open >= average_line_288')
if dataframe['min50'].shift(3).iloc[-1] != row['min50']:
print('min50.shift(3) != min50')
# result['conditions_failed'].append({'buy_fractal': failed_conditions})
print('------------------------------------------------')
print('Test buy_max_diff_015 ' + pair + ' buy_level=' + str(buy_level))
# Deuxième condition : 'buy_max_diff_015'
if not (
(dataframe['max200_diff'].shift(4).iloc[-1] >= 0.015) and
(row['close'] <= row['lowest_4_average'] * 1.002) and
(row['close'] <= row['min200'] * 1.002) and
(dataframe['max50_diff'].shift(4).iloc[-1] >= 0.01) and
(row['haclose'] < row['bb_middleband']) and
(row['close'] < buy_level) and
(row['open'] < row['average_line_288']) and
(dataframe['min50'].shift(3).iloc[-1] == row['min50'])
):
if dataframe['max200_diff'].shift(4).iloc[-1] < 0.015:
print('max200_diff.shift(4) < 0.015')
if row['close'] > row['lowest_4_average'] * 1.002:
print('close > lowest_4_average * 1.002')
if row['close'] > row['min200'] * 1.002:
print('close > min200 * 1.002')
if dataframe['max50_diff'].shift(4).iloc[-1] < 0.01:
print('max50_diff.shift(4) < 0.01')
if row['haclose'] >= row['bb_middleband']:
print('haclose >= bb_middleband')
if row['close'] >= buy_level:
print('close >= buy_level')
if row['open'] >= row['average_line_288']:
print('open >= average_line_288')
if dataframe['min50'].shift(3).iloc[-1] != row['min50']:
print('min50.shift(3) != min50')
print('------------------------------------------------')
print('Test buy_min_max_200 ' + pair + ' buy_level=' + str(buy_level))
if not (
(row['close'] <= row['min200'] * 1.002)
and (row['min_max200'] > 0.015)
and (row['haopen'] < buy_level)
and (row['open'] < row['average_line_288'])
):
if row['close'] > row['min200'] * 1.002:
print('close > row[min200] * 1.002')
if row['min_max200'] <= 0.015:
print('row[min_max200] <= 0.015')
if row['haopen'] < buy_level:
print('row[haopen] < buy_level')
if row['open'] < row['average_line_288']:
print('row[open] >= row[average_line_288]')
print('------------------------------------------------')
# Ajouter le résultat à la liste des résultats
# results.append(result)
# print(result)
def getBinanceOrderBook(self, pair, dataframe: DataFrame):
"""Fetch the order book (depth) from Binance."""
# print(dataframe)
@@ -1593,15 +1474,13 @@ class Zeus_8_3_2_B_4_2(IStrategy):
return df
def test_signal_success(self, df, percent=0.03, window_size=36):
def test_signal_success(self, df, condition, percent=0.03, window_size=36):
"""
df : DataFrame avec colonnes ['close', 'high', ...]
percent : hausse recherchée (ex: 0.03 pour +3%)
window_size : nombre de bougies (ex: 36 pour 3h en 5m)
"""
# Exemple condition : RSI < 30 et EMA20 > SMA50
condition = (df['down_count'] == 0) & (df['down_count'].shift(1) < 0) & (df['down_pct'].shift(1) < -3)
hits = 0
total = 0
@@ -1616,7 +1495,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
future_highs = df.loc[future_idx, 'close']
if (future_highs >= price_now * (1 + percent)).any():
print(df.loc[future_idx])
# print(f"{price_now} ==> {df.loc[future_idx]['close']}")
hits += 1
total += 1