NORMALIZATION_TYPES,
RANK_METHODS,
SMOOTHING_METHODS,
+ STANDARDIZATION_TYPES,
WEIGHT_STRATEGIES,
TrendDirection,
WeightStrategy,
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"])
)
)
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"]
)
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"]
"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"]
)
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
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(
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"]
}
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
)
-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)
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)
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)
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)
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
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)
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(
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]
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
repositories / freqai-strategies.git / commitdiff
