refactor(qav3): reuse existing implementation when possible

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 3cc1f554cdc4d26b95d5dc315d7156bcf8fe9764..e5aaea4a08418941566b6e3ec3392f956be96a3c 100644 (file)
--- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
+++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
@@ -21,6 +21,9 @@ from numpy.typing import NDArray
 from sklearn_extra.cluster import KMedoids
 
 from Utils import (
+    EXTREMA_COLUMN,
+    MAXIMA_THRESHOLD_COLUMN,
+    MINIMA_THRESHOLD_COLUMN,
     REGRESSORS,
     Regressor,
     calculate_min_extrema,
@@ -47,10 +50,6 @@ debug = False
 
 TEST_SIZE: Final = 0.1
 
-EXTREMA_COLUMN: Final = "&s-extrema"
-MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold"
-MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold"
-
 warnings.simplefilter(action="ignore", category=FutureWarning)
 
 logger = logging.getLogger(__name__)

diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py
index 12092edb58a2b8c5bb932eae46e944a5b19112bc..4e2dd37b96b724a88d356b3433af42ce40a2d041 100644 (file)
--- a/quickadapter/user_data/strategies/QuickAdapterV3.py
+++ b/quickadapter/user_data/strategies/QuickAdapterV3.py
@@ -13,7 +13,6 @@ from typing import (
     Literal,
     Optional,
     Sequence,
-    Tuple,
 )
 
 import numpy as np
@@ -30,6 +29,9 @@ from technical.pivots_points import pivots_points
 from Utils import (
     DEFAULTS_EXTREMA_SMOOTHING,
     DEFAULTS_EXTREMA_WEIGHTING,
+    EXTREMA_COLUMN,
+    MAXIMA_THRESHOLD_COLUMN,
+    MINIMA_THRESHOLD_COLUMN,
     NORMALIZATION_TYPES,
     RANK_METHODS,
     SMOOTHING_METHODS,
@@ -62,20 +64,16 @@ InterpolationDirection = Literal["direct", "inverse"]
 OrderType = Literal["entry", "exit"]
 TradingMode = Literal["spot", "margin", "futures"]
 
-DfSignature = Tuple[int, Optional[datetime.datetime]]
-CandleDeviationCacheKey = Tuple[
+DfSignature = tuple[int, Optional[datetime.datetime]]
+CandleDeviationCacheKey = tuple[
     str, DfSignature, float, float, int, InterpolationDirection, float
 ]
-CandleThresholdCacheKey = Tuple[str, DfSignature, str, int, float, float]
+CandleThresholdCacheKey = tuple[str, DfSignature, str, int, float, float]
 
 debug = False
 
 logger = logging.getLogger(__name__)
 
-EXTREMA_COLUMN: Final = "&s-extrema"
-MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold"
-MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold"
-
 
 class QuickAdapterV3(IStrategy):
     """

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index 7eb986a0a93f730e6bc1c704447a63be567f73b7..dd26e10e007e60b8f86b232d0913c7e13b1f3118 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -25,6 +25,10 @@ WEIGHT_STRATEGIES: Final[tuple[WeightStrategy, ...]] = (
     "amplitude_excess",
 )
 
+EXTREMA_COLUMN: Final = "&s-extrema"
+MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold"
+MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold"
+
 NormalizationType = Literal[
     "minmax", "zscore", "l1", "l2", "robust", "softmax", "tanh", "rank", "none"
 ]
@@ -225,19 +229,7 @@ def _normalize_zscore(
     if not rescale_to_unit_range:
         return z_scores
 
-    z_min = np.min(z_scores)
-    z_max = np.max(z_scores)
-    z_range = z_max - z_min
-
-    if np.isclose(z_range, 0.0):
-        return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float)
-
-    normalized_weights = (z_scores - z_min) / z_range
-
-    if np.isnan(normalized_weights).any():
-        return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float)
-
-    return normalized_weights
+    return _normalize_minmax(z_scores)
 
 
 def _normalize_minmax(weights: NDArray[np.floating]) -> NDArray[np.floating]:
@@ -292,17 +284,8 @@ def _normalize_robust(
     if np.isclose(iqr, 0.0):
         return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float)
 
-    robust_scores = (weights - median) / iqr
-
-    r_min = np.min(robust_scores)
-    r_max = np.max(robust_scores)
-    r_range = r_max - r_min
-
-    if np.isclose(r_range, 0.0):
-        return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float)
-
-    normalized_weights = (robust_scores - r_min) / r_range
-    return normalized_weights
+    robust_weights = (weights - median) / iqr
+    return _normalize_minmax(robust_weights)
 
 
 def _normalize_softmax(