From e97401ae1668b7591b4f628d0e2b0b2a45698fb2 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 20 May 2025 12:44:39 +0200 Subject: [PATCH] refactor(qav3): factor out quantile computation MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- .../freqaimodels/QuickAdapterRegressorV3.py | 32 +++++++++++++------ .../user_data/strategies/QuickAdapterV3.py | 22 ++----------- quickadapter/user_data/strategies/Utils.py | 30 +++++++++++------ 3 files changed, 46 insertions(+), 38 deletions(-) diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index 70a7d6d..676a4e3 100644 --- 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 589e0b0..f6e1ef6 100644 --- 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 45cf4cf..400316d 100644 --- 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 -- 2.43.0