From bd5ee79d8dcb53ca436b55a6bf7bd4d3bab4552f Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 7 Dec 2025 23:19:37 +0100 Subject: [PATCH] refactor(qav3): decouple extrema weighting standardization and normalization MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- README.md | 13 +- .../user_data/strategies/QuickAdapterV3.py | 141 +++++--- quickadapter/user_data/strategies/Utils.py | 335 ++++++++++++------ 3 files changed, 312 insertions(+), 177 deletions(-) diff --git a/README.md b/README.md index c1bc95e..e82d488 100644 --- a/README.md +++ b/README.md @@ -63,13 +63,14 @@ docker compose up -d --build | freqai.extrema_smoothing.beta | 8.0 | float > 0 | Kaiser kernel shape parameter. | | _Extrema weighting_ | | | | | freqai.extrema_weighting.strategy | `none` | enum {`none`,`amplitude`,`amplitude_threshold_ratio`} | Extrema weighting source: unweighted (`none`), swing amplitude (`amplitude`), or volatility-threshold ratio adjusted swing amplitude (`amplitude_threshold_ratio`). | -| freqai.extrema_weighting.normalization | `minmax` | enum {`minmax`,`zscore`,`l1`,`l2`,`robust`,`softmax`,`tanh`,`rank`,`none`} | Normalization method for weights. | -| freqai.extrema_weighting.gamma | 1.0 | float (0,10] | Contrast exponent applied after normalization (>1 emphasizes extrema, 0 0 | Temperature parameter for softmax normalization (lower values sharpen distribution, higher values flatten it). | -| freqai.extrema_weighting.tanh_scale | 1.0 | float > 0 | Scale parameter for tanh normalization. | -| freqai.extrema_weighting.tanh_gain | 1.0 | float > 0 | Gain parameter for tanh normalization. | -| freqai.extrema_weighting.robust_quantiles | [0.25, 0.75] | list[float] where 0 <= q_low < q_high <= 1 | Quantile range for robust normalization. | +| freqai.extrema_weighting.standardization | `none` | enum {`none`,`zscore`,`robust`} | Standardization method applied before normalization. `none`=no standardization, `zscore`=(w-μ)/σ, `robust`=(w-median)/IQR. | +| freqai.extrema_weighting.robust_quantiles | [0.25, 0.75] | list[float] where 0 <= q_low < q_high <= 1 | Quantile range for robust standardization, Q1 and Q3. | +| freqai.extrema_weighting.normalization | `minmax` | enum {`minmax`,`sigmoid`,`softmax`,`l1`,`l2`,`rank`,`none`} | Normalization method for weights. | +| freqai.extrema_weighting.minmax_range | [0.0, 1.0] | list[float, float] | Target range for minmax normalization, min and max. | +| freqai.extrema_weighting.sigmoid_scale | 1.0 | float > 0 | Scale parameter for sigmoid normalization, controls steepness. | +| freqai.extrema_weighting.softmax_temperature | 1.0 | float > 0 | Temperature parameter for softmax normalization: lower values sharpen distribution, higher values flatten it. | | freqai.extrema_weighting.rank_method | `average` | enum {`average`,`min`,`max`,`dense`,`ordinal`} | Ranking method for rank normalization. | +| freqai.extrema_weighting.gamma | 1.0 | float (0,10] | Contrast exponent applied after normalization: >1 emphasizes extrema, values between 0 and 1 soften. | | _Feature parameters_ | | | | | freqai.feature_parameters.label_period_candles | min/max midpoint | int >= 1 | Zigzag labeling NATR horizon. | | freqai.feature_parameters.min_label_period_candles | 12 | int >= 1 | Minimum labeling NATR horizon used for reversals labeling HPO. | diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index cc501ef..830c2d1 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -35,6 +35,7 @@ from Utils import ( NORMALIZATION_TYPES, RANK_METHODS, SMOOTHING_METHODS, + STANDARDIZATION_TYPES, WEIGHT_STRATEGIES, TrendDirection, WeightStrategy, @@ -628,6 +629,7 @@ class QuickAdapterV3(IStrategy): def _get_extrema_weighting_params( extrema_weighting: dict[str, Any], pair: str ) -> dict[str, Any]: + # Strategy weighting_strategy = str( extrema_weighting.get("strategy", DEFAULTS_EXTREMA_WEIGHTING["strategy"]) ) @@ -637,6 +639,41 @@ class QuickAdapterV3(IStrategy): ) weighting_strategy = WEIGHT_STRATEGIES[0] + # Phase 1: Standardization + weighting_standardization = str( + extrema_weighting.get( + "standardization", DEFAULTS_EXTREMA_WEIGHTING["standardization"] + ) + ) + if weighting_standardization not in set(STANDARDIZATION_TYPES): + logger.warning( + f"{pair}: invalid extrema_weighting standardization '{weighting_standardization}', using default '{STANDARDIZATION_TYPES[0]}'" + ) + weighting_standardization = STANDARDIZATION_TYPES[0] + + weighting_robust_quantiles = extrema_weighting.get( + "robust_quantiles", DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"] + ) + if ( + not isinstance(weighting_robust_quantiles, (list, tuple)) + or len(weighting_robust_quantiles) != 2 + or not all( + isinstance(q, (int, float)) and np.isfinite(q) and 0 <= q <= 1 + for q in weighting_robust_quantiles + ) + or weighting_robust_quantiles[0] >= weighting_robust_quantiles[1] + ): + logger.warning( + f"{pair}: invalid extrema_weighting robust_quantiles {weighting_robust_quantiles}, must be (q_low, q_high) with 0 <= q_low < q_high <= 1, using default {DEFAULTS_EXTREMA_WEIGHTING['robust_quantiles']}" + ) + weighting_robust_quantiles = DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"] + else: + weighting_robust_quantiles = ( + float(weighting_robust_quantiles[0]), + float(weighting_robust_quantiles[1]), + ) + + # Phase 2: Normalization weighting_normalization = str( extrema_weighting.get( "normalization", DEFAULTS_EXTREMA_WEIGHTING["normalization"] @@ -648,18 +685,40 @@ class QuickAdapterV3(IStrategy): ) weighting_normalization = NORMALIZATION_TYPES[0] - weighting_gamma = extrema_weighting.get( - "gamma", DEFAULTS_EXTREMA_WEIGHTING["gamma"] + weighting_minmax_range = extrema_weighting.get( + "minmax_range", DEFAULTS_EXTREMA_WEIGHTING["minmax_range"] ) if ( - not isinstance(weighting_gamma, (int, float)) - or not np.isfinite(weighting_gamma) - or not (0 < weighting_gamma <= 10.0) + not isinstance(weighting_minmax_range, (list, tuple)) + or len(weighting_minmax_range) != 2 + or not all( + isinstance(x, (int, float)) and np.isfinite(x) + for x in weighting_minmax_range + ) + or weighting_minmax_range[0] >= weighting_minmax_range[1] ): logger.warning( - f"{pair}: invalid extrema_weighting gamma {weighting_gamma}, must be a finite number in (0, 10], using default {DEFAULTS_EXTREMA_WEIGHTING['gamma']}" + f"{pair}: invalid extrema_weighting minmax_range {weighting_minmax_range}, must be (min, max) with min < max, using default {DEFAULTS_EXTREMA_WEIGHTING['minmax_range']}" ) - weighting_gamma = DEFAULTS_EXTREMA_WEIGHTING["gamma"] + weighting_minmax_range = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"] + else: + weighting_minmax_range = ( + float(weighting_minmax_range[0]), + float(weighting_minmax_range[1]), + ) + + weighting_sigmoid_scale = extrema_weighting.get( + "sigmoid_scale", DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"] + ) + if ( + not isinstance(weighting_sigmoid_scale, (int, float)) + or not np.isfinite(weighting_sigmoid_scale) + or weighting_sigmoid_scale <= 0 + ): + logger.warning( + f"{pair}: invalid extrema_weighting sigmoid_scale {weighting_sigmoid_scale}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['sigmoid_scale']}" + ) + weighting_sigmoid_scale = DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"] weighting_softmax_temperature = extrema_weighting.get( "softmax_temperature", DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"] @@ -676,28 +735,6 @@ class QuickAdapterV3(IStrategy): "softmax_temperature" ] - weighting_robust_quantiles = extrema_weighting.get( - "robust_quantiles", DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"] - ) - if ( - not isinstance(weighting_robust_quantiles, (list, tuple)) - or len(weighting_robust_quantiles) != 2 - or not all( - isinstance(q, (int, float)) and np.isfinite(q) and 0 <= q <= 1 - for q in weighting_robust_quantiles - ) - or weighting_robust_quantiles[0] >= weighting_robust_quantiles[1] - ): - logger.warning( - f"{pair}: invalid extrema_weighting robust_quantiles {weighting_robust_quantiles}, must be (q_low, q_high) with 0 <= q_low < q_high <= 1, using default {DEFAULTS_EXTREMA_WEIGHTING['robust_quantiles']}" - ) - weighting_robust_quantiles = DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"] - else: - weighting_robust_quantiles = ( - float(weighting_robust_quantiles[0]), - float(weighting_robust_quantiles[1]), - ) - weighting_rank_method = str( extrema_weighting.get( "rank_method", DEFAULTS_EXTREMA_WEIGHTING["rank_method"] @@ -709,41 +746,30 @@ class QuickAdapterV3(IStrategy): ) weighting_rank_method = RANK_METHODS[0] - weighting_tanh_scale = extrema_weighting.get( - "tanh_scale", DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"] - ) - if ( - not isinstance(weighting_tanh_scale, (int, float)) - or not np.isfinite(weighting_tanh_scale) - or weighting_tanh_scale <= 0 - ): - logger.warning( - f"{pair}: invalid extrema_weighting tanh_scale {weighting_tanh_scale}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['tanh_scale']}" - ) - weighting_tanh_scale = DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"] - - weighting_tanh_gain = extrema_weighting.get( - "tanh_gain", DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"] + # Phase 3: Post-processing + weighting_gamma = extrema_weighting.get( + "gamma", DEFAULTS_EXTREMA_WEIGHTING["gamma"] ) if ( - not isinstance(weighting_tanh_gain, (int, float)) - or not np.isfinite(weighting_tanh_gain) - or weighting_tanh_gain <= 0 + not isinstance(weighting_gamma, (int, float)) + or not np.isfinite(weighting_gamma) + or not (0 < weighting_gamma <= 10.0) ): logger.warning( - f"{pair}: invalid extrema_weighting tanh_gain {weighting_tanh_gain}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['tanh_gain']}" + f"{pair}: invalid extrema_weighting gamma {weighting_gamma}, must be a finite number in (0, 10], using default {DEFAULTS_EXTREMA_WEIGHTING['gamma']}" ) - weighting_tanh_gain = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"] + weighting_gamma = DEFAULTS_EXTREMA_WEIGHTING["gamma"] return { "strategy": weighting_strategy, + "standardization": weighting_standardization, + "robust_quantiles": weighting_robust_quantiles, "normalization": weighting_normalization, - "gamma": weighting_gamma, + "minmax_range": weighting_minmax_range, + "sigmoid_scale": weighting_sigmoid_scale, "softmax_temperature": weighting_softmax_temperature, - "tanh_scale": weighting_tanh_scale, - "tanh_gain": weighting_tanh_gain, - "robust_quantiles": weighting_robust_quantiles, "rank_method": weighting_rank_method, + "gamma": weighting_gamma, } @staticmethod @@ -867,13 +893,14 @@ class QuickAdapterV3(IStrategy): indices=pivots_indices, weights=np.array(pivot_weights), strategy=self.extrema_weighting["strategy"], + standardization=self.extrema_weighting["standardization"], + robust_quantiles=self.extrema_weighting["robust_quantiles"], normalization=self.extrema_weighting["normalization"], - gamma=self.extrema_weighting["gamma"], + minmax_range=self.extrema_weighting["minmax_range"], + sigmoid_scale=self.extrema_weighting["sigmoid_scale"], softmax_temperature=self.extrema_weighting["softmax_temperature"], - tanh_scale=self.extrema_weighting["tanh_scale"], - tanh_gain=self.extrema_weighting["tanh_gain"], - robust_quantiles=self.extrema_weighting["robust_quantiles"], rank_method=self.extrema_weighting["rank_method"], + gamma=self.extrema_weighting["gamma"], ) dataframe[EXTREMA_COLUMN] = smooth_extrema( diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 1db99a9..317151e 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -29,19 +29,22 @@ EXTREMA_COLUMN: Final = "&s-extrema" MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold" MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold" -NormalizationType = Literal[ - "minmax", "zscore", "l1", "l2", "robust", "softmax", "tanh", "rank", "none" -] +StandardizationType = Literal["none", "zscore", "robust"] +STANDARDIZATION_TYPES: Final[tuple[StandardizationType, ...]] = ( + "none", # 0 - No standardization + "zscore", # 1 - (w - μ) / σ + "robust", # 2 - (w - median) / IQR +) + +NormalizationType = Literal["minmax", "sigmoid", "softmax", "l1", "l2", "rank", "none"] NORMALIZATION_TYPES: Final[tuple[NormalizationType, ...]] = ( - "minmax", # 0 - "zscore", # 1 - "l1", # 2 - "l2", # 3 - "robust", # 4 - "softmax", # 5 - "tanh", # 6 - "rank", # 7 - "none", # 8 + "minmax", # 0 - (w - min) / (max - min) + "sigmoid", # 1 - 1 / (1 + exp(-scale × w)) + "softmax", # 2 - exp(w/T) / Σexp(w/T) + "l1", # 3 - w / Σ|w| + "l2", # 4 - w / ||w||₂ + "rank", # 5 - (rank(w) - 1) / (n - 1) + "none", # 6 - w (identity) ) RankMethod = Literal["average", "min", "max", "dense", "ordinal"] @@ -71,14 +74,18 @@ DEFAULTS_EXTREMA_SMOOTHING: Final[dict[str, Any]] = { } DEFAULTS_EXTREMA_WEIGHTING: Final[dict[str, Any]] = { - "normalization": NORMALIZATION_TYPES[0], # "minmax" - "gamma": 1.0, "strategy": WEIGHT_STRATEGIES[0], # "none" - "softmax_temperature": 1.0, - "tanh_scale": 1.0, - "tanh_gain": 1.0, + # Phase 1: Standardization + "standardization": STANDARDIZATION_TYPES[0], # "none" "robust_quantiles": (0.25, 0.75), + # Phase 2: Normalization + "normalization": NORMALIZATION_TYPES[0], # "minmax" + "minmax_range": (0.0, 1.0), + "sigmoid_scale": 1.0, + "softmax_temperature": 1.0, "rank_method": RANK_METHODS[0], # "average" + # Phase 3: Post-processing + "gamma": 1.0, } DEFAULT_EXTREMA_WEIGHT: Final[float] = 1.0 @@ -200,54 +207,130 @@ def smooth_extrema( ) -def _normalize_zscore( - weights: NDArray[np.floating], - rescale_to_unit_range: bool = True, -) -> NDArray[np.floating]: +def _standardize_zscore(weights: NDArray[np.floating]) -> NDArray[np.floating]: + """ + Z-score standardization: (w - μ) / σ + Returns: mean≈0, std≈1 + """ if weights.size == 0: return weights weights = weights.astype(float, copy=False) if np.isnan(weights).any(): - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) + return np.zeros_like(weights, dtype=float) if weights.size == 1 or np.allclose(weights, weights[0]): - if rescale_to_unit_range: - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - else: - return np.zeros_like(weights, dtype=float) + return np.zeros_like(weights, dtype=float) try: z_scores = sp.stats.zscore(weights, ddof=1, nan_policy="raise") except Exception: - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) + return np.zeros_like(weights, dtype=float) if np.isnan(z_scores).any() or not np.isfinite(z_scores).all(): + return np.zeros_like(weights, dtype=float) + + return z_scores + + +def _standardize_robust( + weights: NDArray[np.floating], + quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"], +) -> NDArray[np.floating]: + """ + Robust standardization: (w - median) / IQR + Returns: median≈0, IQR≈1 (outlier-resistant) + """ + weights = weights.astype(float, copy=False) + if np.isnan(weights).any(): + return np.zeros_like(weights, dtype=float) + + median = np.median(weights) + q_low, q_high = np.quantile(weights, quantiles) + iqr = q_high - q_low + + if np.isclose(iqr, 0.0): + return np.zeros_like(weights, dtype=float) + + return (weights - median) / iqr + + +def standardize_weights( + weights: NDArray[np.floating], + method: StandardizationType = STANDARDIZATION_TYPES[0], + robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ + "robust_quantiles" + ], +) -> NDArray[np.floating]: + """ + Phase 1: Standardize weights (centering/scaling, not [0,1] mapping). + Methods: "none", "zscore", "robust" + """ + if weights.size == 0: + return weights + + if method == STANDARDIZATION_TYPES[0]: # "none" + return weights + + elif method == STANDARDIZATION_TYPES[1]: # "zscore" + return _standardize_zscore(weights) + + elif method == STANDARDIZATION_TYPES[2]: # "robust" + return _standardize_robust(weights, quantiles=robust_quantiles) + + else: + raise ValueError(f"Unknown standardization method: {method}") + + +def _normalize_sigmoid( + weights: NDArray[np.floating], + scale: float = DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"], +) -> NDArray[np.floating]: + """ + Sigmoid normalization: 1 / (1 + exp(-scale × w)) + Returns: [0, 1] with soft compression + """ + weights = weights.astype(float, copy=False) + if np.isnan(weights).any(): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - if not rescale_to_unit_range: - return z_scores + if scale <= 0 or not np.isfinite(scale): + scale = 1.0 - return _normalize_minmax(z_scores) + scaled = scale * weights + return sp.special.expit(scaled) -def _normalize_minmax(weights: NDArray[np.floating]) -> NDArray[np.floating]: +def _normalize_minmax( + weights: NDArray[np.floating], + range: tuple[float, float] = (0.0, 1.0), +) -> NDArray[np.floating]: + """ + MinMax normalization: range_min + [(w - min) / (max - min)] × (range_max - range_min) + Returns: [range_min, range_max] + """ weights = weights.astype(float, copy=False) if np.isnan(weights).any(): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) + w_min = np.min(weights) w_max = np.max(weights) + if not (np.isfinite(w_min) and np.isfinite(w_max)): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) + w_range = w_max - w_min if np.isclose(w_range, 0.0): - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - normalized_weights = (weights - w_min) / w_range - return normalized_weights + range_midpoint = midpoint(range[0], range[1]) + return np.full_like(weights, range_midpoint, dtype=float) + + normalized = (weights - w_min) / w_range + return range[0] + normalized * (range[1] - range[0]) def _normalize_l1(weights: NDArray[np.floating]) -> NDArray[np.floating]: + """L1 normalization: w / Σ|w| → Σ|w| = 1""" weights_sum = np.sum(np.abs(weights)) if weights_sum <= 0 or not np.isfinite(weights_sum): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) @@ -256,6 +339,7 @@ def _normalize_l1(weights: NDArray[np.floating]) -> NDArray[np.floating]: def _normalize_l2(weights: NDArray[np.floating]) -> NDArray[np.floating]: + """L2 normalization: w / ||w||₂ → ||w||₂ = 1""" weights = weights.astype(float, copy=False) if np.isnan(weights).any(): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) @@ -269,29 +353,11 @@ def _normalize_l2(weights: NDArray[np.floating]) -> NDArray[np.floating]: return normalized_weights -def _normalize_robust( - weights: NDArray[np.floating], - quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"], -) -> NDArray[np.floating]: - weights = weights.astype(float, copy=False) - if np.isnan(weights).any(): - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - - median = np.median(weights) - q_low, q_high = np.quantile(weights, quantiles) - iqr = q_high - q_low - - if np.isclose(iqr, 0.0): - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - - robust_weights = (weights - median) / iqr - return _normalize_minmax(robust_weights) - - def _normalize_softmax( weights: NDArray[np.floating], temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], ) -> NDArray[np.floating]: + """Softmax normalization: exp(w/T) / Σexp(w/T) → Σw = 1, range [0,1]""" weights = weights.astype(float, copy=False) if np.isnan(weights).any(): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) @@ -300,24 +366,11 @@ def _normalize_softmax( return sp.special.softmax(weights) -def _normalize_tanh( - weights: NDArray[np.floating], - scale: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"], - gain: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"], -) -> NDArray[np.floating]: - weights = weights.astype(float, copy=False) - if np.isnan(weights).any(): - return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) - - z_scores = _normalize_zscore(weights, rescale_to_unit_range=False) - normalized_weights = gain * 0.5 * (np.tanh(scale * z_scores) + 1.0) - return normalized_weights - - def _normalize_rank( weights: NDArray[np.floating], method: RankMethod = DEFAULTS_EXTREMA_WEIGHTING["rank_method"], ) -> NDArray[np.floating]: + """Rank normalization: [rank(w) - 1] / (n - 1) → [0, 1] uniformly distributed""" weights = weights.astype(float, copy=False) if np.isnan(weights).any(): return np.full_like(weights, float(DEFAULT_EXTREMA_WEIGHT), dtype=float) @@ -333,52 +386,68 @@ def _normalize_rank( def normalize_weights( weights: NDArray[np.floating], - normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], - gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], - softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], - tanh_scale: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"], - tanh_gain: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"], + # Phase 1: Standardization + standardization: StandardizationType = DEFAULTS_EXTREMA_WEIGHTING[ + "standardization" + ], robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + # Phase 2: Normalization + normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], + minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], + sigmoid_scale: float = DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"], + softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], rank_method: RankMethod = DEFAULTS_EXTREMA_WEIGHTING["rank_method"], + # Phase 3: Post-processing + gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], ) -> NDArray[np.floating]: + """ + 3-phase weight normalization: + 1. Standardization: zscore (w-μ)/σ | robust (w-median)/IQR | none + 2. Normalization: minmax, sigmoid, softmax, l1, l2, rank, none + 3. Post-processing: gamma correction w^γ + """ if weights.size == 0: return weights - if normalization == NORMALIZATION_TYPES[8]: # "none" - return weights - normalized_weights: NDArray[np.floating] - - if normalization == NORMALIZATION_TYPES[0]: # "minmax" - normalized_weights = _normalize_minmax(weights) - - elif normalization == NORMALIZATION_TYPES[1]: # "zscore" - normalized_weights = _normalize_zscore(weights, rescale_to_unit_range=True) + # Phase 1: Standardization + standardized_weights = standardize_weights( + weights, + method=standardization, + robust_quantiles=robust_quantiles, + ) - elif normalization == NORMALIZATION_TYPES[2]: # "l1" - normalized_weights = _normalize_l1(weights) + # Phase 2: Normalization + if normalization == NORMALIZATION_TYPES[6]: # "none" + normalized_weights = standardized_weights - elif normalization == NORMALIZATION_TYPES[3]: # "l2" - normalized_weights = _normalize_l2(weights) + elif normalization == NORMALIZATION_TYPES[0]: # "minmax" + normalized_weights = _normalize_minmax(standardized_weights, range=minmax_range) - elif normalization == NORMALIZATION_TYPES[4]: # "robust" - normalized_weights = _normalize_robust(weights, quantiles=robust_quantiles) + elif normalization == NORMALIZATION_TYPES[1]: # "sigmoid" + normalized_weights = _normalize_sigmoid( + standardized_weights, scale=sigmoid_scale + ) - elif normalization == NORMALIZATION_TYPES[5]: # "softmax" + elif normalization == NORMALIZATION_TYPES[2]: # "softmax" normalized_weights = _normalize_softmax( - weights, temperature=softmax_temperature + standardized_weights, temperature=softmax_temperature ) - elif normalization == NORMALIZATION_TYPES[6]: # "tanh" - normalized_weights = _normalize_tanh(weights, scale=tanh_scale, gain=tanh_gain) + elif normalization == NORMALIZATION_TYPES[3]: # "l1" + normalized_weights = _normalize_l1(standardized_weights) + + elif normalization == NORMALIZATION_TYPES[4]: # "l2" + normalized_weights = _normalize_l2(standardized_weights) - elif normalization == NORMALIZATION_TYPES[7]: # "rank" - normalized_weights = _normalize_rank(weights, method=rank_method) + elif normalization == NORMALIZATION_TYPES[5]: # "rank" + normalized_weights = _normalize_rank(standardized_weights, method=rank_method) else: raise ValueError(f"Unknown normalization method: {normalization}") + # Phase 3: Post-processing if not np.isclose(gamma, 1.0) and np.isfinite(gamma) and gamma > 0: normalized_weights = np.power(np.abs(normalized_weights), gamma) * np.sign( normalized_weights @@ -394,16 +463,26 @@ def calculate_extrema_weights( series: pd.Series, indices: list[int], weights: NDArray[np.floating], - normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], - gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], - softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], - tanh_scale: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"], - tanh_gain: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"], + # Phase 1: Standardization + standardization: StandardizationType = DEFAULTS_EXTREMA_WEIGHTING[ + "standardization" + ], robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + # Phase 2: Normalization + normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], + minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], + sigmoid_scale: float = DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"], + softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], rank_method: RankMethod = DEFAULTS_EXTREMA_WEIGHTING["rank_method"], + # Phase 3: Post-processing + gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], ) -> pd.Series: + """ + Calculate normalized weights for extrema points. + Returns: Series with weights at extrema indices (rest filled with default). + """ if len(indices) == 0 or len(weights) == 0: return pd.Series(float(DEFAULT_EXTREMA_WEIGHT), index=series.index) @@ -414,13 +493,14 @@ def calculate_extrema_weights( normalized_weights = normalize_weights( weights, - normalization, - gamma, - softmax_temperature, - tanh_scale, - tanh_gain, - robust_quantiles, - rank_method, + standardization=standardization, + robust_quantiles=robust_quantiles, + normalization=normalization, + minmax_range=minmax_range, + sigmoid_scale=sigmoid_scale, + softmax_temperature=softmax_temperature, + rank_method=rank_method, + gamma=gamma, ) if normalized_weights.size == 0 or np.allclose( @@ -444,16 +524,42 @@ def get_weighted_extrema( indices: list[int], weights: NDArray[np.floating], strategy: WeightStrategy = DEFAULTS_EXTREMA_WEIGHTING["strategy"], - normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], - gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], - softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], - tanh_scale: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_scale"], - tanh_gain: float = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"], + # Phase 1: Standardization + standardization: StandardizationType = DEFAULTS_EXTREMA_WEIGHTING[ + "standardization" + ], robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + # Phase 2: Normalization + normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], + minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], + sigmoid_scale: float = DEFAULTS_EXTREMA_WEIGHTING["sigmoid_scale"], + softmax_temperature: float = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"], rank_method: RankMethod = DEFAULTS_EXTREMA_WEIGHTING["rank_method"], + # Phase 3: Post-processing + gamma: float = DEFAULTS_EXTREMA_WEIGHTING["gamma"], ) -> tuple[pd.Series, pd.Series]: + """ + Apply weighted normalization to extrema series. + + Args: + extrema: Extrema series + indices: Indices of extrema points + weights: Raw weights for each extremum + strategy: Weight strategy ("none", "amplitude", "amplitude_threshold_ratio") + standardization: Standardization method + robust_quantiles: Quantiles for robust standardization + normalization: Normalization method + minmax_range: Target range for minmax + sigmoid_scale: Scale for sigmoid + softmax_temperature: Temperature for softmax + rank_method: Method for rank normalization + gamma: Gamma correction + + Returns: + Tuple of (weighted_extrema, extrema_weights) + """ default_weights = pd.Series(float(DEFAULT_EXTREMA_WEIGHT), index=extrema.index) if ( len(indices) == 0 or len(weights) == 0 or strategy == WEIGHT_STRATEGIES[0] @@ -468,13 +574,14 @@ def get_weighted_extrema( series=extrema, indices=indices, weights=weights, + standardization=standardization, + robust_quantiles=robust_quantiles, normalization=normalization, - gamma=gamma, + minmax_range=minmax_range, + sigmoid_scale=sigmoid_scale, softmax_temperature=softmax_temperature, - tanh_scale=tanh_scale, - tanh_gain=tanh_gain, - robust_quantiles=robust_quantiles, rank_method=rank_method, + gamma=gamma, ) if np.allclose(extrema_weights, DEFAULT_EXTREMA_WEIGHT): return extrema, default_weights -- 2.43.0