feat(qav3): add volatility quantile based method to compute trade exit

price

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

diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py
index 1c022f39009830ef898909aeec5c122f01e54608..9cbf23ec501c3143b1076c3ba023d29439d0aca1 100644 (file)
--- a/quickadapter/user_data/strategies/QuickAdapterV3.py
+++ b/quickadapter/user_data/strategies/QuickAdapterV3.py
@@ -18,6 +18,7 @@ import pandas_ta as pta
 from Utils import (
     alligator,
     bottom_change_percent,
+    calculate_quantile,
     get_zl_ma_fn,
     zigzag,
     ewo,
@@ -483,7 +484,30 @@ class QuickAdapterV3(IStrategy):
             isna(trade_duration) or trade_duration <= 0
         )
 
-    def get_trade_natr(
+    def get_trade_quantile_natr(self, df: DataFrame, trade: Trade) -> Optional[float]:
+        label_natr = df.get("natr_label_period_candles")
+        if label_natr is None or label_natr.empty:
+            return None
+        entry_date = self.get_trade_entry_date(trade)
+        entry_mask = df.get("date") == entry_date
+        if not entry_mask.any():
+            return None
+        entry_natr = label_natr[entry_mask].iloc[0]
+        if isna(entry_natr) or entry_natr < 0:
+            return None
+        current_natr = label_natr.iloc[-1]
+        if isna(current_natr) or current_natr < 0:
+            return None
+        trade_volatility_quantile = calculate_quantile(
+            label_natr.to_numpy(), entry_natr
+        )
+        if isna(trade_volatility_quantile):
+            return None
+        return entry_natr * trade_volatility_quantile + current_natr * (
+            1.0 - trade_volatility_quantile
+        )
+
+    def get_trade_moving_average_natr(
         self, df: DataFrame, pair: str, trade_duration_candles: int
     ) -> Optional[float]:
         if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
@@ -491,7 +515,7 @@ class QuickAdapterV3(IStrategy):
         label_natr = df.get("natr_label_period_candles")
         if label_natr is None or label_natr.empty:
             return None
-        trade_natr = np.nan
+        trade_moving_average_natr = np.nan
         if trade_duration_candles >= 2:
             zl_kama = get_zl_ma_fn("kama")
             try:
@@ -502,14 +526,33 @@ class QuickAdapterV3(IStrategy):
                     ~np.isnan(trade_kama_natr_values)
                 ]
                 if trade_kama_natr_values.size > 0:
-                    trade_natr = trade_kama_natr_values[-1]
+                    trade_moving_average_natr = trade_kama_natr_values[-1]
             except Exception as e:
                 logger.error(
                     f"Failed to calculate KAMA for pair {pair}: {str(e)}", exc_info=True
                 )
-        if isna(trade_natr):
-            trade_natr = zlema(label_natr, period=trade_duration_candles).iloc[-1]
-        return trade_natr
+        if isna(trade_moving_average_natr):
+            trade_moving_average_natr = zlema(
+                label_natr, period=trade_duration_candles
+            ).iloc[-1]
+        return trade_moving_average_natr
+
+    def get_trade_natr(
+        self, df: DataFrame, trade: Trade, trade_duration_candles: int
+    ) -> Optional[float]:
+        trade_price_target = self.config.get("exit_pricing", {}).get(
+            "trade_price_target", "moving_average"
+        )
+        if trade_price_target == "quantile":
+            return self.get_trade_quantile_natr(df, trade)
+        elif trade_price_target == "moving_average":
+            return self.get_trade_moving_average_natr(
+                df, trade.pair, trade_duration_candles
+            )
+        else:
+            raise ValueError(
+                f"Invalid trade_price_target: {trade_price_target}. Expected 'quantile' or 'moving_average'."
+            )
 
     def get_stoploss_distance(
         self, df: DataFrame, trade: Trade, current_rate: float
@@ -517,7 +560,7 @@ class QuickAdapterV3(IStrategy):
         trade_duration_candles = self.get_trade_duration_candles(df, trade)
         if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
             return None
-        trade_natr = self.get_trade_natr(df, trade.pair, trade_duration_candles)
+        trade_natr = self.get_trade_natr(df, trade, trade_duration_candles)
         if isna(trade_natr) or trade_natr < 0:
             return None
         return (
@@ -531,7 +574,7 @@ class QuickAdapterV3(IStrategy):
         trade_duration_candles = self.get_trade_duration_candles(df, trade)
         if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
             return None
-        trade_natr = self.get_trade_natr(df, trade.pair, trade_duration_candles)
+        trade_natr = self.get_trade_natr(df, trade, trade_duration_candles)
         if isna(trade_natr) or trade_natr < 0:
             return None
         return (