From: Jérôme Benoit Date: Fri, 19 Jun 2026 16:03:56 +0000 (+0200) Subject: feat(quickadapter): add Kaiser-Bessel-derived smoothing X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=48d55f6de9be1db093d914da95fd9ff0e31bb461;p=freqai-strategies.git feat(quickadapter): add Kaiser-Bessel-derived smoothing --- diff --git a/quickadapter/user_data/strategies/LabelTransformer.py b/quickadapter/user_data/strategies/LabelTransformer.py index 1ce4cb1..6e9a17b 100644 --- a/quickadapter/user_data/strategies/LabelTransformer.py +++ b/quickadapter/user_data/strategies/LabelTransformer.py @@ -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", diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index a6a6b9a..4f9334d 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -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:") diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index a489c90..265dbbd 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -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(),