]> Piment Noir Git Repositories - freqai-strategies.git/commitdiff
feat(quickadapter): add Kaiser-Bessel-derived smoothing
authorJérôme Benoit <jerome.benoit@piment-noir.org>
Fri, 19 Jun 2026 16:03:56 +0000 (18:03 +0200)
committerJérôme Benoit <jerome.benoit@piment-noir.org>
Fri, 19 Jun 2026 16:03:56 +0000 (18:03 +0200)
quickadapter/user_data/strategies/LabelTransformer.py
quickadapter/user_data/strategies/QuickAdapterV3.py
quickadapter/user_data/strategies/Utils.py

index 1ce4cb19af8a104b8dfffe3c9f5b6135ad89f54b..6e9a17b302d5bff47bcc78d39bc284c306c5b1f8 100644 (file)
@@ -127,12 +127,21 @@ DEFAULTS_LABEL_PIPELINE: Final[dict[str, Any]] = {
 
 
 SmoothingMethod = Literal[
-    "none", "gaussian", "kaiser", "triang", "smm", "sma", "savgol", "gaussian_filter1d"
+    "none",
+    "gaussian",
+    "kaiser",
+    "kaiser_bessel_derived",
+    "triang",
+    "smm",
+    "sma",
+    "savgol",
+    "gaussian_filter1d",
 ]
 SMOOTHING_METHODS: Final[tuple[SmoothingMethod, ...]] = (
     "none",
     "gaussian",
     "kaiser",
+    "kaiser_bessel_derived",
     "triang",
     "smm",
     "sma",
index a6a6b9a4bede2f540c5d4a757a6d4b1e8b0949d9..4f9334d6fc0d41a3dc1bc85acc8687007b270744 100644 (file)
@@ -536,9 +536,9 @@ class QuickAdapterV3(IStrategy):
 
             method = col_smoothing["method"]
             if col_weighting["strategy"] != WEIGHT_STRATEGIES[0] and (  # "none"
-                method == SMOOTHING_METHODS[4]  # "smm"
+                method == SMOOTHING_METHODS[5]  # "smm"
                 or (
-                    method == SMOOTHING_METHODS[6]  # "savgol"
+                    method == SMOOTHING_METHODS[7]  # "savgol"
                     and col_smoothing["polyorder"] >= 2
                 )
             ):
@@ -550,8 +550,8 @@ class QuickAdapterV3(IStrategy):
                     f"to zero), which may trip the all-rows-dropped guard in "
                     f"compose_sample_weights once a non-'none' "
                     f"label_weighting strategy is configured. Prefer a "
-                    f"non-negative linear kernel (gaussian, kaiser, triang, "
-                    f"sma, gaussian_filter1d)."
+                    f"non-negative linear kernel (gaussian, kaiser, "
+                    f"kaiser_bessel_derived, triang, sma, gaussian_filter1d)."
                 )
 
         logger.info("Reversal Confirmation:")
index a489c906d04605f73772837fd15770b6b5f200fd..265dbbd6b83ebfc968a15847c72bfb2ac4a4b0a8 100644 (file)
@@ -383,10 +383,11 @@ def generate_label_data(
     return generator(dataframe, params)
 
 
-SmoothingKernel = Literal["gaussian", "kaiser", "triang"]
+SmoothingKernel = Literal["gaussian", "kaiser", "kaiser_bessel_derived", "triang"]
 SMOOTHING_KERNELS: Final[tuple[SmoothingKernel, ...]] = (
     "gaussian",
     "kaiser",
+    "kaiser_bessel_derived",
     "triang",
 )
 
@@ -905,6 +906,13 @@ def get_odd_window(window: int) -> int:
     return window if window % 2 == 1 else window + 1
 
 
+@lru_cache(maxsize=8)
+def get_even_window(window: int) -> int:
+    if window < 1:
+        raise ValueError(f"Invalid window value {window!r}: must be > 0")
+    return window if window % 2 == 0 else window + 1
+
+
 @lru_cache(maxsize=8)
 def get_gaussian_std(window: int) -> float:
     return (window - 1) / 6.0 if window > 1 else 0.5
@@ -931,7 +939,13 @@ def _calculate_coeffs(
         coeffs = sp.signal.windows.gaussian(M=window, std=std, sym=True)
     elif win_type == SMOOTHING_KERNELS[1]:  # "kaiser"
         coeffs = sp.signal.windows.kaiser(M=window, beta=beta, sym=True)
-    elif win_type == SMOOTHING_KERNELS[2]:  # "triang"
+    elif win_type == SMOOTHING_KERNELS[2]:  # "kaiser_bessel_derived"
+        coeffs = sp.signal.windows.kaiser_bessel_derived(
+            M=window,
+            beta=beta,
+            sym=True,
+        )
+    elif win_type == SMOOTHING_KERNELS[3]:  # "triang"
         coeffs = sp.signal.windows.triang(M=window, sym=True)
     else:
         raise ValueError(
@@ -1005,19 +1019,28 @@ def smooth(
             std=std,
             beta=beta,
         )
-    elif method == SMOOTHING_METHODS[3]:  # "triang"
+    elif method == SMOOTHING_METHODS[3]:  # "kaiser_bessel_derived"
+        even_window = get_even_window(window_candles)
+        return zero_phase_filter(
+            series=series,
+            window=even_window,
+            win_type=SMOOTHING_KERNELS[2],  # "kaiser_bessel_derived"
+            std=std,
+            beta=beta,
+        )
+    elif method == SMOOTHING_METHODS[4]:  # "triang"
         return zero_phase_filter(
             series=series,
             window=odd_window,
-            win_type=SMOOTHING_KERNELS[2],  # "triang"
+            win_type=SMOOTHING_KERNELS[3],  # "triang"
             std=std,
             beta=beta,
         )
-    elif method == SMOOTHING_METHODS[4]:  # "smm" (Simple Moving Median)
+    elif method == SMOOTHING_METHODS[5]:  # "smm" (Simple Moving Median)
         return series.rolling(window=odd_window, center=True, min_periods=1).median()
-    elif method == SMOOTHING_METHODS[5]:  # "sma" (Simple Moving Average)
+    elif method == SMOOTHING_METHODS[6]:  # "sma" (Simple Moving Average)
         return series.rolling(window=odd_window, center=True, min_periods=1).mean()
-    elif method == SMOOTHING_METHODS[6]:  # "savgol" (Savitzky-Golay)
+    elif method == SMOOTHING_METHODS[7]:  # "savgol" (Savitzky-Golay)
         w, p, m = get_savgol_params(odd_window, polyorder, mode)
         if n < w:
             return series
@@ -1030,7 +1053,7 @@ def smooth(
             ),
             index=series.index,
         )
-    elif method == SMOOTHING_METHODS[7]:  # "gaussian_filter1d"
+    elif method == SMOOTHING_METHODS[8]:  # "gaussian_filter1d"
         return pd.Series(
             gaussian_filter1d(
                 series.to_numpy(),