From 1322ec21d923b630f7f9484286e3bfaf63ef7781 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 9 Dec 2025 00:43:51 +0100 Subject: [PATCH] feat(qav3): add MMAD standardization for extrema weighting MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- README.md | 3 +- .../user_data/strategies/QuickAdapterV3.py | 17 ++++++++ quickadapter/user_data/strategies/Utils.py | 41 +++++++++++++++++-- 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 009b6e1..e0bb105 100644 --- a/README.md +++ b/README.md @@ -66,8 +66,9 @@ docker compose up -d --build | freqai.extrema_smoothing.bandwidth | 1.0 | float > 0 | Gaussian bandwidth for `nadaraya_watson`. | | _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.standardization | `none` | enum {`none`,`zscore`,`robust`} | Standardization method applied before normalization. `none`=no standardization, `zscore`=(w-μ)/σ, `robust`=(w-median)/IQR. | +| freqai.extrema_weighting.standardization | `none` | enum {`none`,`zscore`,`robust`,`mmad`} | Standardization method applied before normalization. `none`=no standardization, `zscore`=(w-μ)/σ, `robust`=(w-median)/IQR, `mmad`=(w-median)/MAD. | | freqai.extrema_weighting.robust_quantiles | [0.25, 0.75] | list[float] where 0 <= Q1 < Q3 <= 1 | Quantile range for robust standardization, Q1 and Q3. | +| freqai.extrema_weighting.mmad_scaling_factor | 1.4826 | float > 0 | Scaling factor for MMAD standardization. | | 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] | Target range for `minmax` normalization, min and max. | | freqai.extrema_weighting.sigmoid_scale | 1.0 | float > 0 | Scale parameter for `sigmoid` normalization, controls steepness. | diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index 63a63f6..0750df1 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -674,6 +674,21 @@ class QuickAdapterV3(IStrategy): float(weighting_robust_quantiles[1]), ) + weighting_mmad_scaling_factor = extrema_weighting.get( + "mmad_scaling_factor", DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"] + ) + if ( + not isinstance(weighting_mmad_scaling_factor, (int, float)) + or not np.isfinite(weighting_mmad_scaling_factor) + or weighting_mmad_scaling_factor <= 0 + ): + logger.warning( + f"{pair}: invalid extrema_weighting mmad_scaling_factor {weighting_mmad_scaling_factor}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['mmad_scaling_factor']}" + ) + weighting_mmad_scaling_factor = DEFAULTS_EXTREMA_WEIGHTING[ + "mmad_scaling_factor" + ] + # Phase 2: Normalization weighting_normalization = str( extrema_weighting.get( @@ -765,6 +780,7 @@ class QuickAdapterV3(IStrategy): "strategy": weighting_strategy, "standardization": weighting_standardization, "robust_quantiles": weighting_robust_quantiles, + "mmad_scaling_factor": weighting_mmad_scaling_factor, "normalization": weighting_normalization, "minmax_range": weighting_minmax_range, "sigmoid_scale": weighting_sigmoid_scale, @@ -930,6 +946,7 @@ class QuickAdapterV3(IStrategy): strategy=self.extrema_weighting["strategy"], standardization=self.extrema_weighting["standardization"], robust_quantiles=self.extrema_weighting["robust_quantiles"], + mmad_scaling_factor=self.extrema_weighting["mmad_scaling_factor"], normalization=self.extrema_weighting["normalization"], minmax_range=self.extrema_weighting["minmax_range"], sigmoid_scale=self.extrema_weighting["sigmoid_scale"], diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 480b719..b701349 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -30,11 +30,12 @@ EXTREMA_COLUMN: Final = "&s-extrema" MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold" MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold" -StandardizationType = Literal["none", "zscore", "robust"] +StandardizationType = Literal["none", "zscore", "robust", "mmad"] STANDARDIZATION_TYPES: Final[tuple[StandardizationType, ...]] = ( "none", # 0 - No standardization "zscore", # 1 - (w - μ) / σ "robust", # 2 - (w - median) / IQR + "mmad", # 3 - (w - median) / MAD ) NormalizationType = Literal["minmax", "sigmoid", "softmax", "l1", "l2", "rank", "none"] @@ -95,6 +96,7 @@ DEFAULTS_EXTREMA_WEIGHTING: Final[dict[str, Any]] = { # Phase 1: Standardization "standardization": STANDARDIZATION_TYPES[0], # "none" "robust_quantiles": (0.25, 0.75), + "mmad_scaling_factor": 1.4826, # Phase 2: Normalization "normalization": NORMALIZATION_TYPES[0], # "minmax" "minmax_range": (0.0, 1.0), @@ -135,6 +137,7 @@ def get_gaussian_std(window: int) -> float: return (window - 1) / 6.0 if window > 1 else 0.5 +@lru_cache(maxsize=8) def get_savgol_params( window: int, polyorder: int, mode: SmoothingMode ) -> tuple[int, int, str]: @@ -317,16 +320,38 @@ def _standardize_robust( return (weights - median) / iqr +def _standardize_mmad( + weights: NDArray[np.floating], + scaling_factor: float = DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"], +) -> NDArray[np.floating]: + """ + MMAD standardization: (w - median) / MAD + Returns: median≈0, MAD≈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) + mad = np.median(np.abs(weights - median)) + + if np.isclose(mad, 0.0): + return np.zeros_like(weights, dtype=float) + + return (weights - median) / (scaling_factor * mad) + + def standardize_weights( weights: NDArray[np.floating], method: StandardizationType = STANDARDIZATION_TYPES[0], robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + mmad_scaling_factor: float = DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"], ) -> NDArray[np.floating]: """ Phase 1: Standardize weights (centering/scaling, not [0,1] mapping). - Methods: "none", "zscore", "robust" + Methods: "none", "zscore", "robust", "mmad" """ if weights.size == 0: return weights @@ -340,6 +365,9 @@ def standardize_weights( elif method == STANDARDIZATION_TYPES[2]: # "robust" return _standardize_robust(weights, quantiles=robust_quantiles) + elif method == STANDARDIZATION_TYPES[3]: # "mmad" + return _standardize_mmad(weights, scaling_factor=mmad_scaling_factor) + else: raise ValueError(f"Unknown standardization method: {method}") @@ -454,6 +482,7 @@ def normalize_weights( robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + mmad_scaling_factor: float = DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"], # Phase 2: Normalization normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], @@ -465,7 +494,7 @@ def normalize_weights( ) -> NDArray[np.floating]: """ 3-phase weight normalization: - 1. Standardization: zscore (w-μ)/σ | robust (w-median)/IQR | none + 1. Standardization: zscore (w-μ)/σ | robust (w-median)/IQR | mmad (w-median)/MAD | none 2. Normalization: minmax, sigmoid, softmax, l1, l2, rank, none 3. Post-processing: gamma correction w^γ """ @@ -477,6 +506,7 @@ def normalize_weights( weights, method=standardization, robust_quantiles=robust_quantiles, + mmad_scaling_factor=mmad_scaling_factor, ) # Phase 2: Normalization @@ -531,6 +561,7 @@ def calculate_extrema_weights( robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + mmad_scaling_factor: float = DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"], # Phase 2: Normalization normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], @@ -556,6 +587,7 @@ def calculate_extrema_weights( weights, standardization=standardization, robust_quantiles=robust_quantiles, + mmad_scaling_factor=mmad_scaling_factor, normalization=normalization, minmax_range=minmax_range, sigmoid_scale=sigmoid_scale, @@ -592,6 +624,7 @@ def get_weighted_extrema( robust_quantiles: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING[ "robust_quantiles" ], + mmad_scaling_factor: float = DEFAULTS_EXTREMA_WEIGHTING["mmad_scaling_factor"], # Phase 2: Normalization normalization: NormalizationType = DEFAULTS_EXTREMA_WEIGHTING["normalization"], minmax_range: tuple[float, float] = DEFAULTS_EXTREMA_WEIGHTING["minmax_range"], @@ -611,6 +644,7 @@ def get_weighted_extrema( strategy: Weight strategy ("none", "amplitude", "amplitude_threshold_ratio") standardization: Standardization method robust_quantiles: Quantiles for robust standardization + mmad_scaling_factor: Scaling factor for MMAD standardization normalization: Normalization method minmax_range: Target range for minmax sigmoid_scale: Scale for sigmoid @@ -637,6 +671,7 @@ def get_weighted_extrema( weights=weights, standardization=standardization, robust_quantiles=robust_quantiles, + mmad_scaling_factor=mmad_scaling_factor, normalization=normalization, minmax_range=minmax_range, sigmoid_scale=sigmoid_scale, -- 2.43.0