From: Jérôme Benoit Date: Sat, 25 Jul 2026 14:10:01 +0000 (+0200) Subject: refactor(quickadapter): centralize exit_pricing and reversal_confirmation validation... X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=3810d3f252c0a78affbc24102324c152104d39c8;p=freqai-strategies.git refactor(quickadapter): centralize exit_pricing and reversal_confirmation validation (#125) 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. --- diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index 842bfaf..0fec811 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -7,7 +7,7 @@ import time 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, @@ -1266,7 +1266,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ), } - @property + @cached_property def _optuna_config(self) -> dict[str, Any]: optuna_default_config = { "enabled": False, @@ -1324,7 +1324,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): 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)) @@ -1356,32 +1356,32 @@ class QuickAdapterRegressorV3(BaseRegressionModel): 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) diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index d402dd1..d37bb9f 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -58,10 +58,12 @@ from Utils import ( 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, @@ -76,7 +78,6 @@ from Utils import ( safe_divide, smooth, top_log_return, - validate_range, vwapb, weight_fill_radius, zlema, @@ -149,13 +150,6 @@ class QuickAdapterV3(IStrategy): "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)} @@ -346,100 +340,37 @@ class QuickAdapterV3(IStrategy): 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) diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 247b803..ebb81e2 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -307,13 +307,23 @@ def safe_log_ratio( 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) ) @@ -339,7 +349,7 @@ class _NumericValidator: 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: @@ -376,7 +386,7 @@ class _RangeValidator: 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 @@ -1184,6 +1194,84 @@ def get_label_prediction_config( 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 @@ -5119,7 +5207,7 @@ def validate_range( 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(