From 0795d32da645500e994d6d8f7d62b11574b9033c Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Wed, 6 Aug 2025 14:49:48 +0200 Subject: [PATCH] refactor(qav3): refactor trade entry/exit threshold computation code MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- .../freqaimodels/QuickAdapterRegressorV3.py | 27 +++++--- .../user_data/strategies/QuickAdapterV3.py | 64 +++++++------------ 2 files changed, 41 insertions(+), 50 deletions(-) diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index a740257..7e177fb 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -345,7 +345,12 @@ class QuickAdapterRegressorV3(BaseRegressionModel): y_test = y_test.iloc[-test_period_candles:] test_weights = test_weights[-test_period_candles:] - eval_set, eval_weights = self.eval_set_and_weights(X_test, y_test, test_weights) + eval_set, eval_weights = QuickAdapterRegressorV3.eval_set_and_weights( + X_test, + y_test, + test_weights, + self.data_split_parameters.get("test_size", TEST_SIZE), + ) model = fit_regressor( regressor=str(self.freqai_info.get("regressor", "xgboost")), @@ -435,7 +440,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) warmed_up = False - pred_df_full = ( + pred_df = ( self.dd.historic_predictions[pair] .iloc[-fit_live_predictions_candles:] .reset_index(drop=True) @@ -446,7 +451,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): dk.data["extra_returns_per_train"][MAXIMA_THRESHOLD_COLUMN] = 2 else: min_pred, max_pred = self.min_max_pred( - pred_df_full, + pred_df, fit_live_predictions_candles, self.get_optuna_params(pair, "label").get("label_period_candles"), ) @@ -455,16 +460,16 @@ class QuickAdapterRegressorV3(BaseRegressionModel): dk.data["labels_mean"], dk.data["labels_std"] = {}, {} for label in dk.label_list + dk.unique_class_list: - pred_df_full_label = pred_df_full.get(label) - if pred_df_full_label is None or pred_df_full_label.dtype == object: + pred_df_label = pred_df.get(label) + if pred_df_label is None or pred_df_label.dtype == object: continue if not warmed_up: f = [0, 0] else: - f = sp.stats.norm.fit(pred_df_full_label) + f = sp.stats.norm.fit(pred_df_label) dk.data["labels_mean"][label], dk.data["labels_std"][label] = f[0], f[1] - di_values = pred_df_full.get("DI_values") + di_values = pred_df.get("DI_values") # fit the DI_threshold if not warmed_up: @@ -499,12 +504,16 @@ class QuickAdapterRegressorV3(BaseRegressionModel): pair, "train" ) + @staticmethod def eval_set_and_weights( - self, X_test: pd.DataFrame, y_test: pd.DataFrame, test_weights: np.ndarray + X_test: pd.DataFrame, + y_test: pd.DataFrame, + test_weights: np.ndarray, + test_size: float, ) -> tuple[ Optional[list[tuple[pd.DataFrame, pd.DataFrame]]], Optional[list[np.ndarray]] ]: - if self.data_split_parameters.get("test_size", TEST_SIZE) == 0: + if test_size == 0: eval_set = None eval_weights = None else: diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index 7c61bd7..d872d27 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -64,7 +64,7 @@ class QuickAdapterV3(IStrategy): INTERFACE_VERSION = 3 def version(self) -> str: - return "3.3.136" + return "3.3.137" timeframe = "5m" @@ -835,7 +835,7 @@ class QuickAdapterV3(IStrategy): else None ) if previous_take_profit_price != take_profit_price: - QuickAdapterV3.append_trade_take_profit_price(trade, take_profit_price) + self.append_trade_take_profit_price(trade, take_profit_price) if exit_stage not in self.partial_exit_stages: if not trade_take_profit_price_history: @@ -872,8 +872,9 @@ class QuickAdapterV3(IStrategy): history["unrealized_pnl"] = pnl_history[-self._max_pnl_history_size :] trade.set_custom_data("history", history) - @staticmethod - def append_trade_take_profit_price(trade: Trade, take_profit_price: float) -> None: + def append_trade_take_profit_price( + self, trade: Trade, take_profit_price: float + ) -> None: history = QuickAdapterV3._get_trade_history(trade) history.setdefault("take_profit_price", []).append(take_profit_price) trade.set_custom_data("history", history) @@ -935,7 +936,7 @@ class QuickAdapterV3(IStrategy): def weighted_close(series: Series) -> float: return (series.get("high") + series.get("low") + 2 * series.get("close")) / 4.0 - def calculate_current_deviation( + def _calculate_current_deviation( self, df: DataFrame, pair: str, @@ -972,10 +973,18 @@ class QuickAdapterV3(IStrategy): pair, natr_ratio_percent ) - @staticmethod - def calculate_current_threshold( - side: str, last_candle: Series, deviation: float - ) -> float: + def calculate_current_threshold(self, df: DataFrame, pair: str, side: str) -> float: + current_deviation = self._calculate_current_deviation( + df, + pair, + min_natr_ratio_percent=0.0095, + max_natr_ratio_percent=0.095, + interpolation_direction="direct", + ) + if isna(current_deviation): + return np.inf if side == "short" else -np.inf + + last_candle = df.iloc[-1] last_candle_close = last_candle.get("close") last_candle_open = last_candle.get("open") is_last_candle_bullish = last_candle_close > last_candle_open @@ -987,14 +996,14 @@ class QuickAdapterV3(IStrategy): if is_last_candle_bearish else last_candle_close ) - return base_price * (1 + deviation) + return base_price * (1 + current_deviation) elif side == "short": base_price = ( QuickAdapterV3.weighted_close(last_candle) if is_last_candle_bullish else last_candle_close ) - return base_price * (1 - deviation) + return base_price * (1 - current_deviation) raise ValueError(f"Invalid side: {side}. Expected 'long' or 'short'") @@ -1068,24 +1077,12 @@ class QuickAdapterV3(IStrategy): trade.set_custom_data("last_outlier_date", last_candle_date) entry_tag = trade.enter_tag - current_deviation = self.calculate_current_deviation( - df, - pair, - min_natr_ratio_percent=0.0095, - max_natr_ratio_percent=0.085, - interpolation_direction="direct", - ) - if isna(current_deviation): - return None if ( entry_tag == "short" and last_candle.get("do_predict") == 1 and last_candle.get("DI_catch") == 1 and last_candle.get(EXTREMA_COLUMN) < last_candle.get("minima_threshold") - and current_rate - > QuickAdapterV3.calculate_current_threshold( - "long", last_candle, current_deviation - ) + and current_rate > self.calculate_current_threshold(df, pair, "long") ): return "minima_detected_short" if ( @@ -1093,10 +1090,7 @@ class QuickAdapterV3(IStrategy): and last_candle.get("do_predict") == 1 and last_candle.get("DI_catch") == 1 and last_candle.get(EXTREMA_COLUMN) > last_candle.get("maxima_threshold") - and current_rate - < QuickAdapterV3.calculate_current_threshold( - "short", last_candle, current_deviation - ) + and current_rate < self.calculate_current_threshold(df, pair, "short") ): return "maxima_detected_long" @@ -1164,19 +1158,7 @@ class QuickAdapterV3(IStrategy): ) if df.empty: return False - last_candle = df.iloc[-1] - current_deviation = self.calculate_current_deviation( - pair, - df, - min_natr_ratio_percent=0.0095, - max_natr_ratio_percent=0.085, - interpolation_direction="direct", - ) - if isna(current_deviation): - return False - current_threshold = QuickAdapterV3.calculate_current_threshold( - side, last_candle, current_deviation - ) + current_threshold = self.calculate_current_threshold(df, pair, side) if (side == "long" and rate > current_threshold) or ( side == "short" and rate < current_threshold ): -- 2.43.0