refactor(qav3): refactor trade entry/exit threshold computation code

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
index a740257c0c82aca43e46a2685fa72fccec5287ff..7e177fb1ff3e36e9796b4457aef083f1f48e9e3a 100644 (file)
--- 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 7c61bd7a9076f468ef407a5ef5769cd40e4f3a30..d872d27cc16866e0057228fb4b32403597129bd7 100644 (file)
--- 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
         ):