chore(qav3): store trade exit stage with TP price for constants

optimization

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 4d4e4550dd867a40824d8122402dcdd9ca666230..d3b45a966c8bd2270526c0509bdffb3a58d169c6 100644 (file)
--- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
+++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
@@ -128,7 +128,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel):
         )
         self._optuna_hp_value: dict[str, float] = {}
         self._optuna_train_value: dict[str, float] = {}
-        self._optuna_label_values: dict[str, list] = {}
+        self._optuna_label_values: dict[str, list[float | int]] = {}
         self._optuna_hp_params: dict[str, dict[str, Any]] = {}
         self._optuna_train_params: dict[str, dict[str, Any]] = {}
         self._optuna_label_params: dict[str, dict[str, Any]] = {}
@@ -210,14 +210,16 @@ class QuickAdapterRegressorV3(BaseRegressionModel):
         else:
             raise ValueError(f"Invalid namespace: {namespace}")
 
-    def get_optuna_values(self, pair: str, namespace: str) -> list:
+    def get_optuna_values(self, pair: str, namespace: str) -> list[float | int]:
         if namespace == "label":
             values = self._optuna_label_values.get(pair)
         else:
             raise ValueError(f"Invalid namespace: {namespace}")
         return values
 
-    def set_optuna_values(self, pair: str, namespace: str, values: list) -> None:
+    def set_optuna_values(
+        self, pair: str, namespace: str, values: list[float | int]
+    ) -> None:
         if namespace == "label":
             self._optuna_label_values[pair] = values
         else:

diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py
index 592cfe0c4e5c2bd7958ecb37f2e6ee1f614ae3dd..f2e76dab3ffc37cda8834e1a58e9c4cc1fef0041 100644 (file)
--- a/quickadapter/user_data/strategies/QuickAdapterV3.py
+++ b/quickadapter/user_data/strategies/QuickAdapterV3.py
@@ -855,12 +855,12 @@ class QuickAdapterV3(IStrategy):
         take_profit_price = (
             trade.open_rate + (-1 if trade.is_short else 1) * take_profit_distance
         )
-        self.safe_append_trade_take_profit_price(trade, take_profit_price)
+        self.safe_append_trade_take_profit_price(trade, take_profit_price, exit_stage)
 
         return take_profit_price
 
     @staticmethod
-    def _get_trade_history(trade: Trade) -> dict[str, list[float]]:
+    def _get_trade_history(trade: Trade) -> dict[str, list[float | tuple[int, float]]]:
         return trade.get_custom_data(
             "history", {"unrealized_pnl": [], "take_profit_price": []}
         )
@@ -871,7 +871,9 @@ class QuickAdapterV3(IStrategy):
         return history.get("unrealized_pnl", [])
 
     @staticmethod
-    def get_trade_take_profit_price_history(trade: Trade) -> list[float]:
+    def get_trade_take_profit_price_history(
+        trade: Trade,
+    ) -> list[float | tuple[int, float]]:
         history = QuickAdapterV3._get_trade_history(trade)
         return history.get("take_profit_price", [])
 
@@ -899,11 +901,11 @@ class QuickAdapterV3(IStrategy):
         return trade_unrealized_pnl_history
 
     def append_trade_take_profit_price(
-        self, trade: Trade, take_profit_price: float
-    ) -> list[float]:
+        self, trade: Trade, take_profit_price: float, exit_stage: int
+    ) -> list[float | tuple[int, float]]:
         history = QuickAdapterV3._get_trade_history(trade)
         price_history = history.setdefault("take_profit_price", [])
-        price_history.append(take_profit_price)
+        price_history.append((exit_stage, take_profit_price))
         if len(price_history) > self._max_history_size:
             price_history = price_history[-self._max_history_size :]
             history["take_profit_price"] = price_history
@@ -911,21 +913,35 @@ class QuickAdapterV3(IStrategy):
         return price_history
 
     def safe_append_trade_take_profit_price(
-        self, trade: Trade, take_profit_price: float
-    ) -> list[float]:
+        self, trade: Trade, take_profit_price: float, exit_stage: int
+    ) -> list[float | tuple[int, float]]:
         trade_take_profit_price_history = (
             QuickAdapterV3.get_trade_take_profit_price_history(trade)
         )
-        previous_take_profit_price = (
+        previous_take_profit_entry = (
             trade_take_profit_price_history[-1]
             if trade_take_profit_price_history
             else None
         )
-        if previous_take_profit_price is None or not np.isclose(
-            previous_take_profit_price, take_profit_price
+        previous_exit_stage = None
+        previous_take_profit_price = None
+        if isinstance(previous_take_profit_entry, tuple):
+            previous_exit_stage = (
+                previous_take_profit_entry[0] if previous_take_profit_entry else None
+            )
+            previous_take_profit_price = (
+                previous_take_profit_entry[1] if previous_take_profit_entry else None
+            )
+        elif isinstance(previous_take_profit_entry, float):
+            previous_exit_stage = -1
+            previous_take_profit_price = previous_take_profit_entry
+        if (
+            previous_take_profit_price is None
+            or (previous_exit_stage is not None and previous_exit_stage != exit_stage)
+            or not np.isclose(previous_take_profit_price, take_profit_price)
         ):
             trade_take_profit_price_history = self.append_trade_take_profit_price(
-                trade, take_profit_price
+                trade, take_profit_price, exit_stage
             )
         return trade_take_profit_price_history