)
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,
_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",
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"
"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
@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(
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
)
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(
) -> 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
)
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
QuickAdapterRegressorV3._SELECTION_METHODS_SET,
QuickAdapterRegressorV3._SELECTION_METHODS,
ctx="label_method",
+ mode="raise",
)
category = QuickAdapterRegressorV3._get_selection_category(label_method)
)
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(
)
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",
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,
@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]:
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] = {}
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(
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
QuickAdapterRegressorV3._SCALER_TYPES_SET,
QuickAdapterRegressorV3._SCALER_TYPES,
ctx="scaler",
+ mode="raise",
)
feature_range = self.ft_params.get(
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))
< 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],
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"
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
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(
"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
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)
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,
),
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,
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],
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)
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,
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
)
)
if (
- selection_method == QuickAdapterRegressorV3._DISTANCE_METHODS[0]
- ): # "compromise_programming"
+ selection_method
+ == QuickAdapterRegressorV3._METHOD_COMPROMISE_PROGRAMMING
+ ):
cluster_center_scores = (
QuickAdapterRegressorV3._compromise_programming_scores(
cluster_centers,
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,
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 "
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
)
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
)
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(
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])
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,
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 (
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,
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:
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).
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,
) -> 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()
)
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)
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(
Literal,
Optional,
Sequence,
+ TypeVar,
)
import numpy as np
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,
str, DfSignature, float, float, int, InterpolationDirection, float
]
CandleThresholdCacheKey = tuple[str, DfSignature, str, int, float, float]
+_PairCacheT = TypeVar("_PairCacheT", bound=dict)
logger = logging.getLogger(__name__)
"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
@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]:
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"
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'])}"
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'])}"
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
)
):
# 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
)
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,
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
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
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()
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
)
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,
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,
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)
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,
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
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)}"
):
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}: "
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"],
):
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"],
) -> 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):
)
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"],
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(
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]]
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)
)
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,
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(
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(
_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
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")
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":
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
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
):
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)
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:
"""
: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
@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):
@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):
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
repositories / freqai-strategies.git / commitdiff
