return label_frequency_candles
+ @cached_property
+ def predictions_extrema(self) -> dict[str, Any]:
+ predictions_extrema = self.freqai_info.get("predictions_extrema", {})
+ if not isinstance(predictions_extrema, dict):
+ predictions_extrema = {}
+
+ threshold_outlier = predictions_extrema.get("threshold_outlier", 0.999)
+ if not isinstance(threshold_outlier, (int, float)) or not (
+ 0 < threshold_outlier < 1
+ ):
+ threshold_outlier = 0.999
+
+ selection_method = str(
+ predictions_extrema.get(
+ "selection_method",
+ QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS[0], # "rank"
+ )
+ )
+ if (
+ selection_method
+ not in QuickAdapterRegressorV3._extrema_selection_methods_set()
+ ):
+ selection_method = QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS[0]
+
+ thresholds_smoothing = str(
+ predictions_extrema.get(
+ "thresholds_smoothing",
+ QuickAdapterRegressorV3._THRESHOLD_METHODS[0], # "mean"
+ )
+ )
+ if thresholds_smoothing not in QuickAdapterRegressorV3._threshold_methods_set():
+ thresholds_smoothing = QuickAdapterRegressorV3._THRESHOLD_METHODS[
+ 0
+ ] # "mean"
+
+ thresholds_alpha = predictions_extrema.get("thresholds_alpha", 12.0)
+ if not isinstance(thresholds_alpha, (int, float)) or thresholds_alpha < 0:
+ thresholds_alpha = 12.0
+
+ return {
+ "threshold_outlier": float(threshold_outlier),
+ "selection_method": selection_method,
+ "thresholds_smoothing": thresholds_smoothing,
+ "thresholds_alpha": float(thresholds_alpha),
+ }
+
@property
def _optuna_label_candle_pool_full(self) -> list[int]:
label_frequency_candles = self._label_frequency_candles
f = sp.stats.weibull_min.fit(
pd.to_numeric(di_values, errors="coerce").dropna()
)
- predictions_extrema = self.freqai_info.get("predictions_extrema", {})
cutoff = sp.stats.weibull_min.ppf(
- predictions_extrema.get("threshold_outlier", 0.999), *f
+ self.predictions_extrema["threshold_outlier"], *f
)
dk.data["DI_value_mean"] = di_values.mean()
pred_extrema = pred_df.get(EXTREMA_COLUMN).iloc[-thresholds_candles:].copy()
- predictions_extrema = self.freqai_info.get("predictions_extrema", {})
- extrema_selection = str(
- predictions_extrema.get(
- "selection_method",
- QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS[0], # "rank"
- )
- )
- if (
- extrema_selection
- not in QuickAdapterRegressorV3._extrema_selection_methods_set()
- ):
- raise ValueError(
- f"Unsupported extrema selection method: {extrema_selection}. "
- f"Supported methods are {', '.join(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)}"
- )
- thresholds_smoothing = str(
- predictions_extrema.get(
- "thresholds_smoothing",
- QuickAdapterRegressorV3._THRESHOLD_METHODS[0], # "mean"
- )
- )
- if thresholds_smoothing not in QuickAdapterRegressorV3._threshold_methods_set():
- raise ValueError(
- f"Unsupported thresholds smoothing method: {thresholds_smoothing}. "
- f"Supported methods are {', '.join(QuickAdapterRegressorV3._THRESHOLD_METHODS)}"
- )
+ # Use cached parameters
+ extrema_selection = self.predictions_extrema["selection_method"]
+ thresholds_smoothing = self.predictions_extrema["thresholds_smoothing"]
+
if (
thresholds_smoothing == QuickAdapterRegressorV3._THRESHOLD_METHODS[7]
): # "median"
elif (
thresholds_smoothing == QuickAdapterRegressorV3._THRESHOLD_METHODS[8]
): # "soft_extremum"
- thresholds_alpha = float(predictions_extrema.get("thresholds_alpha", 12.0))
return QuickAdapterRegressorV3.soft_extremum_min_max(
- pred_extrema, thresholds_alpha, extrema_selection
+ pred_extrema,
+ self.predictions_extrema["thresholds_alpha"],
+ extrema_selection,
)
elif (
thresholds_smoothing
if n_uniques <= 3:
return min(n_uniques, upper_bound)
n_clusters = int(round((np.log2(n_uniques) + np.sqrt(n_uniques)) / 2.0))
- return max(lower_bound, min(n_clusters, upper_bound))
+ return min(max(lower_bound, n_clusters), upper_bound)
def _calculate_distances_to_ideal(
self,
else:
return max_open_trades
+ @cached_property
+ def extrema_weighting(self) -> dict[str, Any]:
+ extrema_weighting = self.freqai_info.get("extrema_weighting", {})
+ if not isinstance(extrema_weighting, dict):
+ extrema_weighting = {}
+ return QuickAdapterV3._get_extrema_weighting_params(extrema_weighting, {})
+
+ @cached_property
+ def extrema_smoothing(self) -> dict[str, Any]:
+ extrema_smoothing = self.freqai_info.get("extrema_smoothing", {})
+ if not isinstance(extrema_smoothing, dict):
+ extrema_smoothing = {}
+ return QuickAdapterV3._get_extrema_smoothing_params(extrema_smoothing, {})
+
def bot_start(self, **kwargs) -> None:
self.pairs: list[str] = self.config.get("exchange", {}).get("pair_whitelist")
if not self.pairs:
"lookback_period"
]
- if not isinstance(decay_ratio, (int, float)) or not (
- 0.0 < float(decay_ratio) <= 1.0
- ):
+ if not isinstance(decay_ratio, (int, float)) or not (0.0 < decay_ratio <= 1.0):
logger.warning(
f"reversal_confirmation: invalid decay_ratio {decay_ratio!r}, using default {QuickAdapterV3.default_reversal_confirmation['decay_ratio']}"
)
decay_ratio = QuickAdapterV3.default_reversal_confirmation["decay_ratio"]
- else:
- decay_ratio = float(decay_ratio)
min_natr_ratio_percent, max_natr_ratio_percent = validate_range(
min_natr_ratio_percent,
if (
not isinstance(weighting_gamma, (int, float))
or not np.isfinite(weighting_gamma)
- or not (0 < float(weighting_gamma) <= 10.0)
+ or not (0 < weighting_gamma <= 10.0)
):
logger.warning(
f"{pair}: invalid extrema_weighting gamma {weighting_gamma}, must be a finite number in (0, 10], using default {DEFAULTS_EXTREMA_WEIGHTING['gamma']}"
)
weighting_gamma = DEFAULTS_EXTREMA_WEIGHTING["gamma"]
- else:
- weighting_gamma = float(weighting_gamma)
weighting_softmax_temperature = extrema_weighting.get(
"softmax_temperature", DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"]
logger.warning(
f"{pair}: invalid extrema_weighting softmax_temperature {weighting_softmax_temperature}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['softmax_temperature']}"
)
- weighting_softmax_temperature = DEFAULTS_EXTREMA_WEIGHTING[
- "softmax_temperature"
- ]
- else:
- weighting_softmax_temperature = float(weighting_softmax_temperature)
+ weighting_softmax_temperature = DEFAULTS_EXTREMA_WEIGHTING["softmax_temperature"]
weighting_robust_quantiles = extrema_weighting.get(
"robust_quantiles", DEFAULTS_EXTREMA_WEIGHTING["robust_quantiles"]
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"]
- else:
- weighting_tanh_scale = float(weighting_tanh_scale)
weighting_tanh_gain = extrema_weighting.get(
"tanh_gain", DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"]
f"{pair}: invalid extrema_weighting tanh_gain {weighting_tanh_gain}, must be > 0, using default {DEFAULTS_EXTREMA_WEIGHTING['tanh_gain']}"
)
weighting_tanh_gain = DEFAULTS_EXTREMA_WEIGHTING["tanh_gain"]
- else:
- weighting_tanh_gain = float(weighting_tanh_gain)
return {
"strategy": weighting_strategy,
f"{pair}: invalid extrema_smoothing beta {smoothing_beta}, must be a finite number > 0, using default {DEFAULTS_EXTREMA_SMOOTHING['beta']}"
)
smoothing_beta = DEFAULTS_EXTREMA_SMOOTHING["beta"]
- else:
- smoothing_beta = float(smoothing_beta)
return {
"method": smoothing_method,
f"{pair}: labeled {len(pivots_indices)} extrema (label_period={QuickAdapterV3._td_format(label_period)} / {label_period_candles=} / {label_natr_ratio=:.2f})"
)
- extrema_weighting = self.freqai_info.get("extrema_weighting", {})
- if not isinstance(extrema_weighting, dict):
- extrema_weighting = {}
- extrema_weighting_params = QuickAdapterV3._get_extrema_weighting_params(
- extrema_weighting, pair
- )
-
pivot_weights = QuickAdapterV3._get_weights(
- extrema_weighting_params["strategy"],
+ self.extrema_weighting["strategy"],
pivots_amplitudes,
pivots_amplitude_threshold_ratios,
)
extrema=dataframe[EXTREMA_COLUMN],
indices=pivots_indices,
weights=np.array(pivot_weights),
- strategy=extrema_weighting_params["strategy"],
- normalization=extrema_weighting_params["normalization"],
- gamma=extrema_weighting_params["gamma"],
- softmax_temperature=extrema_weighting_params["softmax_temperature"],
- tanh_scale=extrema_weighting_params["tanh_scale"],
- tanh_gain=extrema_weighting_params["tanh_gain"],
- robust_quantiles=extrema_weighting_params["robust_quantiles"],
- rank_method=extrema_weighting_params["rank_method"],
- )
-
- extrema_smoothing = self.freqai_info.get("extrema_smoothing", {})
- if not isinstance(extrema_smoothing, dict):
- extrema_smoothing = {}
- extrema_smoothing_params = QuickAdapterV3._get_extrema_smoothing_params(
- extrema_smoothing, pair
+ strategy=self.extrema_weighting["strategy"],
+ normalization=self.extrema_weighting["normalization"],
+ gamma=self.extrema_weighting["gamma"],
+ 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"],
)
dataframe[EXTREMA_COLUMN] = smooth_extrema(
weighted_extrema,
- extrema_smoothing_params["method"],
- extrema_smoothing_params["window"],
- extrema_smoothing_params["beta"],
+ self.extrema_smoothing["method"],
+ self.extrema_smoothing["window"],
+ self.extrema_smoothing["beta"],
)
if debug:
extrema = dataframe[EXTREMA_COLUMN]
repositories / freqai-strategies.git / commitdiff
