From 224e51106d17695428861a6fe7e367d47389570d Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sat, 25 Jul 2026 20:12:24 +0200 Subject: [PATCH] refactor(quickadapter): harmonize patterns and consolidate common logic (#126) MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * refactor(quickadapter): harmonize patterns and consolidate common logic Behavior-preserving refactor of the QuickAdapter codebase (Utils.py, QuickAdapterV3.py, QuickAdapterRegressorV3.py). Every change verified in the project Docker image (freqtrade + optuna/ngboost/catboost) via import smoke and per-change golden/behavior tests; py_compile clean, ruff E,F,W baseline unchanged or reduced, ruff format clean. Consolidation: - C1: single _get_validation_size() replaces 3 identical test_size None-coerce blocks (None-vs-0 distinction preserved). - C2: _resolve_optuna_store() registry replaces 6 near-duplicate get/set_optuna_* dispatchers; hp-only/label-only asymmetry and per-accessor defaults preserved. - C4: as_dict() helper replaces 7 dict-guard blocks across both classes. - C5: _distance_to_reference() extracts the 4-family distance dispatch shared by _compromise_programming_scores and _topsis_scores (apply_abs flag preserves the TOPSIS-only np.abs asymmetry); golden differential over all 17 metrics is bit-identical. - C6: _trade_natr_window() extracts the shared NATR preamble of the weighted/quantile trade-NATR methods (single-candle sentinel preserved). - C7: _invalidate_pair_cache() replaces 2 identical per-pair cache-invalidation blocks (rebind-on-change / same-object-on-noop preserved). - C8: _validate_step_args() shares the guard preamble of round/ceil/floor_to_step. - C9: _pop_early_stopping_rounds() and _apply_verbosity_alias() replace repeated fit_regressor boilerplate (verbose precedence preserved). - C10: drop dead default 2nd-args on _optuna_config[...] lookups (merge is exhaustive; constant_liar path kept). Harmonization: - H2/C3: fold 3 near-identical scalar validators into one _validate_scalar(predicate, constraint); exact messages preserved (subsumes H1 predicate sharing). - H4: single canonical source for label defaults in Utils; RV3 ClassVars reference it (fixes value duplication). - H6: enum_error_message() builder; identical-text regressor/aggregation raises routed through it. - H7: make the intended mode="raise" explicit for label_method and scaler validation (no behavior change). Notes: - H5 (DEFAULT_X vs X_DEFAULT naming): the duplicated default values were removed by H4; the remaining prefix/suffix difference is a consistent module-scalar vs class-ClassVar convention, left as-is. - H3 (magic-index TUPLE[n] -> named members) remains as a follow-up; behavior-neutral cosmetic cleanup. * refactor(quickadapter): replace magic-index dispatch with named constants (H3) Every fragile TUPLE[n] dispatch/comparison site (~64 across the three files) now reads a named Final constant defined next to its tuple, extending the existing _OPTUNA_NAMESPACES / _REGRESSOR_SPECS idiom. Reordering a literals tuple can no longer silently break dispatch, and the trailing `# "name"` comments are removed as redundant. Covered tuples: _DISTANCE_METRICS, _DISTANCE_METHODS, _CLUSTER_METHODS, _SELECTION_METHODS, _DENSITY_METHODS, _DENSITY_AGGREGATIONS, _SCALER_TYPES, _DATA_SPLIT_METHODS, _OPTUNA_STORAGE_BACKENDS (RegressorV3); _TRADE_DIRECTIONS, _ORDER_TYPES, _INTERPOLATION_DIRECTIONS, _TRADING_MODES, and the imported SMOOTHING_METHODS / FILL_METHODS / WEIGHT_STRATEGIES (QuickAdapterV3). The canonical named-derivation definitions (_UNSUPPORTED_WEIGHTS_METRICS and the LABEL_*_DEFAULT / *_DEFAULT ClassVars) keep their index form: they are the single source that maps index -> name. Behaviour preserved by construction (each named constant is bound to the same tuple element); verified in the project Docker image by asserting every constant equals its tuple value and that CP/TOPSIS scoring is unchanged. * refactor(quickadapter): address initial-review findings - M1: hoist the scipy cdist kwargs in _topsis_scores and pass them into both _distance_to_reference calls, so _prepare_distance_kwargs (and its warn-mode validators) run once instead of twice for scipy metrics — restores the original single-warning behavior on an invalid label_distance_p / unsupported weights. Distance scores stay bit-identical (kwargs are deterministic). - N1: make _validate_scalar's predicate/constraint optional; _validate_power_mean_p now calls it with no predicate (finite-only), dropping the dead always-true lambda + unreachable constraint string. - N2: type _invalidate_pair_cache with a bound TypeVar so each caller's cache type flows through instead of a bare dict. - N3: drop 3 redundant `# "name"` comments left next to named constants. Verified in the project Docker image: CP/TOPSIS golden differential vs the prior commit is bit-identical across all 17 metrics; TOPSIS invalid-p warnings 2 -> 1; validator messages unchanged; py_compile clean, ruff E,F,W <= main baseline, ruff format clean. * refactor(quickadapter): address re-review nits (TypeVar soundness, comment cleanup) - NEW-1: rebuild the invalidated cache via type(cache)(...) in _invalidate_pair_cache so the _PairCacheT TypeVar return is sound. pyright: prior plain-dict-comprehension reassignment reported reportAssignmentType (1 error); type(cache)(...) reports 0. Runtime identical (type(cache) is dict for the two concrete caches). - NEW-2: drop the 32 remaining tautological `# "name"` comments sitting next to a named constant (e.g. `== _TRADE_LONG # "long"`), consistent with the H3 cleanup. Index-resolving comments on the canonical definitions (e.g. `_DISTANCE_METRICS[6] # "mahalanobis"`) are kept — they document the literal an index maps to. Verified in the project Docker image: CP/TOPSIS bit-identical vs prior commit across all metrics; _invalidate_pair_cache runtime unchanged; py_compile clean, ruff E,F,W <= main baseline, ruff format clean. * refactor(quickadapter): strip residual tautological named-constant comments Complete NEW-2 cleanup: remove 14 tautological `# "name"` comments on multi-line comparison/closing-paren lines that the single-line strip missed (e.g. `): # "power_mean"`, `) # "short"`), fixing the long-removed/short-kept asymmetry. Removing the pinning comments lets ruff format collapse the parenthesized `if (...)` forms to single lines. Keeps index-resolving comments (`_ARRAY[n] # "x"`) and the LABEL_METHOD_DEFAULT value-documenting comment. Comment-only + format change: AST verified identical to prior commit; ruff E,F,W within baseline; container import smoke passes. --- .../freqaimodels/QuickAdapterRegressorV3.py | 665 ++++++++---------- .../user_data/strategies/QuickAdapterV3.py | 209 +++--- quickadapter/user_data/strategies/Utils.py | 126 ++-- 3 files changed, 451 insertions(+), 549 deletions(-) diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index 0fec811..afa2b87 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -67,7 +67,13 @@ from LabelTransformer import ( ) from Utils import ( + as_dict, + enum_error_message, DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES, + DEFAULT_MAX_LABEL_NATR_MULTIPLIER, + DEFAULT_MAX_LABEL_PERIOD_CANDLES, + DEFAULT_MIN_LABEL_NATR_MULTIPLIER, + DEFAULT_MIN_LABEL_PERIOD_CANDLES, DEFAULT_REGRESSOR, DEFAULTS_LABEL_PREDICTION, LABEL_COLUMNS, @@ -326,6 +332,22 @@ class QuickAdapterRegressorV3(BaseRegressionModel): _UNSUPPORTED_WEIGHTS_METRICS ) + _METHOD_COMPROMISE_PROGRAMMING: Final[str] = _DISTANCE_METHODS[0] + _METHOD_TOPSIS: Final[str] = _DISTANCE_METHODS[1] + _METRIC_EUCLIDEAN: Final[str] = _DISTANCE_METRICS[0] + _METRIC_MINKOWSKI: Final[str] = _DISTANCE_METRICS[1] + _METRIC_HELLINGER: Final[str] = _DISTANCE_METRICS[8] + _METRIC_SHELLINGER: Final[str] = _DISTANCE_METRICS[9] + _METRIC_POWER_MEAN: Final[str] = _DISTANCE_METRICS[15] + _METRIC_WEIGHTED_SUM: Final[str] = _DISTANCE_METRICS[16] + _CLUSTER_KMEANS: Final[str] = _CLUSTER_METHODS[0] + _CLUSTER_KMEANS2: Final[str] = _CLUSTER_METHODS[1] + _SELECTION_KMEANS: Final[str] = _SELECTION_METHODS[2] + _SELECTION_KMEANS2: Final[str] = _SELECTION_METHODS[3] + _SELECTION_KMEDOIDS: Final[str] = _SELECTION_METHODS[4] + _SELECTION_KNN: Final[str] = _SELECTION_METHODS[5] + _SELECTION_MEDOID: Final[str] = _SELECTION_METHODS[6] + _PROBABILITY_DISTANCE_METRICS: Final[tuple[str, ...]] = ( "jensenshannon", "hellinger", @@ -374,10 +396,10 @@ class QuickAdapterRegressorV3(BaseRegressionModel): FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT: Final[int] = ( DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES ) - MIN_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = 12 - MAX_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = 24 - MIN_LABEL_NATR_MULTIPLIER_DEFAULT: Final[float] = 9.0 - MAX_LABEL_NATR_MULTIPLIER_DEFAULT: Final[float] = 12.0 + MIN_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = DEFAULT_MIN_LABEL_PERIOD_CANDLES + MAX_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = DEFAULT_MAX_LABEL_PERIOD_CANDLES + MIN_LABEL_NATR_MULTIPLIER_DEFAULT: Final[float] = DEFAULT_MIN_LABEL_NATR_MULTIPLIER + MAX_LABEL_NATR_MULTIPLIER_DEFAULT: Final[float] = DEFAULT_MAX_LABEL_NATR_MULTIPLIER LABEL_METHOD_DEFAULT: Final[str] = _SELECTION_METHODS[0] # "compromise_programming" @@ -411,6 +433,19 @@ class QuickAdapterRegressorV3(BaseRegressionModel): "timeseries_split", ) DATA_SPLIT_METHOD_DEFAULT: Final[str] = _DATA_SPLIT_METHODS[0] + _DATA_SPLIT_TIMESERIES: Final[str] = _DATA_SPLIT_METHODS[1] + _CLUSTER_KMEDOIDS: Final[str] = _CLUSTER_METHODS[2] + _DENSITY_KNN: Final[str] = _DENSITY_METHODS[0] + _DENSITY_MEDOID: Final[str] = _DENSITY_METHODS[1] + _DENSITY_AGG_POWER_MEAN: Final[str] = _DENSITY_AGGREGATIONS[0] + _DENSITY_AGG_QUANTILE: Final[str] = _DENSITY_AGGREGATIONS[1] + _DENSITY_AGG_MIN: Final[str] = _DENSITY_AGGREGATIONS[2] + _DENSITY_AGG_MAX: Final[str] = _DENSITY_AGGREGATIONS[3] + _SCALER_MAXABS: Final[str] = _SCALER_TYPES[1] + _SCALER_STANDARD: Final[str] = _SCALER_TYPES[2] + _SCALER_ROBUST: Final[str] = _SCALER_TYPES[3] + _STORAGE_FILE: Final[str] = _OPTUNA_STORAGE_BACKENDS[0] + _STORAGE_SQLITE: Final[str] = _OPTUNA_STORAGE_BACKENDS[1] TIMESERIES_N_SPLITS_DEFAULT: Final[int] = 5 TIMESERIES_GAP_DEFAULT: Final[int] = 0 TIMESERIES_MAX_TRAIN_SIZE_DEFAULT: Final[int | None] = None @@ -753,65 +788,71 @@ class QuickAdapterRegressorV3(BaseRegressionModel): @staticmethod def _get_label_p_order_default(distance_metric: str) -> Optional[float]: - if ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[1] - ): # "minkowski" + if distance_metric == QuickAdapterRegressorV3._METRIC_MINKOWSKI: return 2.0 - elif ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[15] - ): # "power_mean" + elif distance_metric == QuickAdapterRegressorV3._METRIC_POWER_MEAN: return 1.0 return None @staticmethod def _get_label_density_metric_default(method: DensityMethod) -> Optional[str]: - if method == QuickAdapterRegressorV3._DENSITY_METHODS[1]: # "medoid" - return QuickAdapterRegressorV3._DISTANCE_METRICS[0] # "euclidean" - elif method == QuickAdapterRegressorV3._DENSITY_METHODS[0]: # "knn" - return QuickAdapterRegressorV3._DISTANCE_METRICS[1] # "minkowski" + if method == QuickAdapterRegressorV3._DENSITY_MEDOID: + return QuickAdapterRegressorV3._METRIC_EUCLIDEAN + elif method == QuickAdapterRegressorV3._DENSITY_KNN: + return QuickAdapterRegressorV3._METRIC_MINKOWSKI return None @staticmethod def _get_label_density_aggregation_param_default( aggregation: DensityAggregation, ) -> Optional[float]: - if ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[0] - ): # "power_mean" + if aggregation == QuickAdapterRegressorV3._DENSITY_AGG_POWER_MEAN: return 1.0 - elif ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[1] - ): # "quantile" + elif aggregation == QuickAdapterRegressorV3._DENSITY_AGG_QUANTILE: return 0.5 return None @staticmethod - def _validate_minkowski_p( - p: Optional[float], + def _validate_scalar( + value: Optional[float], *, ctx: str, - mode: ValidationMode = "raise", + mode: ValidationMode, + predicate: Optional[Callable[[float], bool]] = None, + constraint: str = "", ) -> Optional[float]: - if p is None: + if value is None: return None if mode == "none": - return float(p) if (np.isfinite(p) and p > 0) else None + return ( + float(value) + if (np.isfinite(value) and (predicate is None or predicate(value))) + else None + ) - if not np.isfinite(p): - msg = f"Invalid {ctx} value {p!r}: must be finite" + if not np.isfinite(value): + msg = f"Invalid {ctx} value {value!r}: must be finite" if mode == "raise": raise ValueError(msg) logger.warning(f"{msg}, using default") return None - if p <= 0: - msg = f"Invalid {ctx} value {p!r}: must be > 0" + if predicate is not None and not predicate(value): + msg = f"Invalid {ctx} value {value!r}: {constraint}" if mode == "raise": raise ValueError(msg) logger.warning(f"{msg}, using default") return None - return float(p) + return float(value) + + @staticmethod + def _validate_minkowski_p( + p: Optional[float], *, ctx: str, mode: ValidationMode = "raise" + ) -> Optional[float]: + return QuickAdapterRegressorV3._validate_scalar( + p, ctx=ctx, mode=mode, predicate=lambda v: v > 0, constraint="must be > 0" + ) @staticmethod def _prepare_distance_kwargs( @@ -831,9 +872,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if validated_metric is not None: kwargs["w"] = weights - if ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[1] - ): # "minkowski" + if distance_metric == QuickAdapterRegressorV3._METRIC_MINKOWSKI: validated_p = QuickAdapterRegressorV3._validate_minkowski_p( p, ctx=p_ctx, mode=mode ) @@ -846,44 +885,19 @@ class QuickAdapterRegressorV3(BaseRegressionModel): def _validate_quantile_q( q: Optional[float], *, ctx: str, mode: ValidationMode = "raise" ) -> Optional[float]: - if q is None: - return None - if mode == "none": - return float(q) if (np.isfinite(q) and 0.0 <= q <= 1.0) else None - - if not np.isfinite(q): - msg = f"Invalid {ctx} value {q!r}: must be finite" - if mode == "raise": - raise ValueError(msg) - logger.warning(f"{msg}, using default") - return None - - if q < 0.0 or q > 1.0: - msg = f"Invalid {ctx} value {q!r}: must be in [0, 1]" - if mode == "raise": - raise ValueError(msg) - logger.warning(f"{msg}, using default") - return None - - return float(q) + return QuickAdapterRegressorV3._validate_scalar( + q, + ctx=ctx, + mode=mode, + predicate=lambda v: 0.0 <= v <= 1.0, + constraint="must be in [0, 1]", + ) @staticmethod def _validate_power_mean_p( p: Optional[float], *, ctx: str, mode: ValidationMode = "raise" ) -> Optional[float]: - if p is None: - return None - if mode == "none": - return float(p) if np.isfinite(p) else None - - if not np.isfinite(p): - msg = f"Invalid {ctx} value {p!r}: must be finite" - if mode == "raise": - raise ValueError(msg) - logger.warning(f"{msg}, using default") - return None - - return float(p) + return QuickAdapterRegressorV3._validate_scalar(p, ctx=ctx, mode=mode) @staticmethod def _validate_metric_weights_support( @@ -1049,7 +1063,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) -> dict[str, Any]: knn_kwargs: dict[str, Any] = {} - if distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[1]: + if distance_metric == QuickAdapterRegressorV3._METRIC_MINKOWSKI: validated_p = QuickAdapterRegressorV3._validate_minkowski_p( p, ctx=p_ctx, mode=mode ) @@ -1073,9 +1087,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if label_p_order is not None else QuickAdapterRegressorV3._get_label_p_order_default(distance_metric) ) - if ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[1] - ): # "minkowski" + if distance_metric == QuickAdapterRegressorV3._METRIC_MINKOWSKI: p = QuickAdapterRegressorV3._validate_minkowski_p(p, ctx=ctx, mode=mode) return p @@ -1085,6 +1097,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): QuickAdapterRegressorV3._SELECTION_METHODS_SET, QuickAdapterRegressorV3._SELECTION_METHODS, ctx="label_method", + mode="raise", ) category = QuickAdapterRegressorV3._get_selection_category(label_method) @@ -1163,7 +1176,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) config["distance_metric"] = distance_metric - if density_method == QuickAdapterRegressorV3._DENSITY_METHODS[0]: # "knn" + if density_method == QuickAdapterRegressorV3._DENSITY_KNN: aggregation = cast( DensityAggregation, self.ft_params.get( @@ -1197,16 +1210,12 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) if aggregation_param is not None: - if ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[1] - ): # "quantile" + if aggregation == QuickAdapterRegressorV3._DENSITY_AGG_QUANTILE: QuickAdapterRegressorV3._validate_quantile_q( aggregation_param, ctx="label_density_aggregation_param", ) - elif ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[0] - ): # "power_mean" + elif aggregation == QuickAdapterRegressorV3._DENSITY_AGG_POWER_MEAN: QuickAdapterRegressorV3._validate_power_mean_p( aggregation_param, ctx="label_density_aggregation_param", @@ -1277,7 +1286,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): max(int(self.max_system_threads / 4), 1), ), "sampler": QuickAdapterRegressorV3._OPTUNA_HPO_SAMPLERS.tpe, - "storage": QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS[0], # "file" + "storage": QuickAdapterRegressorV3._STORAGE_FILE, "continuous": True, "warm_start": True, "n_startup_trials": QuickAdapterRegressorV3.OPTUNA_N_STARTUP_TRIALS_DEFAULT, @@ -1358,24 +1367,21 @@ class QuickAdapterRegressorV3(BaseRegressionModel): @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) + return get_label_weighting_config( + as_dict(self.freqai_info.get("label_weighting")), logger + ) @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) + return get_label_pipeline_config( + as_dict(self.freqai_info.get("label_pipeline")), logger + ) @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) + return get_label_prediction_config( + as_dict(self.freqai_info.get("label_prediction")), logger + ) @cached_property def _label_defaults(self) -> tuple[int, float]: @@ -1435,9 +1441,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): self._optuna_hp_params: dict[str, dict[str, Any]] = {} self._optuna_label_params: dict[str, dict[str, Any]] = {} self._optuna_label_candle_pool_full_cache: dict[int, list[int]] = {} - self._optuna_label_shuffle_rng = random.Random( - self._optuna_config.get("seed", QuickAdapterRegressorV3.OPTUNA_SEED_DEFAULT) - ) + self._optuna_label_shuffle_rng = random.Random(self._optuna_config["seed"]) self.init_optuna_label_candle_pool() self._optuna_label_candle: dict[str, int] = {} self._optuna_label_candles: dict[str, int] = {} @@ -1541,8 +1545,8 @@ class QuickAdapterRegressorV3(BaseRegressionModel): else: distance_metric = label_config["distance_metric"] if distance_metric in { - QuickAdapterRegressorV3._DISTANCE_METRICS[1], # "minkowski" - QuickAdapterRegressorV3._DISTANCE_METRICS[15], # "power_mean" + QuickAdapterRegressorV3._METRIC_MINKOWSKI, + QuickAdapterRegressorV3._METRIC_POWER_MEAN, }: label_p_order_default = ( QuickAdapterRegressorV3._get_label_p_order_default( @@ -1661,78 +1665,68 @@ class QuickAdapterRegressorV3(BaseRegressionModel): logger.info("=" * 60) + def _resolve_optuna_store( + self, namespace: OptunaNamespace, stores: dict[OptunaNamespace, Any] + ) -> Any: + if namespace not in stores: + raise ValueError(enum_error_message("namespace", namespace, tuple(stores))) + return stores[namespace] + def get_optuna_params( self, pair: str, namespace: OptunaNamespace ) -> dict[str, Any]: - if namespace == _OPTUNA_NAMESPACES.hp: - params = self._optuna_hp_params.get(pair, {}) - elif namespace == _OPTUNA_NAMESPACES.label: - params = self._optuna_label_params.get(pair, {}) - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {', '.join(_OPTUNA_NAMESPACES)}" - ) - return params + store = self._resolve_optuna_store( + namespace, + { + _OPTUNA_NAMESPACES.hp: self._optuna_hp_params, + _OPTUNA_NAMESPACES.label: self._optuna_label_params, + }, + ) + return store.get(pair, {}) def set_optuna_params( self, pair: str, namespace: OptunaNamespace, params: dict[str, Any] ) -> None: - if namespace == _OPTUNA_NAMESPACES.hp: - self._optuna_hp_params[pair] = params - elif namespace == _OPTUNA_NAMESPACES.label: - self._optuna_label_params[pair] = params - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {', '.join(_OPTUNA_NAMESPACES)}" - ) + store = self._resolve_optuna_store( + namespace, + { + _OPTUNA_NAMESPACES.hp: self._optuna_hp_params, + _OPTUNA_NAMESPACES.label: self._optuna_label_params, + }, + ) + store[pair] = params def get_optuna_value(self, pair: str, namespace: OptunaNamespace) -> float: - if namespace == _OPTUNA_NAMESPACES.hp: - value = self._optuna_hp_value.get(pair, np.nan) - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {_OPTUNA_NAMESPACES.hp!r}" - ) - return value + store = self._resolve_optuna_store( + namespace, {_OPTUNA_NAMESPACES.hp: self._optuna_hp_value} + ) + return store.get(pair, np.nan) def set_optuna_value( self, pair: str, namespace: OptunaNamespace, value: float ) -> None: - if namespace == _OPTUNA_NAMESPACES.hp: - self._optuna_hp_value[pair] = value - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {_OPTUNA_NAMESPACES.hp!r}" - ) + store = self._resolve_optuna_store( + namespace, {_OPTUNA_NAMESPACES.hp: self._optuna_hp_value} + ) + store[pair] = value def get_optuna_values( self, pair: str, namespace: OptunaNamespace ) -> list[float | int]: - if namespace == _OPTUNA_NAMESPACES.label: - values = self._optuna_label_values.get( - pair, [np.nan] * QuickAdapterRegressorV3._OPTUNA_LABEL_N_OBJECTIVES - ) - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {_OPTUNA_NAMESPACES.label}" - ) - return values + store = self._resolve_optuna_store( + namespace, {_OPTUNA_NAMESPACES.label: self._optuna_label_values} + ) + return store.get( + pair, [np.nan] * QuickAdapterRegressorV3._OPTUNA_LABEL_N_OBJECTIVES + ) def set_optuna_values( self, pair: str, namespace: OptunaNamespace, values: list[float | int] ) -> None: - if namespace == _OPTUNA_NAMESPACES.label: - self._optuna_label_values[pair] = values - else: - raise ValueError( - f"Invalid namespace value {namespace!r}: " - f"supported values are {_OPTUNA_NAMESPACES.label}" - ) + store = self._resolve_optuna_store( + namespace, {_OPTUNA_NAMESPACES.label: self._optuna_label_values} + ) + store[pair] = values def init_optuna_label_candle_pool(self) -> None: optuna_label_candle_pool_full = self._optuna_label_candle_pool_full @@ -1798,6 +1792,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): QuickAdapterRegressorV3._SCALER_TYPES_SET, QuickAdapterRegressorV3._SCALER_TYPES, ctx="scaler", + mode="raise", ) feature_range = self.ft_params.get( @@ -1822,11 +1817,11 @@ class QuickAdapterRegressorV3(BaseRegressionModel): pipeline = super().define_data_pipeline(threads) - if scaler == QuickAdapterRegressorV3._SCALER_TYPES[1]: # "maxabs" + if scaler == QuickAdapterRegressorV3._SCALER_MAXABS: scaler_obj = SKLearnWrapper(MaxAbsScaler()) - elif scaler == QuickAdapterRegressorV3._SCALER_TYPES[2]: # "standard" + elif scaler == QuickAdapterRegressorV3._SCALER_STANDARD: scaler_obj = SKLearnWrapper(StandardScaler()) - elif scaler == QuickAdapterRegressorV3._SCALER_TYPES[3]: # "robust" + elif scaler == QuickAdapterRegressorV3._SCALER_ROBUST: scaler_obj = SKLearnWrapper(RobustScaler()) else: # "minmax" scaler_obj = SKLearnWrapper(MinMaxScaler(feature_range=feature_range)) @@ -2236,6 +2231,14 @@ class QuickAdapterRegressorV3(BaseRegressionModel): < cutoff_position ) + def _get_validation_size(self) -> float | int: + validation_size = self.data_split_parameters.get( + "test_size", QuickAdapterRegressorV3._TEST_SIZE + ) + if validation_size is None: + return QuickAdapterRegressorV3._TEST_SIZE + return validation_size + def _add_validation_split( self, data_dictionary: dict[str, Any], @@ -2245,11 +2248,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): pair: str, ) -> dict[str, Any]: """Reserve the chronological tail of the training set for selection.""" - validation_size = self.data_split_parameters.get( - "test_size", QuickAdapterRegressorV3._TEST_SIZE - ) - if validation_size is None: - validation_size = QuickAdapterRegressorV3._TEST_SIZE + validation_size = self._get_validation_size() if validation_size == 0: data_dictionary["validation_features"] = data_dictionary[ "train_features" @@ -2392,11 +2391,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): pair: str, ) -> dict[str, Any]: """Keep all causally available rows for the final post-holdout refit.""" - validation_size = self.data_split_parameters.get( - "test_size", QuickAdapterRegressorV3._TEST_SIZE - ) - if validation_size is None: - validation_size = QuickAdapterRegressorV3._TEST_SIZE + validation_size = self._get_validation_size() if validation_size == 0: return data_dictionary @@ -2419,7 +2414,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): self.data_split_parameters.get( "method", QuickAdapterRegressorV3.DATA_SPLIT_METHOD_DEFAULT ) - == QuickAdapterRegressorV3._DATA_SPLIT_METHODS[1] + == QuickAdapterRegressorV3._DATA_SPLIT_TIMESERIES ): max_train_size = QuickAdapterRegressorV3._coerce_optional_int( self.data_split_parameters.get( @@ -2531,7 +2526,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): "method", QuickAdapterRegressorV3.DATA_SPLIT_METHOD_DEFAULT ) if ( - method == QuickAdapterRegressorV3._DATA_SPLIT_METHODS[1] + method == QuickAdapterRegressorV3._DATA_SPLIT_TIMESERIES ): # timeseries_split n_splits = self.data_split_parameters.get( "n_splits", QuickAdapterRegressorV3.TIMESERIES_N_SPLITS_DEFAULT @@ -2781,11 +2776,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): X_validation = data_dictionary.get("validation_features") y_validation = data_dictionary.get("validation_labels") validation_weights = data_dictionary.get("validation_weights") - validation_size = self.data_split_parameters.get( - "test_size", QuickAdapterRegressorV3._TEST_SIZE - ) - if validation_size is None: - validation_size = QuickAdapterRegressorV3._TEST_SIZE + validation_size = self._get_validation_size() model_training_parameters = copy.deepcopy(self.model_training_parameters) init_model = self.get_init_model(dk.pair) @@ -2807,14 +2798,8 @@ class QuickAdapterRegressorV3(BaseRegressionModel): validation_size, self.get_optuna_params(dk.pair, _OPTUNA_NAMESPACES.hp), model_training_parameters, - self._optuna_config.get( - "space_reduction", - QuickAdapterRegressorV3.OPTUNA_SPACE_REDUCTION_DEFAULT, - ), - self._optuna_config.get( - "space_fraction", - QuickAdapterRegressorV3.OPTUNA_SPACE_FRACTION_DEFAULT, - ), + self._optuna_config["space_reduction"], + self._optuna_config["space_fraction"], dk.data_path, init_model, ), @@ -3205,10 +3190,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): trial, label_dataframe, fit_live_predictions_candles, - self._optuna_config.get( - "label_candles_step", - QuickAdapterRegressorV3.OPTUNA_LABEL_CANDLES_STEP_DEFAULT, - ), + self._optuna_config["label_candles_step"], min_label_period_candles=self._min_label_period_candles, max_label_period_candles=self._max_label_period_candles, min_label_natr_multiplier=self._min_label_natr_multiplier, @@ -3595,83 +3577,106 @@ class QuickAdapterRegressorV3(BaseRegressionModel): return (reference_point - matrix) @ weights @staticmethod - def _compromise_programming_scores( + def _distance_to_reference( normalized_matrix: NDArray[np.floating], + reference_point: NDArray[np.floating], distance_metric: str, *, - weights: Optional[NDArray[np.floating]] = None, - p: Optional[float] = None, + weights: NDArray[np.floating], + p: Optional[float], + method: str, + apply_abs: bool, + cdist_kwargs: Optional[dict[str, Any]] = None, ) -> NDArray[np.floating]: - n_samples, n_objectives = normalized_matrix.shape - - if n_samples == 0: - return np.array([]) - if n_samples == 1: - return np.array([0.0]) - - if weights is None: - weights = np.ones(n_objectives) - - ideal_point = np.ones(n_objectives) - if distance_metric in QuickAdapterRegressorV3._SCIPY_METRICS_SET: - return sp.spatial.distance.cdist( - normalized_matrix, - ideal_point.reshape(1, -1), - metric=distance_metric, - **QuickAdapterRegressorV3._prepare_distance_kwargs( + if cdist_kwargs is None: + cdist_kwargs = QuickAdapterRegressorV3._prepare_distance_kwargs( distance_metric, weights=weights, p=p, mode="warn", metric_ctx="label_distance_metric", p_ctx="label_distance_p", - ), + ) + return sp.spatial.distance.cdist( + normalized_matrix, + reference_point.reshape(1, -1), + metric=distance_metric, + **cdist_kwargs, ).flatten() if distance_metric in { - QuickAdapterRegressorV3._DISTANCE_METRICS[8], # "hellinger" - QuickAdapterRegressorV3._DISTANCE_METRICS[9], # "shellinger" + QuickAdapterRegressorV3._METRIC_HELLINGER, + QuickAdapterRegressorV3._METRIC_SHELLINGER, }: return QuickAdapterRegressorV3._hellinger_distance( normalized_matrix, - ideal_point, + reference_point, weights=weights, standardized=( - distance_metric - == QuickAdapterRegressorV3._DISTANCE_METRICS[9] # "shellinger" + distance_metric == QuickAdapterRegressorV3._METRIC_SHELLINGER ), ) if distance_metric in ( QuickAdapterRegressorV3._POWER_MEAN_METRICS_SET - | {QuickAdapterRegressorV3._DISTANCE_METRICS[15]} # "power_mean" + | {QuickAdapterRegressorV3._METRIC_POWER_MEAN} ): - return QuickAdapterRegressorV3._power_mean_distance( + distances = QuickAdapterRegressorV3._power_mean_distance( normalized_matrix, - ideal_point, + reference_point, distance_metric, weights=weights, p=p, mode="warn", p_ctx="label_distance_p", ) + return np.abs(distances) if apply_abs else distances - if ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[16] - ): # "weighted_sum" + if distance_metric == QuickAdapterRegressorV3._METRIC_WEIGHTED_SUM: assert weights is not None - return QuickAdapterRegressorV3._weighted_sum_distance( + distances = QuickAdapterRegressorV3._weighted_sum_distance( normalized_matrix, - ideal_point, + reference_point, weights=weights, ) + return np.abs(distances) if apply_abs else distances raise ValueError( - f"Invalid distance_metric value {distance_metric!r} for {QuickAdapterRegressorV3._DISTANCE_METHODS[0]}: " + f"Invalid distance_metric value {distance_metric!r} for {method}: " f"supported values are {', '.join(QuickAdapterRegressorV3._DISTANCE_METRICS)}" ) + @staticmethod + def _compromise_programming_scores( + normalized_matrix: NDArray[np.floating], + distance_metric: str, + *, + weights: Optional[NDArray[np.floating]] = None, + p: Optional[float] = None, + ) -> NDArray[np.floating]: + n_samples, n_objectives = normalized_matrix.shape + + if n_samples == 0: + return np.array([]) + if n_samples == 1: + return np.array([0.0]) + + if weights is None: + weights = np.ones(n_objectives) + + ideal_point = np.ones(n_objectives) + + return QuickAdapterRegressorV3._distance_to_reference( + normalized_matrix, + ideal_point, + distance_metric, + weights=weights, + p=p, + method=QuickAdapterRegressorV3._METHOD_COMPROMISE_PROGRAMMING, + apply_abs=False, + ) + @staticmethod def _pairwise_distance_sums( matrix: NDArray[np.floating], @@ -3743,99 +3748,39 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ideal_point = np.ones(n_objectives) anti_ideal_point = np.zeros(n_objectives) - if distance_metric in QuickAdapterRegressorV3._SCIPY_METRICS_SET: - cdist_kwargs = QuickAdapterRegressorV3._prepare_distance_kwargs( - distance_metric=distance_metric, + cdist_kwargs = ( + QuickAdapterRegressorV3._prepare_distance_kwargs( + distance_metric, weights=weights, p=p, mode="warn", metric_ctx="label_distance_metric", p_ctx="label_distance_p", ) + if distance_metric in QuickAdapterRegressorV3._SCIPY_METRICS_SET + else None + ) - dist_to_ideal = sp.spatial.distance.cdist( - normalized_matrix, - ideal_point.reshape(1, -1), - metric=distance_metric, - **cdist_kwargs, - ).flatten() - dist_to_anti_ideal = sp.spatial.distance.cdist( - normalized_matrix, - anti_ideal_point.reshape(1, -1), - metric=distance_metric, - **cdist_kwargs, - ).flatten() - elif distance_metric in { - QuickAdapterRegressorV3._DISTANCE_METRICS[8], # "hellinger" - QuickAdapterRegressorV3._DISTANCE_METRICS[9], # "shellinger" - }: - dist_to_ideal = QuickAdapterRegressorV3._hellinger_distance( - normalized_matrix, - ideal_point, - weights=weights, - standardized=( - distance_metric - == QuickAdapterRegressorV3._DISTANCE_METRICS[9] # "shellinger" - ), - ) - dist_to_anti_ideal = QuickAdapterRegressorV3._hellinger_distance( - normalized_matrix, - anti_ideal_point, - weights=weights, - standardized=( - distance_metric - == QuickAdapterRegressorV3._DISTANCE_METRICS[9] # "shellinger" - ), - ) - elif distance_metric in ( - QuickAdapterRegressorV3._POWER_MEAN_METRICS_SET - | {QuickAdapterRegressorV3._DISTANCE_METRICS[15]} # "power_mean" - ): - dist_to_ideal = np.abs( - QuickAdapterRegressorV3._power_mean_distance( - normalized_matrix, - ideal_point, - distance_metric, - weights=weights, - p=p, - mode="warn", - p_ctx="label_distance_p", - ) - ) - dist_to_anti_ideal = np.abs( - QuickAdapterRegressorV3._power_mean_distance( - normalized_matrix, - anti_ideal_point, - distance_metric, - weights=weights, - p=p, - mode="warn", - p_ctx="label_distance_p", - ) - ) - elif ( - distance_metric == QuickAdapterRegressorV3._DISTANCE_METRICS[16] - ): # "weighted_sum" - assert weights is not None - dist_to_ideal = np.abs( - QuickAdapterRegressorV3._weighted_sum_distance( - normalized_matrix, - ideal_point, - weights=weights, - ) - ) - dist_to_anti_ideal = np.abs( - QuickAdapterRegressorV3._weighted_sum_distance( - normalized_matrix, - anti_ideal_point, - weights=weights, - ) - ) - else: - raise ValueError( - f"Invalid distance_metric value {distance_metric!r} for {QuickAdapterRegressorV3._DISTANCE_METHODS[1]}: " - f"supported values are {', '.join(QuickAdapterRegressorV3._DISTANCE_METRICS)}" - ) + dist_to_ideal = QuickAdapterRegressorV3._distance_to_reference( + normalized_matrix, + ideal_point, + distance_metric, + weights=weights, + p=p, + method=QuickAdapterRegressorV3._METHOD_TOPSIS, + apply_abs=True, + cdist_kwargs=cdist_kwargs, + ) + dist_to_anti_ideal = QuickAdapterRegressorV3._distance_to_reference( + normalized_matrix, + anti_ideal_point, + distance_metric, + weights=weights, + p=p, + method=QuickAdapterRegressorV3._METHOD_TOPSIS, + apply_abs=True, + cdist_kwargs=cdist_kwargs, + ) denominator = dist_to_ideal + dist_to_anti_ideal zero_mask = np.isclose(denominator, 0.0) @@ -3897,17 +3842,16 @@ class QuickAdapterRegressorV3(BaseRegressionModel): return best_trial_index, best_trial_distance if ( - trial_selection_method == QuickAdapterRegressorV3._DISTANCE_METHODS[0] - ): # "compromise_programming" + trial_selection_method + == QuickAdapterRegressorV3._METHOD_COMPROMISE_PROGRAMMING + ): scores = QuickAdapterRegressorV3._compromise_programming_scores( normalized_matrix[best_cluster_indices], distance_metric, weights=weights, p=p, ) - elif ( - trial_selection_method == QuickAdapterRegressorV3._DISTANCE_METHODS[1] - ): # "topsis" + elif trial_selection_method == QuickAdapterRegressorV3._METHOD_TOPSIS: scores = QuickAdapterRegressorV3._topsis_scores( normalized_matrix[best_cluster_indices], distance_metric, @@ -3957,12 +3901,10 @@ class QuickAdapterRegressorV3(BaseRegressionModel): n_clusters = QuickAdapterRegressorV3._get_n_clusters(normalized_matrix) if cluster_method in { - QuickAdapterRegressorV3._CLUSTER_METHODS[0], # "kmeans" - QuickAdapterRegressorV3._CLUSTER_METHODS[1], # "kmeans2" + QuickAdapterRegressorV3._CLUSTER_KMEANS, + QuickAdapterRegressorV3._CLUSTER_KMEANS2, }: - if ( - cluster_method == QuickAdapterRegressorV3._CLUSTER_METHODS[0] - ): # "kmeans" + if cluster_method == QuickAdapterRegressorV3._CLUSTER_KMEANS: kmeans = sklearn.cluster.KMeans( n_clusters=n_clusters, random_state=42, n_init=10 ) @@ -3974,8 +3916,9 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) if ( - selection_method == QuickAdapterRegressorV3._DISTANCE_METHODS[0] - ): # "compromise_programming" + selection_method + == QuickAdapterRegressorV3._METHOD_COMPROMISE_PROGRAMMING + ): cluster_center_scores = ( QuickAdapterRegressorV3._compromise_programming_scores( cluster_centers, @@ -3983,9 +3926,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): p=p, ) ) - elif ( - selection_method == QuickAdapterRegressorV3._DISTANCE_METHODS[1] - ): # "topsis" + elif selection_method == QuickAdapterRegressorV3._METHOD_TOPSIS: cluster_center_scores = QuickAdapterRegressorV3._topsis_scores( cluster_centers, distance_metric, @@ -4021,9 +3962,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): trial_distances[best_trial_index] = best_trial_distance return trial_distances - elif ( - cluster_method == QuickAdapterRegressorV3._CLUSTER_METHODS[2] - ): # "kmedoids" + elif cluster_method == QuickAdapterRegressorV3._CLUSTER_KMEDOIDS: raise DependencyException( "label_method='kmedoids' is temporarily disabled because " "scikit-learn-extra is not compatible with the current " @@ -4073,9 +4012,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if neighbor_distances.shape[1] < 1: return np.full(n_samples, np.inf) - if ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[0] - ): # "power_mean" + if aggregation == QuickAdapterRegressorV3._DENSITY_AGG_POWER_MEAN: power = ( aggregation_param if aggregation_param is not None @@ -4091,9 +4028,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) assert power is not None return np.asarray(sp.stats.pmean(neighbor_distances, p=power, axis=1)) - elif ( - aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[1] - ): # "quantile" + elif aggregation == QuickAdapterRegressorV3._DENSITY_AGG_QUANTILE: quantile = ( aggregation_param if aggregation_param is not None @@ -4109,9 +4044,9 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) assert quantile is not None return np.asarray(np.nanquantile(neighbor_distances, quantile, axis=1)) - elif aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[2]: # "min" + elif aggregation == QuickAdapterRegressorV3._DENSITY_AGG_MIN: return np.nanmin(neighbor_distances, axis=1) - elif aggregation == QuickAdapterRegressorV3._DENSITY_AGGREGATIONS[3]: # "max" + elif aggregation == QuickAdapterRegressorV3._DENSITY_AGG_MAX: return np.nanmax(neighbor_distances, axis=1) else: raise ValueError( @@ -4359,11 +4294,11 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if n_samples == 1: if method in { - QuickAdapterRegressorV3._SELECTION_METHODS[6], # "medoid" - QuickAdapterRegressorV3._SELECTION_METHODS[2], # "kmeans" - QuickAdapterRegressorV3._SELECTION_METHODS[3], # "kmeans2" - QuickAdapterRegressorV3._SELECTION_METHODS[4], # "kmedoids" - QuickAdapterRegressorV3._SELECTION_METHODS[5], # "knn" + QuickAdapterRegressorV3._SELECTION_MEDOID, + QuickAdapterRegressorV3._SELECTION_KMEANS, + QuickAdapterRegressorV3._SELECTION_KMEANS2, + QuickAdapterRegressorV3._SELECTION_KMEDOIDS, + QuickAdapterRegressorV3._SELECTION_KNN, }: return np.array([0.0]) @@ -4376,16 +4311,14 @@ class QuickAdapterRegressorV3(BaseRegressionModel): mode="none", ) - if ( - method == QuickAdapterRegressorV3._DISTANCE_METHODS[0] - ): # "compromise_programming" + if method == QuickAdapterRegressorV3._METHOD_COMPROMISE_PROGRAMMING: return QuickAdapterRegressorV3._compromise_programming_scores( normalized_matrix, distance_metric, weights=weights, p=p, ) - if method == QuickAdapterRegressorV3._DISTANCE_METHODS[1]: # "topsis" + if method == QuickAdapterRegressorV3._METHOD_TOPSIS: return QuickAdapterRegressorV3._topsis_scores( normalized_matrix, distance_metric, @@ -4424,7 +4357,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): mode="none", ) - if density_method == QuickAdapterRegressorV3._DENSITY_METHODS[0]: # "knn" + if density_method == QuickAdapterRegressorV3._DENSITY_KNN: knn_n_neighbors = int(label_config["n_neighbors"]) knn_aggregation = cast(DensityAggregation, label_config["aggregation"]) if ( @@ -4446,9 +4379,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): aggregation_param=knn_aggregation_param, ) - if ( - density_method == QuickAdapterRegressorV3._DENSITY_METHODS[1] - ): # "medoid" + if density_method == QuickAdapterRegressorV3._DENSITY_MEDOID: return QuickAdapterRegressorV3._pairwise_distance_sums( normalized_matrix, density_metric, @@ -4564,15 +4495,9 @@ class QuickAdapterRegressorV3(BaseRegressionModel): try: study.optimize( objective, - n_trials=self._optuna_config.get( - "n_trials", QuickAdapterRegressorV3.OPTUNA_N_TRIALS_DEFAULT - ), - n_jobs=self._optuna_config.get( - "n_jobs", QuickAdapterRegressorV3.OPTUNA_N_JOBS_DEFAULT - ), - timeout=self._optuna_config.get( - "timeout", QuickAdapterRegressorV3.OPTUNA_TIMEOUT_DEFAULT - ), + n_trials=self._optuna_config["n_trials"], + n_jobs=self._optuna_config["n_jobs"], + timeout=self._optuna_config["timeout"], gc_after_trial=True, ) except Exception as e: @@ -4748,9 +4673,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): storage_dir = self.full_path storage_filename = f"optuna-{pair.split('/')[0]}" storage_backend = self._optuna_config.get("storage") - if ( - storage_backend == QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS[0] - ): # "file" + if storage_backend == QuickAdapterRegressorV3._STORAGE_FILE: journal_path = storage_dir / f"{storage_filename}.log" # Pre-validate EOF: close the read_logs deferred-raise gap (see helper). @@ -4777,9 +4700,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if quarantined is None: raise storage = _build_journal_storage() - elif ( - storage_backend == QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS[1] - ): # "sqlite" + elif storage_backend == QuickAdapterRegressorV3._STORAGE_SQLITE: storage = optuna.storages.RDBStorage( url=f"sqlite:///{storage_dir}/{storage_filename}.sqlite", heartbeat_interval=60, @@ -4799,9 +4720,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) -> optuna.pruners.BasePruner: if is_single_objective: return optuna.pruners.HyperbandPruner( - min_resource=self._optuna_config.get( - "min_resource", QuickAdapterRegressorV3.OPTUNA_MIN_RESOURCE_DEFAULT - ) + min_resource=self._optuna_config["min_resource"] ) else: return optuna.pruners.NopPruner() @@ -4821,37 +4740,23 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) case QuickAdapterRegressorV3._OPTUNA_SAMPLERS.tpe: return optuna.samplers.TPESampler( - n_startup_trials=self._optuna_config.get( - "n_startup_trials", - QuickAdapterRegressorV3.OPTUNA_N_STARTUP_TRIALS_DEFAULT, - ), + n_startup_trials=self._optuna_config["n_startup_trials"], multivariate=True, group=True, - constant_liar=self._optuna_config.get( - "n_jobs", QuickAdapterRegressorV3.OPTUNA_N_JOBS_DEFAULT - ) - > 1, - seed=self._optuna_config.get( - "seed", QuickAdapterRegressorV3.OPTUNA_SEED_DEFAULT - ), + constant_liar=self._optuna_config["n_jobs"] > 1, + seed=self._optuna_config["seed"], ) case QuickAdapterRegressorV3._OPTUNA_SAMPLERS.auto: return optunahub.load_module("samplers/auto_sampler").AutoSampler( - seed=self._optuna_config.get( - "seed", QuickAdapterRegressorV3.OPTUNA_SEED_DEFAULT - ) + seed=self._optuna_config["seed"] ) case QuickAdapterRegressorV3._OPTUNA_SAMPLERS.nsgaii: return optuna.samplers.NSGAIISampler( - seed=self._optuna_config.get( - "seed", QuickAdapterRegressorV3.OPTUNA_SEED_DEFAULT - ), + seed=self._optuna_config["seed"], ) case QuickAdapterRegressorV3._OPTUNA_SAMPLERS.nsgaiii: return optuna.samplers.NSGAIIISampler( - seed=self._optuna_config.get( - "seed", QuickAdapterRegressorV3.OPTUNA_SEED_DEFAULT - ), + seed=self._optuna_config["seed"], ) case _: assert_never(sampler) @@ -4863,16 +4768,12 @@ class QuickAdapterRegressorV3(BaseRegressionModel): if namespace == _OPTUNA_NAMESPACES.hp: return ( QuickAdapterRegressorV3._OPTUNA_HPO_SAMPLERS_SET, - self._optuna_config.get( - "sampler", QuickAdapterRegressorV3._OPTUNA_HPO_SAMPLERS.tpe - ), + self._optuna_config["sampler"], ) elif namespace == _OPTUNA_NAMESPACES.label: return ( QuickAdapterRegressorV3._OPTUNA_LABEL_SAMPLERS_SET, - self._optuna_config.get( - "label_sampler", QuickAdapterRegressorV3._OPTUNA_LABEL_SAMPLERS.auto - ), + self._optuna_config["label_sampler"], ) else: raise ValueError( diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index d37bb9f..c0059dc 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -12,6 +12,7 @@ from typing import ( Literal, Optional, Sequence, + TypeVar, ) import numpy as np @@ -34,6 +35,7 @@ from pandas import DataFrame, Series, isna, to_numeric from scipy.stats import pearsonr, t from technical.pivots_points import pivots_points from Utils import ( + as_dict, _OPTUNA_NAMESPACES, DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES, EXTREMA_COLUMN, @@ -93,6 +95,7 @@ CandleDeviationCacheKey = tuple[ str, DfSignature, float, float, int, InterpolationDirection, float ] CandleThresholdCacheKey = tuple[str, DfSignature, str, int, float, float] +_PairCacheT = TypeVar("_PairCacheT", bound=dict) logger = logging.getLogger(__name__) @@ -125,8 +128,24 @@ class QuickAdapterV3(IStrategy): "inverse", ) _ORDER_TYPES: Final[tuple[OrderType, ...]] = ("entry", "exit") + _TRADE_LONG: Final[str] = _TRADE_DIRECTIONS[0] + _TRADE_SHORT: Final[str] = _TRADE_DIRECTIONS[1] + _ORDER_ENTRY: Final[str] = _ORDER_TYPES[0] + _ORDER_EXIT: Final[str] = _ORDER_TYPES[1] _ORDER_TYPES_SET: Final[frozenset[OrderType]] = frozenset(_ORDER_TYPES) _TRADING_MODES: Final[tuple[TradingMode, ...]] = ("spot", "margin", "futures") + _INTERPOLATION_DIRECT: Final[str] = _INTERPOLATION_DIRECTIONS[0] + _INTERPOLATION_INVERSE: Final[str] = _INTERPOLATION_DIRECTIONS[1] + _TRADING_MODE_SPOT: Final[str] = _TRADING_MODES[0] + _TRADING_MODE_MARGIN: Final[str] = _TRADING_MODES[1] + _TRADING_MODE_FUTURES: Final[str] = _TRADING_MODES[2] + _SMOOTHING_SMM: Final[str] = SMOOTHING_METHODS[5] + _SMOOTHING_SAVGOL: Final[str] = SMOOTHING_METHODS[7] + _SMOOTHING_GAUSSIAN_FILTER1D: Final[str] = SMOOTHING_METHODS[8] + _FILL_EPSILON: Final[str] = FILL_METHODS[1] + _FILL_GAUSSIAN: Final[str] = FILL_METHODS[2] + _FILL_EPSILON_GAUSSIAN: Final[str] = FILL_METHODS[3] + _WEIGHT_NONE: Final[str] = WEIGHT_STRATEGIES[0] _CUSTOM_STOPLOSS_NATR_MULTIPLIER_FRACTION: Final[float] = 0.7860 @@ -342,33 +361,27 @@ class QuickAdapterV3(IStrategy): @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) + return get_label_weighting_config( + as_dict(self.freqai_info.get("label_weighting")), logger + ) @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) + return get_label_smoothing_config( + as_dict(self.freqai_info.get("label_smoothing")), logger + ) @cached_property def trade_price_target_method(self) -> str: - exit_pricing = self.config.get("exit_pricing") - if not isinstance(exit_pricing, dict): - exit_pricing = {} - return get_exit_pricing_config(exit_pricing, logger)[ - "trade_price_target_method" - ] + return get_exit_pricing_config( + as_dict(self.config.get("exit_pricing")), logger + )["trade_price_target_method"] @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 = {} - return get_reversal_confirmation_config(reversal_confirmation, logger) + return get_reversal_confirmation_config( + as_dict(self.config.get("reversal_confirmation")), logger + ) @cached_property def _label_defaults(self) -> tuple[int, float]: @@ -403,8 +416,8 @@ class QuickAdapterV3(IStrategy): if ( col_smoothing_config["method"] in ( - SMOOTHING_METHODS[7], # "savgol" - SMOOTHING_METHODS[8], # "gaussian_filter1d" + QuickAdapterV3._SMOOTHING_SAVGOL, + QuickAdapterV3._SMOOTHING_GAUSSIAN_FILTER1D, ) and col_smoothing_config["mode"] == SMOOTHING_MODES[3] ): # "wrap" @@ -482,8 +495,8 @@ class QuickAdapterV3(IStrategy): fill_method = col_weighting["fill_method"] logger.info(f" fill_method: {fill_method}") if fill_method in ( - FILL_METHODS[1], # "epsilon" - FILL_METHODS[3], # "epsilon_gaussian" + QuickAdapterV3._FILL_EPSILON, + QuickAdapterV3._FILL_EPSILON_GAUSSIAN, ): logger.info( f" fill_epsilon: {format_number(col_weighting['fill_epsilon'])}" @@ -492,8 +505,8 @@ class QuickAdapterV3(IStrategy): f" fill_epsilon_baseline: {col_weighting['fill_epsilon_baseline']}" ) if fill_method in ( - FILL_METHODS[2], # "gaussian" - FILL_METHODS[3], # "epsilon_gaussian" + QuickAdapterV3._FILL_GAUSSIAN, + QuickAdapterV3._FILL_EPSILON_GAUSSIAN, ): logger.info( f" fill_sigma_candles: {format_number(col_weighting['fill_sigma_candles'])}" @@ -531,10 +544,10 @@ class QuickAdapterV3(IStrategy): logger.info(f" sigma: {format_number(col_smoothing['sigma'])}") method = col_smoothing["method"] - if col_weighting["strategy"] != WEIGHT_STRATEGIES[0] and ( # "none" - method == SMOOTHING_METHODS[5] # "smm" + if col_weighting["strategy"] != QuickAdapterV3._WEIGHT_NONE and ( + method == QuickAdapterV3._SMOOTHING_SMM or ( - method == SMOOTHING_METHODS[7] # "savgol" + method == QuickAdapterV3._SMOOTHING_SAVGOL and col_smoothing["polyorder"] >= 2 ) ): @@ -953,7 +966,7 @@ class QuickAdapterV3(IStrategy): # Absent column routes downstream to base-weights-only fallback. is_weighting_active = ( - col_weighting_config["strategy"] != WEIGHT_STRATEGIES[0] # "none" + col_weighting_config["strategy"] != QuickAdapterV3._WEIGHT_NONE and len(label_data.indices) > 0 ) @@ -1091,7 +1104,7 @@ class QuickAdapterV3(IStrategy): dataframe.loc[ reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"], - ] = (1, QuickAdapterV3._TRADE_DIRECTIONS[0]) # "long" + ] = (1, QuickAdapterV3._TRADE_LONG) enter_short_conditions = [ dataframe.get("do_predict") == 1, @@ -1101,7 +1114,7 @@ class QuickAdapterV3(IStrategy): dataframe.loc[ reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"], - ] = (1, QuickAdapterV3._TRADE_DIRECTIONS[1]) # "short" + ] = (1, QuickAdapterV3._TRADE_SHORT) return dataframe @@ -1153,9 +1166,9 @@ class QuickAdapterV3(IStrategy): isna(trade_duration) or trade_duration <= 0 ) - def get_trade_weighted_average_natr( + def _trade_natr_window( self, df: DataFrame, trade: Trade - ) -> Optional[float]: + ) -> Optional[tuple[Any, float, Optional[float]]]: label_natr = df.get("natr_label_period_candles") if label_natr is None or label_natr.empty: return None @@ -1170,10 +1183,22 @@ class QuickAdapterV3(IStrategy): if isna(entry_natr) or entry_natr < 0: return None if len(trade_label_natr) == 1: - return entry_natr - current_natr = trade_label_natr.iloc[-1] - if isna(current_natr) or current_natr < 0: + current_natr = None + else: + current_natr = trade_label_natr.iloc[-1] + if isna(current_natr) or current_natr < 0: + return None + return trade_label_natr, entry_natr, current_natr + + def get_trade_weighted_average_natr( + self, df: DataFrame, trade: Trade + ) -> Optional[float]: + window = self._trade_natr_window(df, trade) + if window is None: return None + trade_label_natr, entry_natr, current_natr = window + if current_natr is None: + return entry_natr median_natr = trade_label_natr.median() trade_label_natr_values = trade_label_natr.to_numpy() @@ -1212,24 +1237,12 @@ class QuickAdapterV3(IStrategy): def get_trade_quantile_interpolation_natr( self, df: DataFrame, trade: Trade ) -> Optional[float]: - label_natr = df.get("natr_label_period_candles") - if label_natr is None or label_natr.empty: - return None - dates = df.get("date") - if dates is None or dates.empty: + window = self._trade_natr_window(df, trade) + if window is None: return None - entry_date = self.get_trade_entry_date(trade) - trade_label_natr = label_natr[dates >= entry_date] - if trade_label_natr.empty: - return None - entry_natr = trade_label_natr.iloc[0] - if isna(entry_natr) or entry_natr < 0: - return None - if len(trade_label_natr) == 1: + trade_label_natr, entry_natr, current_natr = window + if current_natr is None: return entry_natr - current_natr = trade_label_natr.iloc[-1] - if isna(current_natr) or current_natr < 0: - return None trade_volatility_quantile = calculate_quantile( trade_label_natr.to_numpy(), entry_natr ) @@ -1635,6 +1648,14 @@ class QuickAdapterV3(IStrategy): idx = length + idx return min(max(0, idx), length - 1) + def _invalidate_pair_cache( + self, cache: _PairCacheT, pair: str, df_signature: DfSignature + ) -> _PairCacheT: + if self._cached_df_signature.get(pair) != df_signature: + cache = type(cache)({k: v for k, v in cache.items() if k[0] != pair}) + self._cached_df_signature[pair] = df_signature + return cache + def _calculate_candle_deviation( self, df: DataFrame, @@ -1646,12 +1667,9 @@ class QuickAdapterV3(IStrategy): quantile_exponent: float = 1.5, ) -> float: df_signature = QuickAdapterV3._df_signature(df) - prev_df_signature = self._cached_df_signature.get(pair) - if prev_df_signature != df_signature: - self._candle_deviation_cache = { - k: v for k, v in self._candle_deviation_cache.items() if k[0] != pair - } - self._cached_df_signature[pair] = df_signature + self._candle_deviation_cache = self._invalidate_pair_cache( + self._candle_deviation_cache, pair, df_signature + ) cache_key: CandleDeviationCacheKey = ( pair, df_signature, @@ -1684,17 +1702,13 @@ class QuickAdapterV3(IStrategy): if isna(candle_label_natr_value_quantile): return np.nan - if ( - interpolation_direction == QuickAdapterV3._INTERPOLATION_DIRECTIONS[0] - ): # "direct" + if interpolation_direction == QuickAdapterV3._INTERPOLATION_DIRECT: natr_multiplier_fraction = ( min_natr_multiplier_fraction + (max_natr_multiplier_fraction - min_natr_multiplier_fraction) * candle_label_natr_value_quantile**quantile_exponent ) - elif ( - interpolation_direction == QuickAdapterV3._INTERPOLATION_DIRECTIONS[1] - ): # "inverse" + elif interpolation_direction == QuickAdapterV3._INTERPOLATION_INVERSE: natr_multiplier_fraction = ( max_natr_multiplier_fraction - (max_natr_multiplier_fraction - min_natr_multiplier_fraction) @@ -1723,12 +1737,9 @@ class QuickAdapterV3(IStrategy): candle_idx: int = -1, ) -> float: df_signature = QuickAdapterV3._df_signature(df) - prev_df_signature = self._cached_df_signature.get(pair) - if prev_df_signature != df_signature: - self._candle_threshold_cache = { - k: v for k, v in self._candle_threshold_cache.items() if k[0] != pair - } - self._cached_df_signature[pair] = df_signature + self._candle_threshold_cache = self._invalidate_pair_cache( + self._candle_threshold_cache, pair, df_signature + ) cache_key: CandleThresholdCacheKey = ( pair, df_signature, @@ -1762,14 +1773,14 @@ class QuickAdapterV3(IStrategy): is_candle_bullish: bool = candle_close > candle_open is_candle_bearish: bool = candle_close < candle_open - if side == QuickAdapterV3._TRADE_DIRECTIONS[0]: # "long" + if side == QuickAdapterV3._TRADE_LONG: base_price = ( QuickAdapterV3.weighted_close(candle) if is_candle_bearish else candle_close ) candle_threshold = base_price * (1 + current_deviation) - elif side == QuickAdapterV3._TRADE_DIRECTIONS[1]: # "short" + elif side == QuickAdapterV3._TRADE_SHORT: base_price = ( QuickAdapterV3.weighted_close(candle) if is_candle_bullish @@ -1854,18 +1865,14 @@ class QuickAdapterV3(IStrategy): candle_idx=-1, ) current_ok = np.isfinite(current_threshold) and ( - ( - side == QuickAdapterV3._TRADE_DIRECTIONS[0] and rate > current_threshold - ) # "long" - or ( - side == QuickAdapterV3._TRADE_DIRECTIONS[1] and rate < current_threshold - ) # "short" - ) - if order == QuickAdapterV3._ORDER_TYPES[1]: # "exit" - if side == QuickAdapterV3._TRADE_DIRECTIONS[0]: # "long" - trade_direction = QuickAdapterV3._TRADE_DIRECTIONS[1] # "short" - if side == QuickAdapterV3._TRADE_DIRECTIONS[1]: # "short" - trade_direction = QuickAdapterV3._TRADE_DIRECTIONS[0] # "long" + (side == QuickAdapterV3._TRADE_LONG and rate > current_threshold) + or (side == QuickAdapterV3._TRADE_SHORT and rate < current_threshold) + ) + if order == QuickAdapterV3._ORDER_EXIT: + if side == QuickAdapterV3._TRADE_LONG: + trade_direction = QuickAdapterV3._TRADE_SHORT + if side == QuickAdapterV3._TRADE_SHORT: + trade_direction = QuickAdapterV3._TRADE_LONG if not current_ok: logger.debug( f"[{pair}] Denied {trade_direction} {order}: rate {format_number(rate)} did not break threshold {format_number(current_threshold)}" @@ -1902,12 +1909,8 @@ class QuickAdapterV3(IStrategy): ): return current_ok - if ( - side == QuickAdapterV3._TRADE_DIRECTIONS[0] - and not (close_k > threshold_k) # "long" - ) or ( - side == QuickAdapterV3._TRADE_DIRECTIONS[1] - and not (close_k < threshold_k) # "short" + if (side == QuickAdapterV3._TRADE_LONG and not (close_k > threshold_k)) or ( + side == QuickAdapterV3._TRADE_SHORT and not (close_k < threshold_k) ): logger.debug( f"[{pair}] Denied {trade_direction} {order}: " @@ -2083,15 +2086,15 @@ class QuickAdapterV3(IStrategy): trade.set_custom_data("last_outlier_date", last_candle_date.isoformat()) if ( - trade.trade_direction == QuickAdapterV3._TRADE_DIRECTIONS[1] # "short" + trade.trade_direction == QuickAdapterV3._TRADE_SHORT and last_candle.get("do_predict") == 1 and last_candle.get("DI_catch") == 1 and last_candle.get(EXTREMA_COLUMN) < last_candle.get("minima_threshold") and self.reversal_confirmed( df, pair, - QuickAdapterV3._TRADE_DIRECTIONS[0], # "long" - QuickAdapterV3._ORDER_TYPES[1], # "exit" + QuickAdapterV3._TRADE_LONG, + QuickAdapterV3._ORDER_EXIT, current_rate, self.reversal_confirmation["lookback_period_candles"], self.reversal_confirmation["decay_fraction"], @@ -2101,15 +2104,15 @@ class QuickAdapterV3(IStrategy): ): return "minima_detected_short" if ( - trade.trade_direction == QuickAdapterV3._TRADE_DIRECTIONS[0] # "long" + trade.trade_direction == QuickAdapterV3._TRADE_LONG and last_candle.get("do_predict") == 1 and last_candle.get("DI_catch") == 1 and last_candle.get(EXTREMA_COLUMN) > last_candle.get("maxima_threshold") and self.reversal_confirmed( df, pair, - QuickAdapterV3._TRADE_DIRECTIONS[1], # "short" - QuickAdapterV3._ORDER_TYPES[1], # "exit" + QuickAdapterV3._TRADE_SHORT, + QuickAdapterV3._ORDER_EXIT, current_rate, self.reversal_confirmation["lookback_period_candles"], self.reversal_confirmation["decay_fraction"], @@ -2245,11 +2248,9 @@ class QuickAdapterV3(IStrategy): ) -> bool: if side not in QuickAdapterV3._TRADE_DIRECTIONS_SET: return False - if ( - side == QuickAdapterV3._TRADE_DIRECTIONS[1] and not self.can_short - ): # "short" + if side == QuickAdapterV3._TRADE_SHORT and not self.can_short: logger.info( - f"[{pair}] Denied short {QuickAdapterV3._ORDER_TYPES[0]}: shorting not allowed" + f"[{pair}] Denied short {QuickAdapterV3._ORDER_ENTRY}: shorting not allowed" ) return False if Trade.get_open_trade_count() >= self.config.get("max_open_trades", 0): @@ -2268,14 +2269,14 @@ class QuickAdapterV3(IStrategy): ) if df.empty: logger.info( - f"[{pair}] Denied {side} {QuickAdapterV3._ORDER_TYPES[0]}: dataframe is empty" + f"[{pair}] Denied {side} {QuickAdapterV3._ORDER_ENTRY}: dataframe is empty" ) return False if self.reversal_confirmed( df, pair, side, - QuickAdapterV3._ORDER_TYPES[0], # "entry" + QuickAdapterV3._ORDER_ENTRY, rate, self.reversal_confirmation["lookback_period_candles"], self.reversal_confirmation["decay_fraction"], @@ -2288,11 +2289,11 @@ class QuickAdapterV3(IStrategy): def is_short_allowed(self) -> bool: trading_mode = self.config.get("trading_mode") if trading_mode in { - QuickAdapterV3._TRADING_MODES[1], - QuickAdapterV3._TRADING_MODES[2], + QuickAdapterV3._TRADING_MODE_MARGIN, + QuickAdapterV3._TRADING_MODE_FUTURES, }: # margin, futures return True - elif trading_mode == QuickAdapterV3._TRADING_MODES[0]: # "spot" + elif trading_mode == QuickAdapterV3._TRADING_MODE_SPOT: return False else: raise ValueError( diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index ebb81e2..cd3822e 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -925,6 +925,19 @@ SPARSE_TRAINING_MASS_THRESHOLD: Final[float] = 0.05 DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES: Final[int] = 100 +DEFAULT_MIN_LABEL_PERIOD_CANDLES: Final[int] = 12 +DEFAULT_MAX_LABEL_PERIOD_CANDLES: Final[int] = 24 +DEFAULT_MIN_LABEL_NATR_MULTIPLIER: Final[float] = 9.0 +DEFAULT_MAX_LABEL_NATR_MULTIPLIER: Final[float] = 12.0 + + +def as_dict(value: Any) -> dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def enum_error_message(ctx: str, value: Any, options: Sequence[str]) -> str: + return f"Invalid {ctx} value {value!r}: supported values are {', '.join(options)}" + ValidateParamsFn = Callable[[dict[str, Any], Logger, str], dict[str, Any]] @@ -2122,7 +2135,7 @@ def _aggregate_metrics( return np.sum(stacked_metrics * combined_weights, axis=0) else: raise ValueError( - f"Invalid aggregation value {aggregation!r}: supported values are {', '.join(COMBINED_AGGREGATIONS)}" + enum_error_message("aggregation", aggregation, COMBINED_AGGREGATIONS) ) @@ -3840,6 +3853,23 @@ def get_refit_model_training_parameters( return refit_parameters +def _pop_early_stopping_rounds( + model_training_parameters: dict[str, Any], has_eval_set: bool +) -> int | None: + if has_eval_set: + return model_training_parameters.pop( + "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT + ) + model_training_parameters.pop("early_stopping_rounds", None) + return None + + +def _apply_verbosity_alias(model_training_parameters: dict[str, Any]) -> None: + verbosity = model_training_parameters.pop("verbosity", None) + if "verbose" not in model_training_parameters and verbosity is not None: + model_training_parameters["verbose"] = verbosity + + def fit_regressor( regressor: Regressor, X: pd.DataFrame, @@ -3881,13 +3911,9 @@ def fit_regressor( from xgboost import XGBRegressor from xgboost.callback import EarlyStopping - early_stopping_rounds = None - if has_eval_set: - early_stopping_rounds = model_training_parameters.pop( - "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT - ) - else: - model_training_parameters.pop("early_stopping_rounds", None) + early_stopping_rounds = _pop_early_stopping_rounds( + model_training_parameters, has_eval_set + ) if early_stopping_rounds is not None: fit_callbacks.append( @@ -3921,13 +3947,9 @@ def fit_regressor( elif regressor == _REGRESSOR_SPECS.lightgbm.name: from lightgbm import LGBMRegressor, early_stopping - early_stopping_rounds = None - if has_eval_set: - early_stopping_rounds = model_training_parameters.pop( - "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT - ) - else: - model_training_parameters.pop("early_stopping_rounds", None) + early_stopping_rounds = _pop_early_stopping_rounds( + model_training_parameters, has_eval_set + ) if early_stopping_rounds is not None: fit_callbacks.append( @@ -3975,9 +3997,7 @@ def fit_regressor( _EARLY_STOPPING_ROUNDS_DEFAULT ) - verbosity = model_training_parameters.pop("verbosity", None) - if "verbose" not in model_training_parameters and verbosity is not None: - model_training_parameters["verbose"] = verbosity + _apply_verbosity_alias(model_training_parameters) X_val = None y_val = None @@ -4006,19 +4026,13 @@ def fit_regressor( from ngboost import NGBRegressor from sklearn.tree import DecisionTreeRegressor - verbosity = model_training_parameters.pop("verbosity", None) - if "verbose" not in model_training_parameters and verbosity is not None: - model_training_parameters["verbose"] = verbosity + _apply_verbosity_alias(model_training_parameters) model_training_parameters.pop("n_jobs", None) - early_stopping_rounds = None - if has_eval_set: - early_stopping_rounds = model_training_parameters.pop( - "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT - ) - else: - model_training_parameters.pop("early_stopping_rounds", None) + early_stopping_rounds = _pop_early_stopping_rounds( + model_training_parameters, has_eval_set + ) dist = model_training_parameters.pop("dist", "lognormal") @@ -4082,17 +4096,11 @@ def fit_regressor( model_training_parameters.setdefault("thread_count", n_jobs) model_training_parameters.setdefault("max_ctr_complexity", 2) - early_stopping_rounds = None - if has_eval_set: - early_stopping_rounds = model_training_parameters.pop( - "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT - ) - else: - model_training_parameters.pop("early_stopping_rounds", None) + early_stopping_rounds = _pop_early_stopping_rounds( + model_training_parameters, has_eval_set + ) - verbosity = model_training_parameters.pop("verbosity", None) - if "verbose" not in model_training_parameters and verbosity is not None: - model_training_parameters["verbose"] = verbosity + _apply_verbosity_alias(model_training_parameters) pruning_callback = None if trial is not None and has_eval_set and task_type != "GPU": @@ -4124,9 +4132,7 @@ def fit_regressor( if pruning_callback is not None: pruning_callback.check_pruned() else: - raise ValueError( - f"Invalid regressor value {regressor!r}: supported values are {', '.join(REGRESSORS)}" - ) + raise ValueError(enum_error_message("regressor", regressor, REGRESSORS)) return model @@ -4396,9 +4402,7 @@ def get_optuna_study_model_parameters( space_fraction: float, ) -> dict[str, Any]: if regressor not in set(REGRESSORS): - raise ValueError( - f"Invalid regressor value {regressor!r}: supported values are {', '.join(REGRESSORS)}" - ) + raise ValueError(enum_error_message("regressor", regressor, REGRESSORS)) if not isinstance(space_fraction, (int, float)) or not ( 0.0 <= space_fraction <= 1.0 ): @@ -5007,9 +5011,7 @@ def get_optuna_study_model_parameters( return params else: - raise ValueError( - f"Invalid regressor value {regressor!r}: supported values are {', '.join(REGRESSORS)}" - ) + raise ValueError(enum_error_message("regressor", regressor, REGRESSORS)) @lru_cache(maxsize=128) @@ -5114,6 +5116,13 @@ def get_min_max_label_period_candles( return low, high, candles_step +def _validate_step_args(value: float | int, step: int) -> None: + if not isinstance(value, (int, float)): + raise ValueError(f"Invalid value {value!r}: must be an integer or float") + if not isinstance(step, int) or step <= 0: + raise ValueError(f"Invalid step value {step!r}: must be a positive integer") + + @lru_cache(maxsize=128) def round_to_step(value: float | int, step: int) -> int: """ @@ -5123,10 +5132,7 @@ def round_to_step(value: float | int, step: int) -> int: :return: The rounded value. :raises ValueError: If step is not a positive integer or value is not finite. """ - if not isinstance(value, (int, float)): - raise ValueError(f"Invalid value {value!r}: must be an integer or float") - if not isinstance(step, int) or step <= 0: - raise ValueError(f"Invalid step value {step!r}: must be a positive integer") + _validate_step_args(value, step) if isinstance(value, (int, np.integer)): q, r = divmod(value, step) twice_r = r * 2 @@ -5142,10 +5148,7 @@ def round_to_step(value: float | int, step: int) -> int: @lru_cache(maxsize=128) def ceil_to_step(value: float | int, step: int) -> int: - if not isinstance(value, (int, float)): - raise ValueError(f"Invalid value {value!r}: must be an integer or float") - if not isinstance(step, int) or step <= 0: - raise ValueError(f"Invalid step value {step!r}: must be a positive integer") + _validate_step_args(value, step) if isinstance(value, (int, np.integer)): return int(-(-int(value) // step) * step) if not np.isfinite(value): @@ -5155,10 +5158,7 @@ def ceil_to_step(value: float | int, step: int) -> int: @lru_cache(maxsize=128) def floor_to_step(value: float | int, step: int) -> int: - if not isinstance(value, (int, float)): - raise ValueError(f"Invalid value {value!r}: must be an integer or float") - if not isinstance(step, int) or step <= 0: - raise ValueError(f"Invalid step value {step!r}: must be a positive integer") + _validate_step_args(value, step) if isinstance(value, (int, np.integer)): return int((int(value) // step) * step) if not np.isfinite(value): @@ -5244,10 +5244,10 @@ def get_label_defaults( feature_parameters: dict[str, Any], logger: Logger, *, - default_min_label_period_candles: int = 12, - default_max_label_period_candles: int = 24, - default_min_label_natr_multiplier: float = 9.0, - default_max_label_natr_multiplier: float = 12.0, + default_min_label_period_candles: int = DEFAULT_MIN_LABEL_PERIOD_CANDLES, + default_max_label_period_candles: int = DEFAULT_MAX_LABEL_PERIOD_CANDLES, + default_min_label_natr_multiplier: float = DEFAULT_MIN_LABEL_NATR_MULTIPLIER, + default_max_label_natr_multiplier: float = DEFAULT_MAX_LABEL_NATR_MULTIPLIER, ) -> tuple[int, float]: min_label_natr_multiplier = feature_parameters.get( "min_label_natr_multiplier", default_min_label_natr_multiplier -- 2.53.0