| _Extrema weighting_ | | | |
| freqai.extrema_weighting.strategy | `none` | enum {`none`,`amplitude`,`amplitude_threshold_ratio`,`volume_rate`,`speed`,`efficiency_ratio`,`volume_weighted_efficiency_ratio`,`combined`} | Extrema weighting metric: none (`none`), swing amplitude (`amplitude`), swing amplitude / median volatility-threshold ratio (`amplitude_threshold_ratio`), swing volume per candle (`volume_rate`), swing speed (`speed`), swing efficiency ratio (`efficiency_ratio`), swing volume-weighted efficiency ratio (`volume_weighted_efficiency_ratio`), or combined metrics aggregation (`combined`). Switching between `none` and any other strategy requires deleting trained models. |
| freqai.extrema_weighting.metric_coefficients | {} | dict[str, float] | Per-metric coefficients for `combined` strategy. Keys: `amplitude`, `amplitude_threshold_ratio`, `volume_rate`, `speed`, `efficiency_ratio`, `volume_weighted_efficiency_ratio`. |
-| freqai.extrema_weighting.aggregation | `weighted_average` | enum {`weighted_average`,`geometric_mean`} | Metric aggregation method for `combined` strategy. `weighted_average`=Σ(coef·metric)/Σ(coef), `geometric_mean`=∏(metric^coef)^(1/Σcoef). |
+| freqai.extrema_weighting.aggregation | `arithmetic_mean` | enum {`arithmetic_mean`,`geometric_mean`,`harmonic_mean`,`quadratic_mean`,`weighted_median`,`softmax`} | Metric aggregation method for `combined` strategy. `arithmetic_mean`=(Σ(w·m)/Σ(w)), `geometric_mean`=(∏(m^w))^(1/Σw), `harmonic_mean`=Σ(w)/(Σ(w/m)), `quadratic_mean`=(Σ(w·m²)/Σ(w))^(1/2), `weighted_median`=Q₀.₅(m,w), `softmax`=Σ(m·s_i) where s_i=w_i·exp(m_i/T)/Σ(w_j·exp(m_j/T)). |
+| freqai.extrema_weighting.softmax_temperature | 1.0 | float > 0 | Temperature T for `softmax` aggregation, controls distribution sharpness. |
| freqai.extrema_weighting.standardization | `none` | enum {`none`,`zscore`,`robust`,`mmad`,`power_yj`} | Standardization method applied to smoothed weighted extrema before normalization. `none`=w, `zscore`=(w-μ)/σ, `robust`=(w-median)/IQR, `mmad`=(w-median)/(MAD·k), `power_yj`=YJ(w). |
| 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. |
"volume_weighted_efficiency_ratio",
)
-CombinedAggregation = Literal["weighted_average", "geometric_mean"]
+CombinedAggregation = Literal[
+ "arithmetic_mean",
+ "geometric_mean",
+ "harmonic_mean",
+ "quadratic_mean",
+ "weighted_median",
+ "softmax",
+]
COMBINED_AGGREGATIONS: Final[tuple[CombinedAggregation, ...]] = (
- "weighted_average",
+ "arithmetic_mean",
"geometric_mean",
+ "harmonic_mean",
+ "quadratic_mean",
+ "weighted_median",
+ "softmax",
)
WEIGHT_STRATEGIES: Final[tuple[WeightStrategy, ...]] = (
DEFAULTS_EXTREMA_WEIGHTING: Final[dict[str, Any]] = {
"strategy": WEIGHT_STRATEGIES[0], # "none"
"metric_coefficients": {},
- "aggregation": COMBINED_AGGREGATIONS[0], # "weighted_average"
+ "aggregation": COMBINED_AGGREGATIONS[0], # "arithmetic_mean"
+ "softmax_temperature": 1.0,
# Phase 1: Standardization
"standardization": STANDARDIZATION_TYPES[0], # "none"
"robust_quantiles": (0.25, 0.75),
)
aggregation = DEFAULTS_EXTREMA_WEIGHTING["aggregation"]
+ softmax_temperature = extrema_weighting.get(
+ "softmax_temperature", DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"]
+ )
+ if (
+ not isinstance(softmax_temperature, (int, float))
+ or not np.isfinite(softmax_temperature)
+ or softmax_temperature <= 0
+ ):
+ logger.warning(
+ f"Invalid extrema_weighting softmax_temperature value {softmax_temperature!r}: must be a finite number > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['softmax_temperature']!r}"
+ )
+ softmax_temperature = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"]
+
# Phase 1: Standardization
standardization = extrema_weighting.get(
"standardization", DEFAULTS_EXTREMA_WEIGHTING["standardization"]
"strategy": strategy,
"metric_coefficients": metric_coefficients,
"aggregation": aggregation,
+ "softmax_temperature": softmax_temperature,
# Phase 1: Standardization
"standardization": standardization,
"robust_quantiles": robust_quantiles,
stacked_metrics: NDArray[np.floating],
coefficients: NDArray[np.floating],
aggregation: CombinedAggregation,
+ softmax_temperature: float,
) -> NDArray[np.floating]:
- if aggregation == COMBINED_AGGREGATIONS[0]: # "weighted_average"
- return np.average(stacked_metrics, axis=0, weights=coefficients)
+ if aggregation == COMBINED_AGGREGATIONS[0]: # "arithmetic_mean"
+ return sp.stats.pmean(stacked_metrics.T, p=1.0, weights=coefficients, axis=1)
elif aggregation == COMBINED_AGGREGATIONS[1]: # "geometric_mean"
- return np.asarray(
- sp.stats.gmean(stacked_metrics.T, weights=coefficients, axis=1),
- dtype=float,
+ return sp.stats.pmean(stacked_metrics.T, p=0.0, weights=coefficients, axis=1)
+ elif aggregation == COMBINED_AGGREGATIONS[2]: # "harmonic_mean"
+ return sp.stats.pmean(stacked_metrics.T, p=-1.0, weights=coefficients, axis=1)
+ elif aggregation == COMBINED_AGGREGATIONS[3]: # "quadratic_mean"
+ return sp.stats.pmean(stacked_metrics.T, p=2.0, weights=coefficients, axis=1)
+ elif aggregation == COMBINED_AGGREGATIONS[4]: # "weighted_median"
+ return np.array(
+ [
+ np.quantile(
+ stacked_metrics[:, i],
+ 0.5,
+ weights=coefficients,
+ method="inverted_cdf",
+ )
+ for i in range(stacked_metrics.shape[1])
+ ]
)
+ elif aggregation == COMBINED_AGGREGATIONS[5]: # "softmax"
+ scaled_metrics = stacked_metrics / softmax_temperature
+ softmax_weights = sp.special.softmax(scaled_metrics, axis=0)
+ combined_weights = softmax_weights * coefficients[:, np.newaxis]
+ combined_weights = combined_weights / np.sum(
+ combined_weights, axis=0, keepdims=True
+ )
+ return np.sum(stacked_metrics * combined_weights, axis=0)
else:
raise ValueError(
f"Invalid aggregation value {aggregation!r}: supported values are {', '.join(COMBINED_AGGREGATIONS)}"
volume_weighted_efficiency_ratios: list[float],
metric_coefficients: dict[str, Any],
aggregation: CombinedAggregation,
+ softmax_temperature: float,
) -> NDArray[np.floating]:
if len(indices) == 0:
return np.asarray([], dtype=float)
stacked_metrics = np.vstack(imputed_metrics)
coefficients_array = np.asarray(coefficients_list, dtype=float)
- return _aggregate_metrics(stacked_metrics, coefficients_array, aggregation)
+ return _aggregate_metrics(
+ stacked_metrics, coefficients_array, aggregation, softmax_temperature
+ )
def compute_extrema_weights(
volume_weighted_efficiency_ratios=volume_weighted_efficiency_ratios,
metric_coefficients=extrema_weighting["metric_coefficients"],
aggregation=extrema_weighting["aggregation"],
+ softmax_temperature=extrema_weighting["softmax_temperature"],
)
else:
repositories / freqai-strategies.git / commitdiff
