]> Piment Noir Git Repositories - freqai-strategies.git/commitdiff
refactor(quickadapter): consolidate Utils numeric-helper twins (#178)
authorJérôme Benoit <jerome.benoit@piment-noir.org>
Thu, 30 Jul 2026 17:22:57 +0000 (19:22 +0200)
committerGitHub <noreply@github.com>
Thu, 30 Jul 2026 17:22:57 +0000 (19:22 +0200)
* 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

quickadapter/user_data/strategies/Utils.py

index 3f4e4ac9a0ca7b1d605d130583229ae6348519cc..e0e47f44fb14168114661de72074f50771039f99 100644 (file)
@@ -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(