From 4b695dd9ee004177023c611c276fca625904a1d9 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Thu, 12 Jun 2025 00:59:40 +0200 Subject: [PATCH] perf(qav3): refine pivots labeling optimization objectives MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- .../freqaimodels/QuickAdapterRegressorV3.py | 52 ++++++++++++------- .../user_data/strategies/QuickAdapterV3.py | 2 +- quickadapter/user_data/strategies/Utils.py | 44 +++++++++++----- 3 files changed, 65 insertions(+), 33 deletions(-) diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index e4c5020..af3870a 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -1185,7 +1185,7 @@ class TrendDirection(IntEnum): DOWN = -1 -zigzag_cache: dict[str, tuple[list[int], list[float], list[int]]] = {} +zigzag_cache: dict[str, tuple[list[int], list[float], list[int]], list[float]] = {} def zigzag_cached( @@ -1193,7 +1193,7 @@ def zigzag_cached( natr_period: int = 14, natr_ratio: float = 6.0, cache_size: int = 2048, -) -> tuple[list[int], list[float], list[int]]: +) -> tuple[list[int], list[float], list[int], list[float]]: def hash_df(df: pd.DataFrame) -> str: hasher = hashlib.sha256() @@ -1210,7 +1210,7 @@ def zigzag_cached( if cache_key in zigzag_cache: return zigzag_cache[cache_key] - pivots_indices, pivots_values, pivots_directions = zigzag( + pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs = zigzag( df, natr_period=natr_period, natr_ratio=natr_ratio ) if len(zigzag_cache) >= cache_size: @@ -1219,15 +1219,16 @@ def zigzag_cached( pivots_indices, pivots_values, pivots_directions, + pivots_scaled_natrs, ) - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs def zigzag( df: pd.DataFrame, natr_period: int = 14, natr_ratio: float = 6.0, -) -> tuple[list[int], list[float], list[int]]: +) -> tuple[list[int], list[float], list[int], list[float]]: min_confirmation_window: int = 3 max_confirmation_window: int = 6 n = len(df) @@ -1245,12 +1246,18 @@ def zigzag( state: TrendDirection = TrendDirection.NEUTRAL depth = -1 + pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs = ( + [], + [], + [], + [], + ) last_pivot_pos = -1 - pivots_indices, pivots_values, pivots_directions = [], [], [] - candidate_pivot_pos = -1 - candidate_pivot_value = np.nan + candidate_pivot_pos: int = -1 + candidate_pivot_value: float = np.nan candidate_pivot_direction: TrendDirection = TrendDirection.NEUTRAL + candidate_pivot_scaled_natr: float = np.nan volatility_quantile_cache: dict[int, float] = {} @@ -1309,17 +1316,27 @@ def zigzag( return min_strength + (max_strength - min_strength) * volatility_quantile def update_candidate_pivot(pos: int, value: float, direction: TrendDirection): - nonlocal candidate_pivot_pos, candidate_pivot_value, candidate_pivot_direction + nonlocal \ + candidate_pivot_pos, \ + candidate_pivot_value, \ + candidate_pivot_direction, \ + candidate_pivot_scaled_natr if 0 <= pos < n: candidate_pivot_pos = pos candidate_pivot_value = value candidate_pivot_direction = direction + candidate_pivot_scaled_natr = thresholds[pos] def reset_candidate_pivot(): - nonlocal candidate_pivot_pos, candidate_pivot_value, candidate_pivot_direction + nonlocal \ + candidate_pivot_pos, \ + candidate_pivot_value, \ + candidate_pivot_direction, \ + candidate_pivot_scaled_natr candidate_pivot_pos = -1 candidate_pivot_value = np.nan candidate_pivot_direction = TrendDirection.NEUTRAL + candidate_pivot_scaled_natr = np.nan def add_pivot(pos: int, value: float, direction: TrendDirection): nonlocal last_pivot_pos, depth @@ -1328,6 +1345,7 @@ def zigzag( pivots_indices.append(indices[pos]) pivots_values.append(value) pivots_directions.append(direction) + pivots_scaled_natrs.append(thresholds[pos]) last_pivot_pos = pos depth = calculate_depth(pos) reset_candidate_pivot() @@ -1458,10 +1476,10 @@ def zigzag( state = TrendDirection.UP break else: - return [], [], [] + return [], [], [], [] if n - last_pivot_pos - 1 < depth: - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs for i in range(last_pivot_pos + 1, n): current_high = highs[i] @@ -1492,7 +1510,7 @@ def zigzag( ) state = TrendDirection.UP - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs def label_objective( @@ -1526,17 +1544,13 @@ def label_objective( if df.empty: return -np.inf, -np.inf - _, pivots_values, _ = zigzag_cached( + _, pivots_values, _, pivots_scaled_natrs = zigzag_cached( df, natr_period=label_period_candles, natr_ratio=label_natr_ratio, ) - scaled_natr_label_period_candles = ( - ta.NATR(df, timeperiod=label_period_candles).bfill() / 100.0 - ) * label_natr_ratio - - return scaled_natr_label_period_candles.median(), len(pivots_values) + return np.median(pivots_scaled_natrs), len(pivots_values) def smoothed_max(series: pd.Series, temperature=1.0) -> float: diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index 0d54790..7a50995 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -407,7 +407,7 @@ class QuickAdapterV3(IStrategy): pair = str(metadata.get("pair")) label_period_candles = self.get_label_period_candles(pair) label_natr_ratio = self.get_label_natr_ratio(pair) - pivots_indices, _, pivots_directions = zigzag( + pivots_indices, _, pivots_directions, _ = zigzag( dataframe, natr_period=label_period_candles, natr_ratio=label_natr_ratio, diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index dc579aa..826cbc7 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -378,7 +378,7 @@ class TrendDirection(IntEnum): DOWN = -1 -zigzag_cache: dict[str, tuple[list[int], list[float], list[int]]] = {} +zigzag_cache: dict[str, tuple[list[int], list[float], list[int]], list[float]] = {} def zigzag_cached( @@ -386,7 +386,7 @@ def zigzag_cached( natr_period: int = 14, natr_ratio: float = 6.0, cache_size: int = 2048, -) -> tuple[list[int], list[float], list[int]]: +) -> tuple[list[int], list[float], list[int], list[float]]: def hash_df(df: pd.DataFrame) -> str: hasher = hashlib.sha256() @@ -403,7 +403,7 @@ def zigzag_cached( if cache_key in zigzag_cache: return zigzag_cache[cache_key] - pivots_indices, pivots_values, pivots_directions = zigzag( + pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs = zigzag( df, natr_period=natr_period, natr_ratio=natr_ratio ) if len(zigzag_cache) >= cache_size: @@ -412,15 +412,16 @@ def zigzag_cached( pivots_indices, pivots_values, pivots_directions, + pivots_scaled_natrs, ) - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs def zigzag( df: pd.DataFrame, natr_period: int = 14, natr_ratio: float = 6.0, -) -> tuple[list[int], list[float], list[int]]: +) -> tuple[list[int], list[float], list[int], list[float]]: min_confirmation_window: int = 3 max_confirmation_window: int = 6 n = len(df) @@ -438,12 +439,18 @@ def zigzag( state: TrendDirection = TrendDirection.NEUTRAL depth = -1 + pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs = ( + [], + [], + [], + [], + ) last_pivot_pos = -1 - pivots_indices, pivots_values, pivots_directions = [], [], [] - candidate_pivot_pos = -1 - candidate_pivot_value = np.nan + candidate_pivot_pos: int = -1 + candidate_pivot_value: float = np.nan candidate_pivot_direction: TrendDirection = TrendDirection.NEUTRAL + candidate_pivot_scaled_natr: float = np.nan volatility_quantile_cache: dict[int, float] = {} @@ -502,17 +509,27 @@ def zigzag( return min_strength + (max_strength - min_strength) * volatility_quantile def update_candidate_pivot(pos: int, value: float, direction: TrendDirection): - nonlocal candidate_pivot_pos, candidate_pivot_value, candidate_pivot_direction + nonlocal \ + candidate_pivot_pos, \ + candidate_pivot_value, \ + candidate_pivot_direction, \ + candidate_pivot_scaled_natr if 0 <= pos < n: candidate_pivot_pos = pos candidate_pivot_value = value candidate_pivot_direction = direction + candidate_pivot_scaled_natr = thresholds[pos] def reset_candidate_pivot(): - nonlocal candidate_pivot_pos, candidate_pivot_value, candidate_pivot_direction + nonlocal \ + candidate_pivot_pos, \ + candidate_pivot_value, \ + candidate_pivot_direction, \ + candidate_pivot_scaled_natr candidate_pivot_pos = -1 candidate_pivot_value = np.nan candidate_pivot_direction = TrendDirection.NEUTRAL + candidate_pivot_scaled_natr = np.nan def add_pivot(pos: int, value: float, direction: TrendDirection): nonlocal last_pivot_pos, depth @@ -521,6 +538,7 @@ def zigzag( pivots_indices.append(indices[pos]) pivots_values.append(value) pivots_directions.append(direction) + pivots_scaled_natrs.append(thresholds[pos]) last_pivot_pos = pos depth = calculate_depth(pos) reset_candidate_pivot() @@ -651,10 +669,10 @@ def zigzag( state = TrendDirection.UP break else: - return [], [], [] + return [], [], [], [] if n - last_pivot_pos - 1 < depth: - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs for i in range(last_pivot_pos + 1, n): current_high = highs[i] @@ -685,4 +703,4 @@ def zigzag( ) state = TrendDirection.UP - return pivots_indices, pivots_values, pivots_directions + return pivots_indices, pivots_values, pivots_directions, pivots_scaled_natrs -- 2.43.0