Fold trade_price_target_method and reversal_confirmation inline validation onto the declarative spec-table pattern (get_exit_pricing_config, get_reversal_confirmation_config); coupled min/max NATR pair kept on validate_range. Remove the vestigial default_reversal_confirmation ClassVar. Guard np.isfinite via a shared _is_finite_value helper so every numeric validation path is total (no raise on ints >= 2**64 or non-scalar input). Cache config-derived validation properties (cached_property) in the strategy and regressor. Behavior preserved for all realistic inputs.
import warnings
from dataclasses import dataclass
from datetime import datetime, timezone
-from functools import lru_cache
+from functools import cached_property, lru_cache
from pathlib import Path
from typing import (
AbstractSet,
),
}
- @property
+ @cached_property
def _optuna_config(self) -> dict[str, Any]:
optuna_default_config = {
"enabled": False,
QuickAdapterRegressorV3.MAX_LABEL_NATR_MULTIPLIER_DEFAULT,
)
- @property
+ @cached_property
def _label_frequency_candles(self) -> int:
default_label_frequency_candles = max(2, 2 * len(self.pairs))
return label_frequency_candles
- @property
+ @cached_property
def label_weighting(self) -> dict[str, Any]:
label_weighting_raw = self.freqai_info.get("label_weighting")
if not isinstance(label_weighting_raw, dict):
label_weighting_raw = {}
return get_label_weighting_config(label_weighting_raw, logger)
- @property
+ @cached_property
def label_pipeline(self) -> dict[str, Any]:
label_pipeline_raw = self.freqai_info.get("label_pipeline")
if not isinstance(label_pipeline_raw, dict):
label_pipeline_raw = {}
return get_label_pipeline_config(label_pipeline_raw, logger)
- @property
+ @cached_property
def label_prediction(self) -> dict[str, Any]:
label_prediction_raw = self.freqai_info.get("label_prediction")
if not isinstance(label_prediction_raw, dict):
label_prediction_raw = {}
return get_label_prediction_config(label_prediction_raw, logger)
- @property
+ @cached_property
def _label_defaults(self) -> tuple[int, float]:
return get_label_defaults(self.ft_params, logger)
- @property
+ @cached_property
def _causal_mode(self) -> bool:
return get_causal_mode(self.ft_params, logger)
get_callable_sha256,
get_causal_mode,
get_distance,
+ get_exit_pricing_config,
get_label_defaults,
get_label_horizon_candles,
get_label_smoothing_config,
get_label_weighting_config,
+ get_reversal_confirmation_config,
get_smoothing_kernel_half_width,
get_zl_ma_fn,
is_finite_number,
safe_divide,
smooth,
top_log_return,
- validate_range,
vwapb,
weight_fill_radius,
zlema,
"decline_quantile": 0.5,
}
- default_reversal_confirmation: ClassVar[dict[str, int | float]] = {
- "lookback_period_candles": 0,
- "decay_fraction": 0.5,
- "min_natr_multiplier_fraction": 0.0095,
- "max_natr_multiplier_fraction": 0.0125,
- }
-
position_adjustment_enable = True
# {stage: (natr_multiplier_fraction, stake_percent, color)}
else:
return max_open_trades
- @property
+ @cached_property
def label_weighting(self) -> dict[str, Any]:
label_weighting_raw = self.freqai_info.get("label_weighting")
if not isinstance(label_weighting_raw, dict):
label_weighting_raw = {}
return get_label_weighting_config(label_weighting_raw, logger)
- @property
+ @cached_property
def label_smoothing(self) -> dict[str, Any]:
label_smoothing_raw = self.freqai_info.get("label_smoothing", {})
if not isinstance(label_smoothing_raw, dict):
label_smoothing_raw = {}
return get_label_smoothing_config(label_smoothing_raw, logger)
- @property
+ @cached_property
def trade_price_target_method(self) -> str:
exit_pricing = self.config.get("exit_pricing")
if not isinstance(exit_pricing, dict):
exit_pricing = {}
- trade_price_target_method = exit_pricing.get(
- "trade_price_target_method",
- TRADE_PRICE_TARGETS[0], # "moving_average"
- )
- if trade_price_target_method not in set(TRADE_PRICE_TARGETS):
- logger.warning(
- f"Invalid trade_price_target_method value {trade_price_target_method!r}: "
- f"supported values are {', '.join(TRADE_PRICE_TARGETS)}, "
- f"using default {TRADE_PRICE_TARGETS[0]!r}"
- )
- trade_price_target_method = TRADE_PRICE_TARGETS[0]
- return str(trade_price_target_method)
+ return get_exit_pricing_config(exit_pricing, logger)[
+ "trade_price_target_method"
+ ]
- @property
+ @cached_property
def reversal_confirmation(self) -> dict[str, int | float]:
reversal_confirmation = self.config.get("reversal_confirmation")
if not isinstance(reversal_confirmation, dict):
reversal_confirmation = {}
- defaults = QuickAdapterV3.default_reversal_confirmation
+ return get_reversal_confirmation_config(reversal_confirmation, logger)
- lookback_period_candles = reversal_confirmation.get(
- "lookback_period_candles", defaults["lookback_period_candles"]
- )
- decay_fraction = reversal_confirmation.get(
- "decay_fraction", defaults["decay_fraction"]
- )
- min_natr_multiplier_fraction = reversal_confirmation.get(
- "min_natr_multiplier_fraction", defaults["min_natr_multiplier_fraction"]
- )
- max_natr_multiplier_fraction = reversal_confirmation.get(
- "max_natr_multiplier_fraction", defaults["max_natr_multiplier_fraction"]
- )
-
- if not isinstance(lookback_period_candles, int) or lookback_period_candles < 0:
- logger.warning(
- f"Invalid reversal_confirmation lookback_period_candles value {lookback_period_candles!r}: must be >= 0, using default {QuickAdapterV3.default_reversal_confirmation['lookback_period_candles']!r}"
- )
- lookback_period_candles = QuickAdapterV3.default_reversal_confirmation[
- "lookback_period_candles"
- ]
-
- if not isinstance(decay_fraction, (int, float)) or not (
- 0.0 < decay_fraction <= 1.0
- ):
- logger.warning(
- f"Invalid reversal_confirmation decay_fraction value {decay_fraction!r}: must be in range (0, 1], using default {QuickAdapterV3.default_reversal_confirmation['decay_fraction']!r}"
- )
- decay_fraction = QuickAdapterV3.default_reversal_confirmation[
- "decay_fraction"
- ]
-
- min_natr_multiplier_fraction, max_natr_multiplier_fraction = validate_range(
- min_natr_multiplier_fraction,
- max_natr_multiplier_fraction,
- logger,
- name="natr_multiplier_fraction",
- default_min=QuickAdapterV3.default_reversal_confirmation[
- "min_natr_multiplier_fraction"
- ],
- default_max=QuickAdapterV3.default_reversal_confirmation[
- "max_natr_multiplier_fraction"
- ],
- allow_equal=False,
- non_negative=True,
- finite_only=True,
- )
-
- return {
- "lookback_period_candles": int(lookback_period_candles),
- "decay_fraction": float(decay_fraction),
- "min_natr_multiplier_fraction": float(min_natr_multiplier_fraction),
- "max_natr_multiplier_fraction": float(max_natr_multiplier_fraction),
- }
-
- @property
+ @cached_property
def _label_defaults(self) -> tuple[int, float]:
feature_parameters = self.freqai_info.get("feature_parameters", {})
return get_label_defaults(feature_parameters, logger)
return _safe_numeric_result(np.asarray(result, dtype=float), numerator, denominator)
+def _is_finite_value(value: Any) -> bool:
+ # np.isfinite raises/returns non-scalar on inputs it cannot reduce to a
+ # single finite float64 (Python ints >= 2**64 -> TypeError/OverflowError;
+ # array-likes -> a ValueError on bool()); treat all of those as non-finite.
+ try:
+ return bool(np.isfinite(value))
+ except (TypeError, OverflowError, ValueError):
+ return False
+
+
def is_finite_number(value: Any) -> bool:
- # Reject bool (int(True) == 1) and non-numeric (str/object) before
- # np.isfinite, which raises on non-numeric input.
+ # Reject bool (int(True) == 1) and non-numeric (str/object) before the
+ # finiteness check.
return (
not isinstance(value, bool)
and isinstance(value, (int, float, np.integer, np.floating))
- and bool(np.isfinite(value))
+ and _is_finite_value(value)
)
def __call__(self, value: Any) -> bool:
if self.require_int and not isinstance(value, int):
return False
- if not isinstance(value, (int, float)) or not np.isfinite(value):
+ if not isinstance(value, (int, float)) or not _is_finite_value(value):
return False
if self.min_value is not None:
if self.min_exclusive and value <= self.min_value:
def __call__(self, value: Any) -> bool:
if not isinstance(value, (list, tuple)) or len(value) != 2:
return False
- if not all(isinstance(x, (int, float)) and np.isfinite(x) for x in value):
+ if not all(isinstance(x, (int, float)) and _is_finite_value(x) for x in value):
return False
if value[0] >= value[1]:
return False
return get_label_kind_config("label_prediction", config, logger)
+DEFAULTS_EXIT_PRICING: Final[dict[str, Any]] = {
+ "trade_price_target_method": TRADE_PRICE_TARGETS[0], # "moving_average"
+}
+
+_EXIT_PRICING_SPECS: Final[dict[str, _ParamSpec]] = {
+ "trade_price_target_method": _ParamSpec(
+ _EnumValidator(TRADE_PRICE_TARGETS), output_type=str
+ ),
+}
+
+
+def get_exit_pricing_config(config: dict[str, Any], logger: Logger) -> dict[str, str]:
+ return _validate_params(
+ config, logger, "exit_pricing", _EXIT_PRICING_SPECS, DEFAULTS_EXIT_PRICING
+ )
+
+
+DEFAULTS_REVERSAL_CONFIRMATION: Final[dict[str, Any]] = {
+ "lookback_period_candles": 0,
+ "decay_fraction": 0.5,
+ "min_natr_multiplier_fraction": 0.0095,
+ "max_natr_multiplier_fraction": 0.0125,
+}
+
+# Scalars only: the coupled (min, max) natr pair needs validate_range's
+# cross-field ordering and per-component fallback, which _validate_params
+# cannot express.
+_REVERSAL_CONFIRMATION_SCALAR_SPECS: Final[dict[str, _ParamSpec]] = {
+ "lookback_period_candles": _ParamSpec(
+ _NumericValidator(min_value=0, require_int=True), output_type=int
+ ),
+ "decay_fraction": _ParamSpec(
+ _NumericValidator(min_value=0, max_value=1, min_exclusive=True),
+ output_type=float,
+ ),
+}
+
+
+def get_reversal_confirmation_config(
+ config: dict[str, Any], logger: Logger
+) -> dict[str, int | float]:
+ validated = _validate_params(
+ config,
+ logger,
+ "reversal_confirmation",
+ _REVERSAL_CONFIRMATION_SCALAR_SPECS,
+ {
+ key: DEFAULTS_REVERSAL_CONFIRMATION[key]
+ for key in _REVERSAL_CONFIRMATION_SCALAR_SPECS
+ },
+ )
+
+ min_natr_multiplier_fraction, max_natr_multiplier_fraction = validate_range(
+ config.get(
+ "min_natr_multiplier_fraction",
+ DEFAULTS_REVERSAL_CONFIRMATION["min_natr_multiplier_fraction"],
+ ),
+ config.get(
+ "max_natr_multiplier_fraction",
+ DEFAULTS_REVERSAL_CONFIRMATION["max_natr_multiplier_fraction"],
+ ),
+ logger,
+ name="natr_multiplier_fraction",
+ default_min=DEFAULTS_REVERSAL_CONFIRMATION["min_natr_multiplier_fraction"],
+ default_max=DEFAULTS_REVERSAL_CONFIRMATION["max_natr_multiplier_fraction"],
+ allow_equal=False,
+ non_negative=True,
+ finite_only=True,
+ )
+
+ return {
+ "lookback_period_candles": int(validated["lookback_period_candles"]),
+ "decay_fraction": float(validated["decay_fraction"]),
+ "min_natr_multiplier_fraction": float(min_natr_multiplier_fraction),
+ "max_natr_multiplier_fraction": float(max_natr_multiplier_fraction),
+ }
+
+
_CAUSAL_MODE_FALSE_WARNED: bool = False
if (
not isinstance(value, (int, float))
or isinstance(value, bool)
- or (finite_only and not np.isfinite(value))
+ or (finite_only and not _is_finite_value(value))
or (non_negative and value < 0)
):
logger.warning(
repositories / freqai-strategies.git / commitdiff
