refactor(qav3): factor out quantile computation

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 70a7d6d3d2ae5c44b515c4ac89f24b9e10ae12fb..676a4e30a5430a4611c37b15c1e7b29e0bb8eecb 100644 (file)
--- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
+++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
@@ -45,7 +45,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel):
     https://github.com/sponsors/robcaulk
     """
 
-    version = "3.7.49"
+    version = "3.7.50"
 
     @cached_property
     def _optuna_config(self) -> dict:
@@ -842,6 +842,21 @@ def hp_objective(
     return error
 
 
+def calculate_quantile(values: np.ndarray, value: float) -> float:
+    if values.size == 0:
+        return np.nan
+
+    first_value = values[0]
+    if np.all(np.isclose(values, first_value)):
+        return (
+            0.5
+            if np.isclose(value, first_value)
+            else (0.0 if value < first_value else 1.0)
+        )
+
+    return np.sum(values <= value) / values.size
+
+
 class TrendDirection(IntEnum):
     NEUTRAL = 0
     UP = 1
@@ -884,13 +899,6 @@ def zigzag(
     candidate_pivot_value = np.nan
     candidate_pivot_direction: TrendDirection = TrendDirection.NEUTRAL
 
-    def calculate_quantile(values: np.ndarray, pos: int, start: int, end: int) -> float:
-        if start >= end:
-            return np.nan
-        if np.all(np.isclose(values[start:end], values[start])):
-            return 0.5
-        return np.sum(values[start:end] <= values[pos]) / len(values[start:end])
-
     def calculate_depth_factor(
         pos: int,
         min_factor: float = 0.5,
@@ -901,7 +909,9 @@ def zigzag(
         if start >= end:
             return (min_factor + max_factor) / 2
 
-        quantile = calculate_quantile(get_natr_values(natr_period), pos, start, end)
+        natr_values = get_natr_values(natr_period)
+        lookback_natr_values = natr_values[start:end]
+        quantile = calculate_quantile(lookback_natr_values, natr_values[pos])
 
         return max_factor - (max_factor - min_factor) * quantile
 
@@ -931,7 +941,9 @@ def zigzag(
         if start >= end:
             return min_strength
 
-        quantile = calculate_quantile(get_natr_values(natr_period), pos, start, end)
+        natr_values = get_natr_values(natr_period)
+        lookback_natr_values = natr_values[start:end]
+        quantile = calculate_quantile(lookback_natr_values, natr_values[pos])
 
         return min_strength + (max_strength - min_strength) * quantile
 

diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py
index 589e0b09bd6fca6d5642606acd4f5464545a57a5..f6e1ef637a152208a7b7d49b32b2e843d3df5afb 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,
     zigzag,
     ewo,
     non_zero_diff,
@@ -58,7 +59,7 @@ class QuickAdapterV3(IStrategy):
     INTERFACE_VERSION = 3
 
     def version(self) -> str:
-        return "3.3.47"
+        return "3.3.48"
 
     timeframe = "5m"
 
@@ -500,21 +501,6 @@ class QuickAdapterV3(IStrategy):
             * (1 / math.log10(3.75 + 0.25 * trade_duration_candles))
         )
 
-    @staticmethod
-    def calculate_quantile(values: np.ndarray, value: float) -> float:
-        if values.size == 0:
-            return np.nan
-
-        if np.all(np.isclose(values, values[0])):
-            if np.isclose(value, values[0]):
-                return 0.5
-            elif value < values[0]:
-                return 0.0
-            else:
-                return 1.0
-
-        return np.sum(values <= value) / values.size
-
     def get_take_profit_distance(self, df: DataFrame, trade: Trade) -> Optional[float]:
         trade_duration_candles = QuickAdapterV3.get_trade_duration_candles(df, trade)
         if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
@@ -529,9 +515,7 @@ class QuickAdapterV3(IStrategy):
         trade_natr_values = df.loc[
             df["date"] >= entry_date, "natr_label_period_candles"
         ].values
-        current_natr_quantile = QuickAdapterV3.calculate_quantile(
-            trade_natr_values, current_natr
-        )
+        current_natr_quantile = calculate_quantile(trade_natr_values, current_natr)
         if isna(current_natr_quantile):
             return None
         entry_natr_weight = 1.0 - current_natr_quantile

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index 45cf4cf20a770577032eba2bf4830e5a4bfd1e3d..400316dfd584c047bf452d6194373859c55a2261 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -339,6 +339,21 @@ def find_fractals(
     )
 
 
+def calculate_quantile(values: np.ndarray, value: float) -> float:
+    if values.size == 0:
+        return np.nan
+
+    first_value = values[0]
+    if np.all(np.isclose(values, first_value)):
+        return (
+            0.5
+            if np.isclose(value, first_value)
+            else (0.0 if value < first_value else 1.0)
+        )
+
+    return np.sum(values <= value) / values.size
+
+
 class TrendDirection(IntEnum):
     NEUTRAL = 0
     UP = 1
@@ -381,13 +396,6 @@ def zigzag(
     candidate_pivot_value = np.nan
     candidate_pivot_direction: TrendDirection = TrendDirection.NEUTRAL
 
-    def calculate_quantile(values: np.ndarray, pos: int, start: int, end: int) -> float:
-        if start >= end:
-            return np.nan
-        if np.all(np.isclose(values[start:end], values[start])):
-            return 0.5
-        return np.sum(values[start:end] <= values[pos]) / len(values[start:end])
-
     def calculate_depth_factor(
         pos: int,
         min_factor: float = 0.5,
@@ -398,7 +406,9 @@ def zigzag(
         if start >= end:
             return (min_factor + max_factor) / 2
 
-        quantile = calculate_quantile(get_natr_values(natr_period), pos, start, end)
+        natr_values = get_natr_values(natr_period)
+        lookback_natr_values = natr_values[start:end]
+        quantile = calculate_quantile(lookback_natr_values, natr_values[pos])
 
         return max_factor - (max_factor - min_factor) * quantile
 
@@ -428,7 +438,9 @@ def zigzag(
         if start >= end:
             return min_strength
 
-        quantile = calculate_quantile(get_natr_values(natr_period), pos, start, end)
+        natr_values = get_natr_values(natr_period)
+        lookback_natr_values = natr_values[start:end]
+        quantile = calculate_quantile(lookback_natr_values, natr_values[pos])
 
         return min_strength + (max_strength - min_strength) * quantile