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

Ajout calcul de probabilité fonction de 2 indicateurs

Browse Source
This commit is contained in:
Jérôme Delacotte
2025-05-15 00:59:04 +02:00
parent abe4019342
commit 775c950306
2 changed files with 265 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
Zeus_8_3_2_B_4_2.py
View File

@@ -209,6 +209,24 @@ class Zeus_8_3_2_B_4_2(IStrategy):
protection_fibo = IntParameter(1, 10, default=2, space='protection')
sell_allow_decrease = DecimalParameter(0.005, 0.02, default=0.2, decimals=2, space='sell', optimize=True, load=True)
data = {
"B5": [41.0, 41.2, 34.1, 27.5, 35.0, 30.6, 25.2, 29.8, 25.7, 30.6, 14.8],
"B4": [47.2, 35.8, 39.7, 27.9, 26.5, 19.9, 28.7, 20.8, 29.4, 27.5, 21.6],
"B3": [48.1, 48.4, 42.8, 32.3, 24.4, 23.6, 28.6, 23.9, 22.7, 25.1, 22.2],
"B2": [45.6, 46.5, 47.0, 33.2, 34.9, 30.8, 25.8, 30.4, 29.8, 22.6, 35.3],
"B1": [74.0, 59.9, 63.3, 61.9, 50.0, 41.9, 35.9, 34.4, 37.7, 30.8, 19.3],
"N0": [65.9, 60.2, 64.5, 67.1, 59.2, 59.2, 44.2, 37.5, 47.1, 34.1, 31.6],
"H1": [66.5, 75.8, 71.5, 70.8, 69.4, 67.5, 60.1, 52.7, 59.9, 50.9, 38.3],
"H2": [83.8, 79.4, 80.4, 79.5, 72.8, 70.6, 68.8, 66.1, 68.5, 59.8, 59.6],
"H3": [77.8, 84.6, 82.0, 81.3, 79.8, 74.0, 67.7, 69.8, 66.5, 57.0, 65.2],
"H4": [72.1, 83.0, 86.6, 73.6, 77.4, 63.0, 69.6, 67.5, 68.6, 68.6, 56.8],
"H5": [81.0, 78.5, 76.6, 81.9, 69.5, 75.0, 80.9, 62.9, 66.4, 63.7, 59.6]
}
index_labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
matrix_df = pd.DataFrame(data, index=index_labels)
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:
@@ -225,34 +243,40 @@ class Zeus_8_3_2_B_4_2(IStrategy):
# print(f"restart {pair} last_sell={self.pairs[pair]['last_sell'] * 0.99} minutes={minutes}")
# self.pairs[pair]['stop'] = False
mid_smooth_label = self.get_mid_smooth_label(last_candle['mid_smooth_deriv1_1h']) # ex. 'B2'
sma24_diff_label = self.get_sma24_diff_label(last_candle['sma24_diff_1h'])
val = self.approx_val_from_bins(row_label=sma24_diff_label, col_label=mid_smooth_label)
# allow_to_buy = True #(not self.stop_all) #& (not self.all_down)
allow_to_buy = not self.pairs[pair]['stop'] #not last_candle['tendency'] in ('B-', 'B--') # (rate <= float(limit)) | (entry_tag == 'force_entry')
self.trades = list()
dispo = round(self.wallets.get_available_stake_amount())
allow_to_buy = not self.pairs[pair]['stop'] and val > 50 #not last_candle['tendency'] in ('B-', 'B--') # (rate <= float(limit)) | (entry_tag == 'force_entry')
self.pairs[pair]['first_buy'] = rate
self.pairs[pair]['last_buy'] = rate
self.pairs[pair]['max_touch'] = last_candle['close']
self.pairs[pair]['last_candle'] = last_candle
self.pairs[pair]['count_of_buys'] = 1
self.pairs[pair]['current_profit'] = 0
if allow_to_buy:
self.trades = list()
self.pairs[pair]['first_buy'] = rate
self.pairs[pair]['last_buy'] = rate
self.pairs[pair]['max_touch'] = last_candle['close']
self.pairs[pair]['last_candle'] = last_candle
self.pairs[pair]['count_of_buys'] = 1
self.pairs[pair]['current_profit'] = 0
self.printLineLog()
dispo = round(self.wallets.get_available_stake_amount())
self.printLineLog()
stake_amount = self.adjust_stake_amount(pair, last_candle)
stake_amount = self.adjust_stake_amount(pair, last_candle)
self.log_trade(
last_candle=last_candle,
date=current_time,
action=("Buy" if allow_to_buy else "Canceled") + " " + str(minutes),
pair=pair,
rate=rate,
dispo=dispo,
profit=0,
trade_type=entry_tag,
buys=1,
stake=round(stake_amount, 2)
)
self.log_trade(
last_candle=last_candle,
date=current_time,
action=("Buy" if allow_to_buy else "Canceled") + " " + str(minutes),
pair=pair,
rate=rate,
dispo=dispo,
profit=0,
trade_type=entry_tag,
buys=1,
stake=round(stake_amount, 2)
)
return allow_to_buy
@@ -265,7 +289,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
allow_to_sell = (last_candle['percent'] < 0)
minutes = round((current_time - trade.date_last_filled_utc).total_seconds() / 60, 0)
minutes = int(round((current_time - trade.date_last_filled_utc).total_seconds() / 60, 0))
if allow_to_sell:
self.trades = list()
@@ -280,7 +304,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
self.log_trade(
last_candle=last_candle,
date=current_time,
action="Sell " + str(round(minutes,0)),
action="Sell " + str(minutes),
pair=pair,
trade_type=exit_reason,
rate=last_candle['close'],
@@ -381,12 +405,9 @@ class Zeus_8_3_2_B_4_2(IStrategy):
if self.config.get('runmode') == 'hyperopt':
return
if self.columns_logged % 30 == 0:
# print(
# f"|{'-' * 18}+{'-' * 12}+{'-' * 12}+{'-' * 20}+{'-' * 14}+{'-' * 8}+{'-' * 10}+{'-' * 7}+{'-' * 13}+{'-' * 14}+{'-' * 14}+{'-' * 7}+{'-' * 12}|"
# )
self.printLog(
f"| {'Date':<16} | {'Action':<10} |{'Pair':<5}| {'Trade Type':<18} |{'Rate':>8} | {'Dispo':>6} | {'Profit':>8} | {'Pct':>6} | {'max_touch':>11} | {'last_lost':>12} | {'last_max':>7} |{'Buys':>4}| {'Stake':>5} |"
f"Tdc|Tdh|Tdd|Tdc|Tdh|Tdd|"
f"| {'Date':<16} | {'Action':<10} |{'Pair':<5}| {'Trade Type':<18} |{'Rate':>8} | {'Dispo':>6} | {'Profit':>8} | {'Pct':>6} | {'max_touch':>11} | {'last_lost':>12} | {'last_max':>7}|{'Buys':>4}| {'Stake':>5} |"
f"sum_1h|sum_1d|Tdc|Tdh|Tdd| drv1 |drv_1h|drv_1d|"
)
self.printLineLog()
@@ -434,13 +455,15 @@ class Zeus_8_3_2_B_4_2(IStrategy):
f"| {round(self.pairs[pair]['last_max'], 0) or '-':>7} |{buys or '-':>4}|{stake or '-':>7}"
f"|{round(last_candle['sma5_diff_sum_1h'], 2) or '-':>6}|{round(last_candle['sma5_diff_sum_1d'], 2) or '-':>6}"
f"|{last_candle['tendency'] or '-':>3}|{last_candle['tendency_1h'] or '-':>3}|{last_candle['tendency_1d'] or '-':>3}"
f"|{round(last_candle['mid_smooth_deriv1'],3) or '-':>6}|{round(last_candle['mid_smooth_deriv1_1h'],3) or '-':>6}|{round(last_candle['mid_smooth_deriv1_1d'],3) or '-' :>6}"
f"|{round(last_candle['mid_smooth_deriv1'],3) or '-':>6}|{round(last_candle['mid_smooth_deriv1_1h'],3) or '-':>6}|{round(last_candle['mid_smooth_deriv1_1d'],3) or '-' :>6}|"
# f"|{round(last_candle['mid_smooth_deriv2']) or '-' :>3 }|{round(last_candle['mid_smooth_deriv2_1h']) or '-':>5}|{round(last_candle['mid_smooth_deriv2_1d']) or '-':>5}"
)
def printLineLog(self):
# f"sum1h|sum1d|Tdc|Tdh|Tdd| drv1 |drv_1h|drv_1d|"
self.printLog(
f"+{'-' * 18}+{'-' * 12}+{'-' * 5}+{'-' * 20}+{'-' * 9}+{'-' * 8}+{'-' * 10}+{'-' * 8}+{'-' * 13}+{'-' * 14}+{'-' * 9}+{'-' * 4}+{'-' * 7}+"
f"{'-' * 6}+{'-' * 6}+{'-' * 3}+{'-' * 3}+{'-' * 3}+{'-' * 6}+{'-' * 6}+{'-' * 6}+"
)
def printLog(self, str):
@@ -547,10 +570,10 @@ class Zeus_8_3_2_B_4_2(IStrategy):
informative['haclose'] = heikinashi['close']
informative['hapercent'] = (informative['haclose'] - informative['haopen']) / informative['haclose']
informative = self.calculateTendency(informative, 3)
informative = self.calculateTendency(informative, 12)
# informative = self.apply_regression_derivatives(informative, column='mid', window=5, degree=3)
informative['volatility'] = talib.STDDEV(informative['close'], timeperiod=14) / informative['close']
informative['atr'] = (talib.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14)) / informative['close']
# informative['volatility'] = talib.STDDEV(informative['close'], timeperiod=14) / informative['close']
# informative['atr'] = (talib.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14)) / informative['close']
informative['rsi'] = talib.RSI(informative['close']) #, timeperiod=7)
informative['rsi_diff'] = informative['rsi'].diff()
informative['rsi_sum'] = (informative['rsi'].rolling(7).sum() - 350) / 7
@@ -573,9 +596,9 @@ class Zeus_8_3_2_B_4_2(IStrategy):
################### INFORMATIVE 1d
informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe="1d")
informative = self.calculateTendency(informative, 3)
informative['volatility'] = talib.STDDEV(informative['close'], timeperiod=14) / informative['close']
informative['atr'] = (talib.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14)) / informative['close']
informative = self.calculateTendency(informative, 7)
# informative['volatility'] = talib.STDDEV(informative['close'], timeperiod=14) / informative['close']
# informative['atr'] = (talib.ATR(informative['high'], informative['low'], informative['close'], timeperiod=14)) / informative['close']
# informative = self.apply_regression_derivatives(informative, column='mid', window=5, degree=3)
informative['max12'] = talib.MAX(informative['close'], timeperiod=12)
@@ -629,9 +652,10 @@ class Zeus_8_3_2_B_4_2(IStrategy):
# dataframe['amount'] = amount
print(f"amount= {amount}")
dataframe['futur_price_1h'] = dataframe['close'].shift(-12)
dataframe['futur_price_2h'] = dataframe['close'].shift(-24)
dataframe['futur_price_3h'] = dataframe['close'].shift(-36)
dataframe['futur_percent_1h'] = 100 * (dataframe['close'].shift(-12) - dataframe['close']) / dataframe['close']
dataframe['futur_percent_3h'] = 100 * (dataframe['close'].shift(-36) - dataframe['close']) / dataframe['close']
dataframe['futur_percent_5h'] = 100 * (dataframe['close'].shift(-60) - dataframe['close']) / dataframe['close']
dataframe['futur_percent_12h'] = 100 * (dataframe['close'].shift(-144) - dataframe['close']) / dataframe['close']
return dataframe
@@ -711,6 +735,64 @@ class Zeus_8_3_2_B_4_2(IStrategy):
if self.dp.runmode.value in ('backtest'):
dataframe.to_feather(f"user_data/data/binance/{metadata['pair'].replace('/', '_')}_df.feather")
df = dataframe
# # Définition des tranches pour les dérivées
# bins_deriv = [-np.inf, -0.05, -0.01, 0.01, 0.05, np.inf]
# labels = ['forte baisse', 'légère baisse', 'neutre', 'légère hausse', 'forte hausse']
#
# # Ajout des colonnes bin (catégorisation)
# df[f"{indic_1}_bin"] = pd.cut(df['mid_smooth_deriv1_1h'], bins=bins_deriv, labels=labels)
# df[f"{indic_2}_bin"] = pd.cut(df['mid_smooth_deriv1_1d'], bins=bins_deriv, labels=labels)
#
# # Colonnes de prix futur à analyser
# futur_cols = ['futur_percent_1h', 'futur_percent_2h', 'futur_percent_3h', 'futur_percent_4h', 'futur_percent_5h']
#
# # Calcul des moyennes et des effectifs
# grouped = df.groupby([f"{indic_2}_bin", f"{indic_1}_bin"])[futur_cols].agg(['mean', 'count'])
#
# pd.set_option('display.width', 200) # largeur max affichage
# pd.set_option('display.max_columns', None)
# Colonnes à traiter
futur_cols = ['futur_percent_1h', 'futur_percent_3h', 'futur_percent_5h', 'futur_percent_12h']
# Tranches équitables par quantiles
# Exemple pour 10 quantiles
labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
indic_1 = 'mid_smooth_deriv1_1h'
indic_2 = 'sma24_diff_1h'
df[f"{indic_1}_bin"], bins_1h = pd.qcut(df[f"{indic_1}"], q=11, labels=labels, retbins=True, duplicates='drop')
df[f"{indic_2}_bin"], bins_1d = pd.qcut(df[f"{indic_2}"], q=11, labels=labels, retbins=True, duplicates='drop')
pd.set_option('display.max_columns', None)
pd.set_option('display.width', 300) # largeur max affichage
# Afficher les bornes
print(f"Bornes des quantiles pour {indic_1} :", bins_1h)
print(f"Bornes des quantiles pour {indic_2} :", bins_1d)
# Agrégation
grouped = df.groupby([f"{indic_2}_bin", f"{indic_1}_bin"], observed=True)[futur_cols].agg(['mean', 'count'])
# Affichage
with pd.option_context('display.max_rows', None, 'display.max_columns', None):
print(grouped.round(4))
# Ajout des probabilités de hausse
for col in futur_cols:
df[f"{col}_is_up"] = df[col] > 0
# Calcul de la proba de hausse
proba_up = df.groupby([f"{indic_2}_bin", f"{indic_1}_bin"], observed=True)[f"{col}_is_up"].mean().unstack()
print(f"\nProbabilité de hausse pour {col} (en %):")
with pd.option_context('display.max_rows', None, 'display.max_columns', None):
print((proba_up * 100).round(1))
return dataframe
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
@@ -788,22 +870,18 @@ class Zeus_8_3_2_B_4_2(IStrategy):
lim = - pct - (count_of_buys * 0.001)
# print(f"{trade.pair} current_profit={current_profit} count_of_buys={count_of_buys} pct_max={pct_max:.3f} lim={lim:.3f} rsi_diff_1f={last_candle['rsi_diff_1h']}")
mid_smooth_label = self.get_mid_smooth_label(last_candle['mid_smooth_deriv1_1h']) # ex. 'B2'
sma24_diff_label = self.get_sma24_diff_label(last_candle['sma24_diff_1h'])
val = self.approx_val_from_bins(row_label=sma24_diff_label, col_label=mid_smooth_label)
# print(f"Valeur approximée pour B3 / H2 : {val:.2f}")
# if (days_since_open > count_of_buys) & (0 < count_of_buys <= max_buys) & (current_rate <= limit) & (last_candle['enter_long'] == 1):
limit_buy = 20
if (count_of_buys < limit_buy) \
and ((last_candle['enter_long'] == 1)
or (last_candle['percent48'] < - 0.03)
or ((last_candle['min50'] == last_candle_3['min50']) and (last_candle['low'] <= last_candle['min50']))
) \
and (last_candle['rsi_diff_1h'] >= -5) \
and (last_candle['tendency'] in ('P', 'H++', 'DH', 'H+')) \
and (last_candle['sma5_diff_sum_1d'] > -1 or count_of_buys <= 9) \
and (pct_max < lim):
and (last_candle['enter_long'] == 1) \
and (pct_max < lim and val > 50):
try:
# and (last_candle['mid_smooth_deriv1_1h'] > 0 and last_candle['mid_smooth_deriv1_1h'] >= last_candle_previous_1h['mid_smooth_deriv1_1h'])
# and (last_candle['mid_smooth_deriv1_1d'] > -1000 or last_candle['mid_smooth_deriv1_1h'] > 200) \
# and (last_candle['mid_smooth_deriv1_1d'] > -1500) \
# and not (last_candle['mid_smooth_deriv1_1d'] < - 500 and last_candle['mid_smooth_deriv1_1h'] < 0) \
max_amount = self.config.get('stake_amount', 100) * 2.5
stake_amount = min(min(max_amount, self.wallets.get_available_stake_amount()),
@@ -835,15 +913,6 @@ class Zeus_8_3_2_B_4_2(IStrategy):
# and (last_candle['tendency'] in ('P', 'H++', 'DH', 'H+')) \
# and (last_candle['mid_smooth_deriv1'] > 0.015):
# try:
#
# # and (last_candle['mid_smooth_deriv1_1d'] > -1000 or last_candle['mid_smooth_deriv1_1h'] > 200) \
# # and (last_candle['mid_smooth_deriv1_1d'] > -1500) \
# # and not (last_candle['mid_smooth_deriv1_1d'] < - 500 and last_candle['mid_smooth_deriv1_1h'] < 0) \
#
# max_amount = self.config.get('stake_amount', 100) * 2.5
# stake_amount = min(min(max_amount, self.wallets.get_available_stake_amount()),
# self.adjust_stake_amount(pair, last_candle) - 10 * pct_first / pct) # min(200, self.adjust_stake_amount(pair, last_candle) * self.fibo[count_of_buys])
#
# trade_type = last_candle['enter_tag'] if last_candle['enter_long'] == 1 else 'pct48'
# self.log_trade(
# last_candle=last_candle,
@@ -981,6 +1050,111 @@ class Zeus_8_3_2_B_4_2(IStrategy):
return df
def get_mid_smooth_label(self, value):
bins = [-2.0622, -0.1618, -0.0717, -0.0353, -0.0135, 0.0, 0.0085, 0.0276, 0.0521, 0.0923, 0.1742, 2.3286]
labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
for i in range(len(bins) - 1):
if bins[i] <= value < bins[i + 1]:
return labels[i]
return labels[-1] # cas limite pour la borne max
def get_sma24_diff_label(self, value):
bins = [-0.84253877, -0.13177195, -0.07485074, -0.04293497, -0.02033502, -0.00215711,
0.01411933, 0.03308264, 0.05661652, 0.09362708, 0.14898214, 0.50579505]
labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
for i in range(len(bins) - 1):
if bins[i] <= value < bins[i + 1]:
return labels[i]
return labels[-1]
import numpy as np
import pandas as pd
def interpolated_val_from_bins(self, row_pos, col_pos):
"""
Renvoie une approximation interpolée (bilinéaire) d'une valeur dans la matrice
à partir de positions flottantes dans l'index (ligne) et les colonnes.
Parameters:
matrix_df (pd.DataFrame): Matrice des probabilités (index/colonnes = labels).
row_pos (float): Position réelle de la ligne (0 = B5, 10 = H5).
col_pos (float): Position réelle de la colonne (0 = B5, 10 = H5).
Returns:
float: Valeur interpolée, ou NaN si en dehors des bornes.
"""
# Labels ordonnés
labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
n = len(labels)
# Vérification des limites
if not (0 <= row_pos <= n - 1) or not (0 <= col_pos <= n - 1):
return np.nan
# Conversion des labels -> matrice
matrix = self.matrix_df.reindex(index=labels, columns=labels).values
# Coordonnées entières (inférieures)
i = int(np.floor(row_pos))
j = int(np.floor(col_pos))
# Coefficients pour interpolation
dx = row_pos - i
dy = col_pos - j
# Précautions sur les bords
if i >= n - 1: i = n - 2; dx = 1.0
if j >= n - 1: j = n - 2; dy = 1.0
# Récupération des 4 valeurs voisines
v00 = matrix[i][j]
v10 = matrix[i + 1][j]
v01 = matrix[i][j + 1]
v11 = matrix[i + 1][j + 1]
# Interpolation bilinéaire
interpolated = (
(1 - dx) * (1 - dy) * v00 +
dx * (1 - dy) * v10 +
(1 - dx) * dy * v01 +
dx * dy * v11
)
return interpolated
def approx_val_from_bins(self, row_label, col_label):
"""
Renvoie une approximation de la valeur à partir des labels binaires (e.g. B5, H1)
en utilisant une interpolation simple basée sur les indices.
Parameters:
matrix_df (pd.DataFrame): Matrice avec les labels binaires en index et colonnes.
row_label (str): Label de la ligne (ex: 'B3').
col_label (str): Label de la colonne (ex: 'H2').
Returns:
float: Valeur approchée si possible, sinon NaN.
"""
# Vérification des labels
if row_label not in self.matrix_df.index or col_label not in self.matrix_df.columns:
return np.nan
# Récupération des labels ordonnés
ordered_labels = ['B5', 'B4', 'B3', 'B2', 'B1', 'N0', 'H1', 'H2', 'H3', 'H4', 'H5']
label_to_index = {label: i for i, label in enumerate(ordered_labels)}
# Index correspondant
row_idx = label_to_index.get(row_label)
col_idx = label_to_index.get(col_label)
# Extraction de la matrice numérique
numeric_matrix = self.matrix_df.reindex(index=ordered_labels, columns=ordered_labels).values
# Approximation directe (aucune interpolation complexe ici, juste une lecture)
return numeric_matrix[row_idx, col_idx]
# @property
# def protections(self):
# return [