From: Jérôme Benoit Date: Thu, 30 Jul 2026 17:22:57 +0000 (+0200) Subject: refactor(quickadapter): consolidate Utils numeric-helper twins (#178) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=bed552c904e71f06955cdba7adaec5db837ec73d;p=freqai-strategies.git refactor(quickadapter): consolidate Utils numeric-helper twins (#178) * refactor(quickadapter): consolidate Utils numeric-helper twins Behavior-preserving deduplication of numeric helpers in Utils.py, proven bit-for-bit identical across a broad input grid (all smooth methods, ceil/floor/round over int/np.integer/float/non-finite/invalid inputs). - smooth: replace the four near-identical zero-phase filter branches and the gaussian else-fallback with a table-driven dispatch `_SMOOTHING_FILTER_SPECS: method -> (kernel, window_selector)`, mirroring the get_ma_fn/get_price_fn `.get(key, default)` idiom. std stays derived from the odd window for every kernel. - ceil_to_step/floor_to_step: extract a shared `_step_round(value, step, int_op, float_op)` core (validation, integer fast-path, finiteness guard, float path); lru_cache stays on the public wrappers only. round_to_step is left untouched (distinct banker's/half-step tie-breaking). - get_odd_window/get_even_window merge (F6) intentionally skipped: they are used as first-class callables in the smooth dispatch table and merging would churn a stable public API and split lru_cache capacity for no gain. Closes #174 * refactor(quickadapter): tighten smooth dispatch comment and step typing Review-driven, no-op refinements (behavior bit-for-bit unchanged): - _SMOOTHING_FILTER_SPECS comment: it does not mirror the get_ma_fn/ get_price_fn idiom (those build a local dict per call); it is a module-level Final table of the _*_SPECS family consumed via .get(method, default). Correct the wording; keep the load-bearing invariants (std stays odd-window-derived; default reproduces the legacy gaussian/odd else-branch). - _step_round: annotate float_op as Callable[[float], int] since only math.ceil/math.floor are passed and both return int. * refactor(quickadapter): rename smooth dispatch table, fix its comment Review-driven, no-op refinement (behavior bit-for-bit unchanged): - Rename _SMOOTHING_FILTER_SPECS -> _ZERO_PHASE_FILTER_DISPATCH. The _*_SPECS suffix is reserved for the dict[str, _ParamSpec] config validation family (e.g. _SMOOTHING_SPECS); this is a runtime dispatch table feeding zero_phase_filter, and the near-homonym _SMOOTHING_SPECS was ambiguous. Module-private symbol, both occurrences updated. - Fix its comment: it is not a _*_SPECS family member, and both the kernel and the window parity vary per method (the invariant is std, which stays odd-window-derived), not "only the window parity". * docs(quickadapter): correct smooth dispatch table comment Review-driven, no-op (comment only, behavior bit-for-bit unchanged): the previous "Both the kernel and the window parity vary per method" put kernel and parity in a false parallel. The kernel varies per method, but the window parity is odd for every method except kaiser_bessel_derived (even). Restate precisely; keep the std-odd-derived and .get-default notes. * refactor(quickadapter): single-source smooth dispatch default Review-driven, no-op (behavior bit-for-bit unchanged): the smooth zero-phase dispatch `.get` default duplicated the gaussian entry value verbatim. Reference the entry (_ZERO_PHASE_FILTER_DISPATCH[SMOOTHING_METHODS[1]]) instead, matching the get_ma_fn/get_price_fn `.get(key, table[default])` idiom and single-sourcing the fallback. The default is only reachable for methods outside the closed SmoothingMethod enum (defensive). * docs(quickadapter): drop redundant dispatch-table paraphrase * docs(quickadapter): drop self-documented dispatch-table header comment --- diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 3f4e4ac..e0e47f4 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -2089,6 +2089,19 @@ def zero_phase_filter( return pd.Series(filtered_values, index=series.index) +_ZERO_PHASE_FILTER_DISPATCH: Final[ + dict[SmoothingMethod, tuple[SmoothingKernel, Callable[[int], int]]] +] = { + SMOOTHING_METHODS[1]: (SMOOTHING_KERNELS[0], get_odd_window), # "gaussian" + SMOOTHING_METHODS[2]: (SMOOTHING_KERNELS[1], get_odd_window), # "kaiser" + SMOOTHING_METHODS[3]: ( + SMOOTHING_KERNELS[2], + get_even_window, + ), # "kaiser_bessel_derived" + SMOOTHING_METHODS[4]: (SMOOTHING_KERNELS[3], get_odd_window), # "triang" +} + + def smooth( series: pd.Series, method: SmoothingMethod = DEFAULTS_LABEL_SMOOTHING["method"], @@ -2114,39 +2127,6 @@ def smooth( if method == SMOOTHING_METHODS[0]: # "none" return series - elif method == SMOOTHING_METHODS[1]: # "gaussian" - return zero_phase_filter( - series=series, - window=odd_window, - win_type=SMOOTHING_KERNELS[0], # "gaussian" - std=std, - beta=beta, - ) - elif method == SMOOTHING_METHODS[2]: # "kaiser" - return zero_phase_filter( - series=series, - window=odd_window, - win_type=SMOOTHING_KERNELS[1], # "kaiser" - std=std, - beta=beta, - ) - 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[3], # "triang" - std=std, - beta=beta, - ) 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[6]: # "sma" (Simple Moving Average) @@ -2173,14 +2153,18 @@ def smooth( ), index=series.index, ) - else: - return zero_phase_filter( - series=series, - window=odd_window, - win_type=SMOOTHING_KERNELS[0], # "gaussian" - std=std, - beta=beta, - ) + + win_type, window_selector = _ZERO_PHASE_FILTER_DISPATCH.get( + method, + _ZERO_PHASE_FILTER_DISPATCH[SMOOTHING_METHODS[1]], # "gaussian"/odd default + ) + return zero_phase_filter( + series=series, + window=window_selector(window_candles), + win_type=win_type, + std=std, + beta=beta, + ) def _impute_weights( @@ -6127,24 +6111,28 @@ def round_to_step(value: float | int, step: int) -> int: return int(round(float(value) / step) * step) -@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE) -def ceil_to_step(value: float | int, step: int) -> int: +def _step_round( + value: float | int, + step: int, + int_op: Callable[[int, int], int], + float_op: Callable[[float], int], +) -> int: _validate_step_args(value, step) if isinstance(value, (int, np.integer)): - return int(-(-int(value) // step) * step) + return int(int_op(int(value), step) * step) if not np.isfinite(value): raise ValueError(f"Invalid value {value!r}: must be finite") - return int(math.ceil(float(value) / step) * step) + return int(float_op(float(value) / step) * step) + + +@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE) +def ceil_to_step(value: float | int, step: int) -> int: + return _step_round(value, step, lambda v, s: -(-v // s), math.ceil) @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE) def floor_to_step(value: float | int, step: int) -> int: - _validate_step_args(value, step) - if isinstance(value, (int, np.integer)): - return int((int(value) // step) * step) - if not np.isfinite(value): - raise ValueError(f"Invalid value {value!r}: must be finite") - return int(math.floor(float(value) / step) * step) + return _step_round(value, step, lambda v, s: v // s, math.floor) def get_label_defaults(