from sklearn_extra.cluster import KMedoids
from Utils import (
+ DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES,
EXTREMA_COLUMN,
MAXIMA_THRESHOLD_COLUMN,
MINIMA_THRESHOLD_COLUMN,
debug = False
-TEST_SIZE: Final = 0.1
-
warnings.simplefilter(action="ignore", category=FutureWarning)
logger = logging.getLogger(__name__)
https://github.com/sponsors/robcaulk
"""
- version = "3.7.133"
+ version = "3.7.134"
+
+ _TEST_SIZE: Final[float] = 0.1
_SQRT_2: Final[float] = np.sqrt(2.0)
*_CUSTOM_METRICS,
)
+ PREDICTIONS_EXTREMA_THRESHOLD_OUTLIER_DEFAULT: Final[float] = 0.999
+ PREDICTIONS_EXTREMA_THRESHOLDS_ALPHA_DEFAULT: Final[float] = 12.0
+ PREDICTIONS_EXTREMA_EXTREMA_FRACTION_DEFAULT: Final[float] = 1.0
+
+ 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_RATIO_DEFAULT: Final[float] = 9.0
+ MAX_LABEL_NATR_RATIO_DEFAULT: Final[float] = 12.0
+ LABEL_KNN_N_NEIGHBORS_DEFAULT: Final[int] = 5
+
@staticmethod
def _extrema_selection_methods_set() -> set[ExtremaSelectionMethod]:
return set(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)
def _metrics_set() -> set[str]:
return set(QuickAdapterRegressorV3._METRICS)
+ @staticmethod
+ def _get_label_p_order_default(metric: str) -> Optional[float]:
+ if metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]: # "minkowski"
+ return 2.0
+ elif metric == QuickAdapterRegressorV3._CUSTOM_METRICS[7]: # "power_mean"
+ return 1.0
+ return None
+
+ @staticmethod
+ def _get_label_knn_p_order_default(metric: str) -> Optional[float]:
+ if metric == QuickAdapterRegressorV3._CUSTOM_METRICS[12]: # "knn_power_mean"
+ return 1.0
+ elif metric == QuickAdapterRegressorV3._CUSTOM_METRICS[13]: # "knn_quantile"
+ return 0.5
+ return None
+
@cached_property
def _optuna_config(self) -> dict[str, Any]:
optuna_default_config = {
if label_frequency_candles is None:
label_frequency_candles = default_label_frequency_candles
- logger.info(f"{label_frequency_candles=} (default)")
elif isinstance(label_frequency_candles, str):
if label_frequency_candles == "auto":
label_frequency_candles = default_label_frequency_candles
- logger.info(f"{label_frequency_candles=} (auto)")
else:
logger.warning(
f"Invalid string value for label_frequency_candles: '{label_frequency_candles}'. "
f"Only 'auto' is supported. Using fallback"
)
label_frequency_candles = default_label_frequency_candles
- logger.info(f"{label_frequency_candles=} (fallback)")
elif isinstance(label_frequency_candles, (int, float)):
if label_frequency_candles >= 2 and label_frequency_candles <= 10000:
label_frequency_candles = int(label_frequency_candles)
- logger.info(f"{label_frequency_candles=} (configuration)")
else:
logger.warning(
f"Invalid numeric value for label_frequency_candles: {label_frequency_candles}. "
f"Must be between 2 and 10000. Using fallback"
)
label_frequency_candles = default_label_frequency_candles
- logger.info(f"{label_frequency_candles=} (fallback)")
else:
logger.warning(
f"Invalid type for label_frequency_candles: {type(label_frequency_candles).__name__}. "
f"Expected int, float, or 'auto'. Using fallback"
)
label_frequency_candles = default_label_frequency_candles
- logger.info(f"{label_frequency_candles=} (fallback)")
return label_frequency_candles
if not isinstance(predictions_extrema, dict):
predictions_extrema = {}
- threshold_outlier = predictions_extrema.get("threshold_outlier", 0.999)
+ threshold_outlier = predictions_extrema.get(
+ "threshold_outlier",
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_THRESHOLD_OUTLIER_DEFAULT,
+ )
if not isinstance(threshold_outlier, (int, float)) or not (
0 < threshold_outlier < 1
):
- threshold_outlier = 0.999
+ threshold_outlier = (
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_THRESHOLD_OUTLIER_DEFAULT
+ )
selection_method = str(
predictions_extrema.get(
0
] # "mean"
- thresholds_alpha = predictions_extrema.get("thresholds_alpha", 12.0)
+ thresholds_alpha = predictions_extrema.get(
+ "thresholds_alpha",
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_THRESHOLDS_ALPHA_DEFAULT,
+ )
if not isinstance(thresholds_alpha, (int, float)) or thresholds_alpha < 0:
- thresholds_alpha = 12.0
+ thresholds_alpha = (
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_THRESHOLDS_ALPHA_DEFAULT
+ )
- extrema_fraction = predictions_extrema.get("extrema_fraction", 1.0)
+ extrema_fraction = predictions_extrema.get(
+ "extrema_fraction",
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_EXTREMA_FRACTION_DEFAULT,
+ )
if not isinstance(extrema_fraction, (int, float)) or not (
0 < extrema_fraction <= 1
):
- extrema_fraction = 1.0
+ extrema_fraction = (
+ QuickAdapterRegressorV3.PREDICTIONS_EXTREMA_EXTREMA_FRACTION_DEFAULT
+ )
return {
"threshold_outlier": float(threshold_outlier),
self._optuna_hyperopt: Optional[bool] = (
self.freqai_info.get("enabled", False)
and self._optuna_config.get("enabled")
- and self.data_split_parameters.get("test_size", TEST_SIZE) > 0
+ and self.data_split_parameters.get(
+ "test_size", QuickAdapterRegressorV3._TEST_SIZE
+ )
+ > 0
)
self._optuna_hp_value: dict[str, float] = {}
self._optuna_train_value: dict[str, float] = {}
if self.regressor not in set(REGRESSORS):
self.regressor = REGRESSORS[0]
self.freqai_info["regressor"] = self.regressor
+ self._log_model_configuration()
+
+ def _log_model_configuration(self) -> None:
+ logger.info("=" * 60)
+ logger.info("QuickAdapterRegressor Configuration")
+ logger.info("=" * 60)
+
+ logger.info(f"Model Version: {self.version}")
+ logger.info(f"Regressor: {self.regressor}")
+
+ logger.info("Optuna Hyperopt Configuration:")
+ optuna_config = self._optuna_config
+ logger.info(f" enabled: {optuna_config.get('enabled')}")
+ if optuna_config.get("enabled"):
+ logger.info(f" n_jobs: {optuna_config.get('n_jobs')}")
+ logger.info(f" sampler: {optuna_config.get('sampler')}")
+ logger.info(f" storage: {optuna_config.get('storage')}")
+ logger.info(f" continuous: {optuna_config.get('continuous')}")
+ logger.info(f" warm_start: {optuna_config.get('warm_start')}")
+ logger.info(f" n_startup_trials: {optuna_config.get('n_startup_trials')}")
+ logger.info(f" n_trials: {optuna_config.get('n_trials')}")
+ logger.info(f" timeout: {optuna_config.get('timeout')}")
+ logger.info(
+ f" label_candles_step: {optuna_config.get('label_candles_step')}"
+ )
+ logger.info(
+ f" train_candles_step: {optuna_config.get('train_candles_step')}"
+ )
+ logger.info(f" space_reduction: {optuna_config.get('space_reduction')}")
+ logger.info(
+ f" expansion_ratio: {format_number(optuna_config.get('expansion_ratio'))}"
+ )
+ logger.info(f" min_resource: {optuna_config.get('min_resource')}")
+ logger.info(f" seed: {optuna_config.get('seed')}")
+ logger.info(
+ f" label_metric: {self.ft_params.get('label_metric', QuickAdapterRegressorV3._SCIPY_METRICS[2])}"
+ )
+
+ label_weights = self.ft_params.get("label_weights")
+ if label_weights is not None:
+ logger.info(f" label_weights: {label_weights}")
+ else:
+ logger.info(
+ " label_weights: [1.0, ...] * n_objectives, normalized (default)"
+ )
+
+ label_p_order_config = self.ft_params.get("label_p_order")
+ label_metric = self.ft_params.get(
+ "label_metric", QuickAdapterRegressorV3._SCIPY_METRICS[2]
+ )
+
+ label_p_order_is_used = False
+ label_p_order_reason = None
+
+ if label_metric in {
+ QuickAdapterRegressorV3._SCIPY_METRICS[5], # "minkowski"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[7], # "power_mean"
+ }:
+ label_p_order_is_used = True
+ label_p_order_reason = label_metric
+ elif (
+ label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[16]
+ ): # "medoid"
+ label_medoid_metric = self.ft_params.get(
+ "label_medoid_metric",
+ QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean" default
+ )
+ if (
+ label_medoid_metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]
+ ): # "minkowski"
+ label_p_order_is_used = True
+ label_p_order_reason = f"{label_metric} (via label_medoid_metric={label_medoid_metric})"
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[9], # "kmeans"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[10], # "kmeans2"
+ }:
+ label_kmeans_metric = self.ft_params.get(
+ "label_kmeans_metric",
+ QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean" default
+ )
+ if (
+ label_kmeans_metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]
+ ): # "minkowski"
+ label_p_order_is_used = True
+ label_p_order_reason = f"{label_metric} (via label_kmeans_metric={label_kmeans_metric})"
+ elif (
+ label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[11]
+ ): # "kmedoids"
+ label_kmedoids_metric = self.ft_params.get(
+ "label_kmedoids_metric",
+ QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean" default
+ )
+ if (
+ label_kmedoids_metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]
+ ): # "minkowski"
+ label_p_order_is_used = True
+ label_p_order_reason = f"{label_metric} (via label_kmedoids_metric={label_kmedoids_metric})"
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[12], # "knn_power_mean"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[13], # "knn_quantile"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[14], # "knn_min"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[15], # "knn_max"
+ }:
+ label_knn_metric = self.ft_params.get(
+ "label_knn_metric",
+ QuickAdapterRegressorV3._SCIPY_METRICS[5], # "minkowski" default
+ )
+ if (
+ label_knn_metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]
+ ): # "minkowski"
+ label_p_order_is_used = True
+ label_p_order_reason = (
+ f"{label_metric} (via label_knn_metric={label_knn_metric})"
+ )
+
+ if label_p_order_config is not None:
+ logger.info(
+ f" label_p_order: {format_number(float(label_p_order_config))}"
+ )
+ elif label_p_order_is_used:
+ if label_metric in {
+ QuickAdapterRegressorV3._SCIPY_METRICS[5], # "minkowski"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[7], # "power_mean"
+ }:
+ label_p_order_default = (
+ QuickAdapterRegressorV3._get_label_p_order_default(label_metric)
+ )
+ else:
+ label_p_order_default = (
+ QuickAdapterRegressorV3._get_label_p_order_default(
+ QuickAdapterRegressorV3._SCIPY_METRICS[
+ 5
+ ] # "minkowski" default
+ )
+ )
+ logger.info(
+ f" label_p_order: {format_number(label_p_order_default)} (default for {label_p_order_reason})"
+ )
+
+ label_medoid_metric_config = self.ft_params.get("label_medoid_metric")
+ if label_medoid_metric_config is not None:
+ logger.info(f" label_medoid_metric: {label_medoid_metric_config}")
+ elif (
+ label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[16]
+ ): # "medoid"
+ logger.info(
+ f" label_medoid_metric: {QuickAdapterRegressorV3._SCIPY_METRICS[2]} (default for {label_metric})"
+ )
+ label_kmeans_metric_config = self.ft_params.get("label_kmeans_metric")
+ if label_kmeans_metric_config is not None:
+ logger.info(f" label_kmeans_metric: {label_kmeans_metric_config}")
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[9], # "kmeans"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[10], # "kmeans2"
+ }:
+ logger.info(
+ f" label_kmeans_metric: {QuickAdapterRegressorV3._SCIPY_METRICS[2]} (default for {label_metric})"
+ )
+
+ label_kmeans_selection_config = self.ft_params.get("label_kmeans_selection")
+ if label_kmeans_selection_config is not None:
+ logger.info(
+ f" label_kmeans_selection: {label_kmeans_selection_config}"
+ )
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[9], # "kmeans"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[10], # "kmeans2"
+ }:
+ logger.info(
+ f" label_kmeans_selection: min (default for {label_metric})"
+ )
+ label_kmedoids_metric_config = self.ft_params.get("label_kmedoids_metric")
+ if label_kmedoids_metric_config is not None:
+ logger.info(f" label_kmedoids_metric: {label_kmedoids_metric_config}")
+ elif (
+ label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[11]
+ ): # "kmedoids"
+ logger.info(
+ f" label_kmedoids_metric: {QuickAdapterRegressorV3._SCIPY_METRICS[2]} (default for {label_metric})"
+ )
+
+ label_kmedoids_selection_config = self.ft_params.get(
+ "label_kmedoids_selection"
+ )
+ if label_kmedoids_selection_config is not None:
+ logger.info(
+ f" label_kmedoids_selection: {label_kmedoids_selection_config}"
+ )
+ elif (
+ label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[11]
+ ): # "kmedoids"
+ logger.info(
+ f" label_kmedoids_selection: min (default for {label_metric})"
+ )
+
+ label_knn_metric_config = self.ft_params.get("label_knn_metric")
+ if label_knn_metric_config is not None:
+ logger.info(f" label_knn_metric: {label_knn_metric_config}")
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[12], # "knn_power_mean"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[13], # "knn_quantile"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[14], # "knn_min"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[15], # "knn_max"
+ }:
+ logger.info(
+ f" label_knn_metric: {QuickAdapterRegressorV3._SCIPY_METRICS[5]} (default for {label_metric})"
+ )
+
+ label_knn_n_neighbors = self.ft_params.get("label_knn_n_neighbors")
+ if label_knn_n_neighbors is not None:
+ logger.info(f" label_knn_n_neighbors: {label_knn_n_neighbors}")
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[12], # "knn_power_mean"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[13], # "knn_quantile"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[14], # "knn_min"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[15], # "knn_max"
+ }:
+ logger.info(
+ f" label_knn_n_neighbors: {QuickAdapterRegressorV3.LABEL_KNN_N_NEIGHBORS_DEFAULT} (default for {label_metric})"
+ )
+
+ label_knn_p_order_config = self.ft_params.get("label_knn_p_order")
+ if label_knn_p_order_config is not None:
+ logger.info(
+ f" label_knn_p_order: {format_number(float(label_knn_p_order_config))}"
+ )
+ elif label_metric in {
+ QuickAdapterRegressorV3._CUSTOM_METRICS[12], # "knn_power_mean"
+ QuickAdapterRegressorV3._CUSTOM_METRICS[13], # "knn_quantile"
+ }:
+ label_knn_p_order_default = (
+ QuickAdapterRegressorV3._get_label_knn_p_order_default(label_metric)
+ )
+ logger.info(
+ f" label_knn_p_order: {format_number(label_knn_p_order_default)} (default for {label_metric})"
+ )
+
+ logger.info("Predictions Extrema Configuration:")
+ predictions_extrema = self.predictions_extrema
logger.info(
- f"Initialized {self.__class__.__name__} {self.regressor} regressor model version {self.version}"
+ f" selection_method: {predictions_extrema.get('selection_method')}"
+ )
+ logger.info(
+ f" thresholds_smoothing: {predictions_extrema.get('thresholds_smoothing')}"
+ )
+ logger.info(
+ f" threshold_outlier: {format_number(predictions_extrema.get('threshold_outlier'))}"
+ )
+ logger.info(
+ f" thresholds_alpha: {format_number(predictions_extrema.get('thresholds_alpha'))}"
+ )
+ logger.info(
+ f" extrema_fraction: {format_number(predictions_extrema.get('extrema_fraction'))}"
+ )
+
+ logger.info("Label Configuration:")
+ logger.info(
+ f" fit_live_predictions_candles: {self.freqai_info.get('fit_live_predictions_candles', QuickAdapterRegressorV3.FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT)}"
+ )
+ logger.info(f" label_frequency_candles: {self._label_frequency_candles}")
+ logger.info(
+ f" min_label_period_candles: {self.ft_params.get('min_label_period_candles', QuickAdapterRegressorV3.MIN_LABEL_PERIOD_CANDLES_DEFAULT)}"
+ )
+ logger.info(
+ f" max_label_period_candles: {self.ft_params.get('max_label_period_candles', QuickAdapterRegressorV3.MAX_LABEL_PERIOD_CANDLES_DEFAULT)}"
+ )
+ logger.info(
+ f" min_label_natr_ratio: {format_number(self.ft_params.get('min_label_natr_ratio', QuickAdapterRegressorV3.MIN_LABEL_NATR_RATIO_DEFAULT))}"
+ )
+ logger.info(
+ f" max_label_natr_ratio: {format_number(self.ft_params.get('max_label_natr_ratio', QuickAdapterRegressorV3.MAX_LABEL_NATR_RATIO_DEFAULT))}"
)
+ if self._optuna_hyperopt:
+ logger.info("Label Parameters:")
+ for pair in self.pairs:
+ params = self._optuna_label_params.get(pair, {})
+ if params:
+ logger.info(
+ f" {pair}: label_period_candles={params.get('label_period_candles')}, "
+ f"label_natr_ratio={format_number(params.get('label_natr_ratio'))}"
+ )
+ else:
+ logger.info("Label Parameters:")
+ logger.info(
+ f" label_period_candles: {self.ft_params.get('label_period_candles', self._default_label_period_candles)}"
+ )
+ logger.info(
+ f" label_natr_ratio: {format_number(float(self.ft_params.get('label_natr_ratio', self._default_label_natr_ratio)))}"
+ )
+
+ logger.info("=" * 60)
+
def get_optuna_params(self, pair: str, namespace: str) -> dict[str, Any]:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]: # "hp"
params = self._optuna_hp_params.get(pair)
X_test,
y_test,
test_weights,
- self.data_split_parameters.get("test_size", TEST_SIZE),
- self.freqai_info.get("fit_live_predictions_candles", 100),
+ self.data_split_parameters.get(
+ "test_size", QuickAdapterRegressorV3._TEST_SIZE
+ ),
+ self.freqai_info.get(
+ "fit_live_predictions_candles",
+ QuickAdapterRegressorV3.FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT,
+ ),
self._optuna_config.get("train_candles_step"),
model_training_parameters,
),
X_test,
y_test,
test_weights,
- self.data_split_parameters.get("test_size", TEST_SIZE),
+ self.data_split_parameters.get(
+ "test_size", QuickAdapterRegressorV3._TEST_SIZE
+ ),
)
model = fit_regressor(
warmed_up = True
fit_live_predictions_candles = self.freqai_info.get(
- "fit_live_predictions_candles", 100
+ "fit_live_predictions_candles",
+ QuickAdapterRegressorV3.FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT,
)
if self._optuna_hyperopt:
fit_live_predictions_candles,
self._optuna_config.get("label_candles_step"),
min_label_period_candles=self.ft_params.get(
- "min_label_period_candles", 12
+ "min_label_period_candles",
+ QuickAdapterRegressorV3.MIN_LABEL_PERIOD_CANDLES_DEFAULT,
),
max_label_period_candles=self.ft_params.get(
- "max_label_period_candles", 24
+ "max_label_period_candles",
+ QuickAdapterRegressorV3.MAX_LABEL_PERIOD_CANDLES_DEFAULT,
),
min_label_natr_ratio=self.ft_params.get(
- "min_label_natr_ratio", 9.0
+ "min_label_natr_ratio",
+ QuickAdapterRegressorV3.MIN_LABEL_NATR_RATIO_DEFAULT,
),
max_label_natr_ratio=self.ft_params.get(
- "max_label_natr_ratio", 12.0
+ "max_label_natr_ratio",
+ QuickAdapterRegressorV3.MAX_LABEL_NATR_RATIO_DEFAULT,
),
),
directions=[
cdist_kwargs["p"] = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(metric)
)
return sp.spatial.distance.cdist(
normalized_matrix,
7
]: label_p_order # "power_mean"
if label_p_order is not None and np.isfinite(label_p_order)
- else 1.0,
+ else self._get_label_p_order_default(metric),
}[metric]
return sp.stats.pmean(
ideal_point, p=p, weights=np_weights
p = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(label_medoid_metric)
)
return self._pairwise_distance_sums(
normalized_matrix,
cdist_kwargs["p"] = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(label_kmeans_metric)
)
cluster_center_distances_to_ideal = sp.spatial.distance.cdist(
cluster_centers,
p = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(label_kmeans_metric)
)
best_medoid_position = np.nanargmin(
self._pairwise_distance_sums(
cdist_kwargs["p"] = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(label_kmedoids_metric)
)
medoid_distances_to_ideal = sp.spatial.distance.cdist(
normalized_matrix[medoid_indices],
knn_kwargs["p"] = (
label_p_order
if label_p_order is not None and np.isfinite(label_p_order)
- else 2.0
+ else self._get_label_p_order_default(label_knn_metric)
)
knn_kwargs["metric_params"] = {"w": np_weights}
label_knn_p_order = self.ft_params.get("label_knn_p_order")
n_neighbors = (
min(
- int(self.ft_params.get("label_knn_n_neighbors", 5)),
+ int(
+ self.ft_params.get(
+ "label_knn_n_neighbors",
+ QuickAdapterRegressorV3.LABEL_KNN_N_NEIGHBORS_DEFAULT,
+ )
+ ),
n_samples - 1,
)
+ 1
label_knn_p_order = (
label_knn_p_order
if label_knn_p_order is not None and np.isfinite(label_knn_p_order)
- else 1.0
+ else self._get_label_knn_p_order_default(metric)
)
return sp.stats.pmean(neighbor_distances, p=label_knn_p_order, axis=1)
elif (
label_knn_p_order = (
label_knn_p_order
if label_knn_p_order is not None and np.isfinite(label_knn_p_order)
- else 0.5
+ else self._get_label_knn_p_order_default(metric)
)
return np.nanquantile(neighbor_distances, label_knn_p_order, axis=1)
elif metric == QuickAdapterRegressorV3._CUSTOM_METRICS[14]: # "knn_min"
df: pd.DataFrame,
fit_live_predictions_candles: int,
candles_step: int,
- min_label_period_candles: int = 12,
- max_label_period_candles: int = 24,
- min_label_natr_ratio: float = 9.0,
- max_label_natr_ratio: float = 12.0,
+ min_label_period_candles: int = QuickAdapterRegressorV3.MIN_LABEL_PERIOD_CANDLES_DEFAULT,
+ max_label_period_candles: int = QuickAdapterRegressorV3.MAX_LABEL_PERIOD_CANDLES_DEFAULT,
+ min_label_natr_ratio: float = QuickAdapterRegressorV3.MIN_LABEL_NATR_RATIO_DEFAULT,
+ max_label_natr_ratio: float = QuickAdapterRegressorV3.MAX_LABEL_NATR_RATIO_DEFAULT,
) -> tuple[int, float, float, float, float, float]:
min_label_period_candles, max_label_period_candles, candles_step = (
get_min_max_label_period_candles(
from technical.pivots_points import pivots_points
from Utils import (
+ DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES,
DEFAULTS_EXTREMA_SMOOTHING,
DEFAULTS_EXTREMA_WEIGHTING,
EXTREMA_COLUMN,
SMOOTHING_MODES,
STANDARDIZATION_TYPES,
WEIGHT_STRATEGIES,
- TrendDirection,
alligator,
bottom_change_percent,
calculate_n_extrema,
_TRADING_MODES: Final[tuple[TradingMode, ...]] = ("spot", "margin", "futures")
def version(self) -> str:
- return "3.3.183"
+ return "3.3.184"
timeframe = "5m"
2: (0.7640, 0.2),
}
+ CUSTOM_STOPLOSS_NATR_RATIO_PERCENT: Final[float] = 0.7860
+
timeframe_minutes = timeframe_to_minutes(timeframe)
minimal_roi = {str(timeframe_minutes * 864): -1}
# def minimal_roi(self) -> dict[str, Any]:
# timeframe_minutes = timeframe_to_minutes(self.config.get("timeframe", "5m"))
# fit_live_predictions_candles = int(
- # self.config.get("freqai", {}).get("fit_live_predictions_candles", 100)
+ # self.config.get("freqai", {}).get("fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES)
# )
# return {str(timeframe_minutes * fit_live_predictions_candles): -1}
@cached_property
def protections(self) -> list[dict[str, Any]]:
fit_live_predictions_candles = int(
- self.config.get("freqai", {}).get("fit_live_predictions_candles", 100)
+ self.config.get("freqai", {}).get(
+ "fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
+ )
)
protections = self.config.get("custom_protections", {})
trade_duration_candles = int(protections.get("trade_duration_candles", 72))
@cached_property
def startup_candle_count(self) -> int:
# Match the predictions warmup period
- return self.config.get("freqai", {}).get("fit_live_predictions_candles", 100)
+ return self.config.get("freqai", {}).get(
+ "fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
+ )
@cached_property
def max_open_trades_per_side(self) -> int:
self._candle_threshold_cache: dict[CandleThresholdCacheKey, float] = {}
self._cached_df_signature: dict[str, DfSignature] = {}
+ self._log_strategy_configuration()
+
+ def _log_strategy_configuration(self) -> None:
+ logger.info("=" * 60)
+ logger.info("QuickAdapter Strategy Configuration")
+ logger.info("=" * 60)
+
+ logger.info("Extrema Weighting:")
+ logger.info(f" strategy: {self.extrema_weighting['strategy']}")
+ logger.info(f" source_weights: {self.extrema_weighting['source_weights']}")
+ logger.info(f" aggregation: {self.extrema_weighting['aggregation']}")
+ logger.info(
+ f" aggregation_normalization: {self.extrema_weighting['aggregation_normalization']}"
+ )
+ logger.info(f" standardization: {self.extrema_weighting['standardization']}")
+ logger.info(
+ f" robust_quantiles: ({format_number(self.extrema_weighting['robust_quantiles'][0])}, {format_number(self.extrema_weighting['robust_quantiles'][1])})"
+ )
+ logger.info(
+ f" mmad_scaling_factor: {format_number(self.extrema_weighting['mmad_scaling_factor'])}"
+ )
+ logger.info(f" normalization: {self.extrema_weighting['normalization']}")
+ logger.info(
+ f" minmax_range: ({format_number(self.extrema_weighting['minmax_range'][0])}, {format_number(self.extrema_weighting['minmax_range'][1])})"
+ )
+ logger.info(
+ f" sigmoid_scale: {format_number(self.extrema_weighting['sigmoid_scale'])}"
+ )
+ logger.info(
+ f" softmax_temperature: {format_number(self.extrema_weighting['softmax_temperature'])}"
+ )
+ logger.info(f" rank_method: {self.extrema_weighting['rank_method']}")
+ logger.info(f" gamma: {format_number(self.extrema_weighting['gamma'])}")
+
+ logger.info("Extrema Smoothing:")
+ logger.info(f" method: {self.extrema_smoothing['method']}")
+ logger.info(f" window: {self.extrema_smoothing['window']}")
+ logger.info(f" beta: {format_number(self.extrema_smoothing['beta'])}")
+ logger.info(f" polyorder: {self.extrema_smoothing['polyorder']}")
+ logger.info(f" mode: {self.extrema_smoothing['mode']}")
+ logger.info(
+ f" bandwidth: {format_number(self.extrema_smoothing['bandwidth'])}"
+ )
+
+ logger.info("Reversal Confirmation:")
+ logger.info(f" lookback_period: {self._reversal_lookback_period}")
+ logger.info(f" decay_ratio: {format_number(self._reversal_decay_ratio)}")
+ logger.info(
+ f" min_natr_ratio_percent: {format_number(self._reversal_min_natr_ratio_percent)}"
+ )
+ logger.info(
+ f" max_natr_ratio_percent: {format_number(self._reversal_max_natr_ratio_percent)}"
+ )
+
+ exit_pricing = self.config.get("exit_pricing", {})
+ trade_price_target = exit_pricing.get("trade_price_target", "moving_average")
+ logger.info("Exit Pricing:")
+ logger.info(f" trade_price_target: {trade_price_target}")
+ logger.info(f" thresholds_calibration: {self._exit_thresholds_calibration}")
+
+ logger.info("Custom Stoploss:")
+ logger.info(
+ f" natr_ratio_percent: {format_number(QuickAdapterV3.CUSTOM_STOPLOSS_NATR_RATIO_PERCENT)}"
+ )
+
+ logger.info("Partial Exit Stages:")
+ for stage, (
+ natr_ratio_percent,
+ stake_percent,
+ ) in QuickAdapterV3.partial_exit_stages.items():
+ logger.info(
+ f" stage {stage}: natr_ratio_percent={format_number(natr_ratio_percent)}, stake_percent={format_number(stake_percent)}"
+ )
+
+ logger.info("Protections:")
+ if self.protections:
+ for protection in self.protections:
+ method = protection.get("method", "Unknown")
+ logger.info(f" {method}:")
+ for key, value in protection.items():
+ if key != "method":
+ if isinstance(value, (int, float)):
+ logger.info(f" {key}: {format_number(value)}")
+ else:
+ logger.info(f" {key}: {value}")
+ else:
+ logger.info(" No protections enabled")
+
+ logger.info("=" * 60)
+
@staticmethod
def _df_signature(df: DataFrame) -> DfSignature:
n = len(df)
self._reversal_min_natr_ratio_percent = float(min_natr_ratio_percent)
self._reversal_max_natr_ratio_percent = float(max_natr_ratio_percent)
- logger.debug(
- "reversal_confirmation: lookback_period=%s, decay_ratio=%s, natr_ratio_percent_range=(%s, %s)",
- self._reversal_lookback_period,
- format_number(self._reversal_decay_ratio),
- format_number(self._reversal_min_natr_ratio_percent),
- format_number(self._reversal_max_natr_ratio_percent),
- )
-
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict[str, Any], **kwargs
) -> DataFrame:
if df.empty:
return None
- stoploss_distance = self.get_stoploss_distance(df, trade, current_rate, 0.7860)
+ stoploss_distance = self.get_stoploss_distance(
+ df, trade, current_rate, QuickAdapterV3.CUSTOM_STOPLOSS_NATR_RATIO_PERCENT
+ )
if isna(stoploss_distance) or stoploss_distance <= 0:
return None
return stoploss_from_absolute(
repositories / freqai-strategies.git / commitdiff
