refactor(qav3): factor out extrema smoothing params handling

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 f565126282341415a8a3043d6dbc8ff42de96ca7..12092edb58a2b8c5bb932eae46e944a5b19112bc 100644 (file)
--- a/quickadapter/user_data/strategies/QuickAdapterV3.py
+++ b/quickadapter/user_data/strategies/QuickAdapterV3.py
@@ -714,6 +714,49 @@ class QuickAdapterV3(IStrategy):
             "rank_method": weighting_rank_method,
         }
 
+    @staticmethod
+    def _get_extrema_smoothing_params(
+        extrema_smoothing: dict[str, Any], pair: str
+    ) -> dict[str, Any]:
+        smoothing_method = str(
+            extrema_smoothing.get("method", DEFAULTS_EXTREMA_SMOOTHING["method"])
+        )
+        if smoothing_method not in set(SMOOTHING_METHODS):
+            logger.warning(
+                f"{pair}: invalid extrema_smoothing method '{smoothing_method}', using default '{SMOOTHING_METHODS[0]}'"
+            )
+            smoothing_method = SMOOTHING_METHODS[0]
+
+        smoothing_window = extrema_smoothing.get(
+            "window", DEFAULTS_EXTREMA_SMOOTHING["window"]
+        )
+        if not isinstance(smoothing_window, int) or smoothing_window < 3:
+            logger.warning(
+                f"{pair}: invalid extrema_smoothing window {smoothing_window}, must be an integer >= 3, using default {DEFAULTS_EXTREMA_SMOOTHING['window']}"
+            )
+            smoothing_window = DEFAULTS_EXTREMA_SMOOTHING["window"]
+
+        smoothing_beta = extrema_smoothing.get(
+            "beta", DEFAULTS_EXTREMA_SMOOTHING["beta"]
+        )
+        if (
+            not isinstance(smoothing_beta, (int, float))
+            or not np.isfinite(smoothing_beta)
+            or smoothing_beta <= 0
+        ):
+            logger.warning(
+                f"{pair}: invalid extrema_smoothing beta {smoothing_beta}, must be a finite number > 0, using default {DEFAULTS_EXTREMA_SMOOTHING['beta']}"
+            )
+            smoothing_beta = DEFAULTS_EXTREMA_SMOOTHING["beta"]
+        else:
+            smoothing_beta = float(smoothing_beta)
+
+        return {
+            "method": smoothing_method,
+            "window": int(smoothing_window),
+            "beta": smoothing_beta,
+        }
+
     @staticmethod
     @lru_cache(maxsize=128)
     def _td_format(
@@ -784,25 +827,6 @@ class QuickAdapterV3(IStrategy):
                 f"{pair}: labeled {len(pivots_indices)} extrema (label_period={QuickAdapterV3._td_format(label_period)} / {label_period_candles=} / {label_natr_ratio=:.2f})"
             )
 
-        extrema_smoothing = self.freqai_info.get("extrema_smoothing", {})
-        if not isinstance(extrema_smoothing, dict):
-            extrema_smoothing = {}
-
-        smoothing_method = str(
-            extrema_smoothing.get("method", DEFAULTS_EXTREMA_SMOOTHING["method"])
-        )
-        if smoothing_method not in set(SMOOTHING_METHODS):
-            logger.warning(
-                f"{pair}: invalid extrema_smoothing method '{smoothing_method}', using default '{SMOOTHING_METHODS[0]}'"
-            )
-            smoothing_method = SMOOTHING_METHODS[0]
-        smoothing_window = int(
-            extrema_smoothing.get("window", DEFAULTS_EXTREMA_SMOOTHING["window"])
-        )
-        smoothing_beta = float(
-            extrema_smoothing.get("beta", DEFAULTS_EXTREMA_SMOOTHING["beta"])
-        )
-
         extrema_weighting = self.freqai_info.get("extrema_weighting", {})
         if not isinstance(extrema_weighting, dict):
             extrema_weighting = {}
@@ -829,11 +853,18 @@ class QuickAdapterV3(IStrategy):
             rank_method=extrema_weighting_params["rank_method"],
         )
 
+        extrema_smoothing = self.freqai_info.get("extrema_smoothing", {})
+        if not isinstance(extrema_smoothing, dict):
+            extrema_smoothing = {}
+        extrema_smoothing_params = QuickAdapterV3._get_extrema_smoothing_params(
+            extrema_smoothing, pair
+        )
+
         dataframe[EXTREMA_COLUMN] = smooth_extrema(
             weighted_extrema,
-            smoothing_method,
-            smoothing_window,
-            smoothing_beta,
+            extrema_smoothing_params["method"],
+            extrema_smoothing_params["window"],
+            extrema_smoothing_params["beta"],
         )
         if debug:
             extrema = dataframe[EXTREMA_COLUMN]

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index d727698985ddb7873e74668708c5ced245a14880..7eb986a0a93f730e6bc1c704447a63be567f73b7 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -477,10 +477,10 @@ def get_weighted_extrema(
     ):  # "none"
         return extrema, default_weights
 
-    if strategy in (
+    if strategy in {
         WEIGHT_STRATEGIES[1],
         WEIGHT_STRATEGIES[2],
-    ):  # "amplitude" or "amplitude_excess"
+    }:  # "amplitude" or "amplitude_excess"
         extrema_weights = calculate_extrema_weights(
             series=extrema,
             indices=indices,