| freqai.feature_parameters.max_label_natr_multiplier | 12.0 | float > 0 | Maximum labeling NATR multiplier used for reversals labeling HPO. (Deprecated alias: `freqai.feature_parameters.max_label_natr_ratio`) |
| freqai.feature_parameters.label_frequency_candles | `auto` | int >= 2 \| `auto` | Reversals labeling frequency. `auto` = max(2, 2 \* number of whitelisted pairs). |
| freqai.feature_parameters.label_metric | `euclidean` | string (supported: `euclidean`,`minkowski`,`cityblock`,`chebyshev`,`mahalanobis`,`seuclidean`,`jensenshannon`,`sqeuclidean`,...) | Metric used in distance calculations to ideal point. |
-| freqai.feature_parameters.label_weights | [1/7,1/7,1/7,1/7,1/7,1/7,1/7] | list[float] | Per-objective weights used in distance calculations to ideal point. Objectives: (1) number of detected reversals, (2) median swing amplitude, (3) median swing amplitude / median volatility-threshold ratio, (4) median swing volume per candle, (5) median swing speed, (6) median swing efficiency ratio, (7) median swing volume-weighted efficiency ratio. |
+| freqai.feature_parameters.label_weights | [1/7,1/7,1/7,1/7,1/7,1/7,1/7] | list[float] | Per-objective weights used in distance calculations to ideal point. Objectives: (1) number of detected reversals, (2) median swing amplitude, (3) median (swing amplitude / median volatility-threshold ratio), (4) median swing volume per candle, (5) median swing speed, (6) median swing efficiency ratio, (7) median swing volume-weighted efficiency ratio. |
| freqai.feature_parameters.label_p_order | `None` | float \| None | p-order used by `minkowski` / `power_mean` (optional). |
| freqai.feature_parameters.label_medoid_metric | `euclidean` | string | Metric used with `medoid`. |
| freqai.feature_parameters.label_kmeans_metric | `euclidean` | string | Metric used for k-means clustering. |
| freqai.predictions_extrema.keep_extrema_fraction | 1.0 | float (0,1] | Fraction of extrema used for thresholds. `1.0` uses all, lower values keep only most significant. Applies to `rank_extrema` and `rank_peaks`; ignored for `partition`. (Deprecated alias: `freqai.predictions_extrema.extrema_fraction`) |
| _Optuna / HPO_ | | | |
| freqai.optuna_hyperopt.enabled | false | bool | Enables HPO. |
-| freqai.optuna_hyperopt.sampler | `tpe` | enum {`tpe`,`auto`} | HPO sampler algorithm. `tpe` uses [TPESampler](https://optuna.readthedocs.io/en/stable/reference/samplers/generated/optuna.samplers.TPESampler.html) with multivariate and group, `auto` uses [AutoSampler](https://hub.optuna.org/samplers/auto_sampler). |
+| freqai.optuna_hyperopt.sampler | `tpe` | enum {`tpe`,`auto`} | HPO sampler algorithm for `hp` and `train` namespaces. `tpe` uses [TPESampler](https://optuna.readthedocs.io/en/stable/reference/samplers/generated/optuna.samplers.TPESampler.html) with multivariate and group, `auto` uses [AutoSampler](https://hub.optuna.org/samplers/auto_sampler). |
+| freqai.optuna_hyperopt.label_sampler | `auto` | enum {`auto`,`tpe`,`nsgaii`,`nsgaiii`} | HPO sampler algorithm for multi-objective `label` namespace. `nsgaii` uses [NSGAIISampler](https://optuna.readthedocs.io/en/stable/reference/samplers/generated/optuna.samplers.NSGAIISampler.html), `nsgaiii` uses [NSGAIIISampler](https://optuna.readthedocs.io/en/stable/reference/samplers/generated/optuna.samplers.NSGAIIISampler.html). |
| freqai.optuna_hyperopt.storage | `file` | enum {`file`,`sqlite`} | HPO storage backend. |
| freqai.optuna_hyperopt.continuous | true | bool | Continuous HPO. |
| freqai.optuna_hyperopt.warm_start | true | bool | Warm start HPO with previous best value(s). |
| freqai.optuna_hyperopt.n_trials | 50 | int >= 1 | Maximum HPO trials. |
| freqai.optuna_hyperopt.n_jobs | CPU threads / 4 | int >= 1 | Parallel HPO workers. |
| freqai.optuna_hyperopt.timeout | 7200 | int >= 0 | HPO wall-clock timeout in seconds. |
-| freqai.optuna_hyperopt.label_candles_step | 1 | int >= 1 | Step for Zigzag NATR horizon search space. |
-| freqai.optuna_hyperopt.train_candles_step | 10 | int >= 1 | Step for training sets size search space. |
-| freqai.optuna_hyperopt.space_reduction | false | bool | Enable/disable HPO search space reduction based on previous best parameters. |
-| freqai.optuna_hyperopt.space_fraction | 0.4 | float [0,1] | Fraction of the HPO search space to use with `space_reduction`. Lower values create narrower search ranges around the best parameters. (Deprecated alias: `freqai.optuna_hyperopt.expansion_ratio`) |
-| freqai.optuna_hyperopt.min_resource | 3 | int >= 1 | Minimum resource per Hyperband pruner rung. |
+| freqai.optuna_hyperopt.label_candles_step | 1 | int >= 1 | Step for Zigzag NATR horizon `label` search space. |
+| freqai.optuna_hyperopt.train_candles_step | 10 | int >= 1 | Step for training sets size `train` search space. |
+| freqai.optuna_hyperopt.space_reduction | false | bool | Enable/disable `hp` search space reduction based on previous best parameters. |
+| freqai.optuna_hyperopt.space_fraction | 0.4 | float [0,1] | Fraction of the `hp` search space to use with `space_reduction`. Lower values create narrower search ranges around the best parameters. (Deprecated alias: `freqai.optuna_hyperopt.expansion_ratio`) |
+| freqai.optuna_hyperopt.min_resource | 3 | int >= 1 | Minimum resource per [HyperbandPruner](https://optuna.readthedocs.io/en/stable/reference/generated/optuna.pruners.HyperbandPruner.html) rung. |
| freqai.optuna_hyperopt.seed | 1 | int >= 0 | HPO RNG seed. |
## ReforceXY
ExtremaSelectionMethod = Literal["rank_extrema", "rank_peaks", "partition"]
OptunaNamespace = Literal["hp", "train", "label"]
+OptunaSampler = Literal["tpe", "auto", "nsgaii", "nsgaiii"]
ClusterSelectionMethod = Literal["medoid", "min"]
CustomThresholdMethod = Literal["median", "soft_extremum"]
SkimageThresholdMethod = Literal[
https://github.com/sponsors/robcaulk
"""
- version = "3.8.3"
+ version = "3.8.4"
_TEST_SIZE: Final[float] = 0.1
) * _OPTUNA_LABEL_N_OBJECTIVES
_OPTUNA_STORAGE_BACKENDS: Final[tuple[str, ...]] = ("file", "sqlite")
- _OPTUNA_SAMPLERS: Final[tuple[str, ...]] = ("tpe", "auto")
+ _OPTUNA_HPO_SAMPLERS: Final[tuple[OptunaSampler, ...]] = ("tpe", "auto")
+ _OPTUNA_LABEL_SAMPLERS: Final[tuple[OptunaSampler, ...]] = (
+ "auto",
+ "tpe",
+ "nsgaii",
+ "nsgaiii",
+ )
+ _OPTUNA_SAMPLERS: Final[tuple[OptunaSampler, ...]] = (
+ "tpe",
+ "auto",
+ "nsgaii",
+ "nsgaiii",
+ )
_OPTUNA_NAMESPACES: Final[tuple[OptunaNamespace, ...]] = ("hp", "train", "label")
_SCIPY_METRICS: Final[tuple[str, ...]] = (
.get("n_jobs", 1),
max(int(self.max_system_threads / 4), 1),
),
- "sampler": QuickAdapterRegressorV3._OPTUNA_SAMPLERS[0], # "tpe"
+ "sampler": QuickAdapterRegressorV3._OPTUNA_HPO_SAMPLERS[0], # "tpe"
"storage": QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS[0], # "file"
"continuous": True,
"warm_start": True,
"n_startup_trials": 15,
"n_trials": 50,
"timeout": 7200,
+ "label_sampler": QuickAdapterRegressorV3._OPTUNA_LABEL_SAMPLERS[
+ 0
+ ], # "auto"
"label_candles_step": 1,
"train_candles_step": 10,
"space_reduction": False,
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" label_sampler: {optuna_config.get('label_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')}")
f" keep_extrema_fraction: {format_number(predictions_extrema.get('keep_extrema_fraction'))}"
)
+ default_label_period_candles, default_label_natr_multiplier = (
+ self._label_defaults
+ )
+ label_period_candles = self.ft_params.get(
+ "label_period_candles", default_label_period_candles
+ )
+ label_natr_multiplier = float(
+ self.ft_params.get("label_natr_multiplier", default_label_natr_multiplier)
+ )
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)}"
)
+ if self._optuna_hyperopt:
+ logger.info(
+ f" label_period_candles: {label_period_candles} (initial value)"
+ )
+ logger.info(
+ f" label_natr_multiplier: {format_number(label_natr_multiplier)} (initial value)"
+ )
logger.info(f" label_frequency_candles: {self._label_frequency_candles}")
logger.info(f" min_label_period_candles: {self._min_label_period_candles}")
logger.info(f" max_label_period_candles: {self._max_label_period_candles}")
f"label_natr_multiplier={format_number(params.get('label_natr_multiplier'))}"
)
else:
- default_label_period_candles, default_label_natr_multiplier = (
- self._label_defaults
- )
logger.info("Label Parameters:")
+ logger.info(f" label_period_candles: {label_period_candles}")
logger.info(
- f" label_period_candles: {self.ft_params.get('label_period_candles', default_label_period_candles)}"
- )
- logger.info(
- f" label_natr_multiplier: {format_number(float(self.ft_params.get('label_natr_multiplier', default_label_natr_multiplier)))}"
+ f" label_natr_multiplier: {format_number(label_natr_multiplier)}"
)
logger.info("=" * 60)
- def get_optuna_params(self, pair: str, namespace: str) -> dict[str, Any]:
+ def get_optuna_params(
+ self, pair: str, namespace: OptunaNamespace
+ ) -> dict[str, Any]:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]: # "hp"
params = self._optuna_hp_params.get(pair)
elif namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[1]: # "train"
return params
def set_optuna_params(
- self, pair: str, namespace: str, params: dict[str, Any]
+ self, pair: str, namespace: OptunaNamespace, params: dict[str, Any]
) -> None:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]: # "hp"
self._optuna_hp_params[pair] = params
f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
)
- def get_optuna_value(self, pair: str, namespace: str) -> float:
+ def get_optuna_value(self, pair: str, namespace: OptunaNamespace) -> float:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]: # "hp"
value = self._optuna_hp_value.get(pair)
elif namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[1]: # "train"
)
return value
- def set_optuna_value(self, pair: str, namespace: str, value: float) -> None:
+ def set_optuna_value(
+ self, pair: str, namespace: OptunaNamespace, value: float
+ ) -> None:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]: # "hp"
self._optuna_hp_value[pair] = value
elif namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[1]: # "train"
f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES[:2])}" # Only hp and train
)
- def get_optuna_values(self, pair: str, namespace: str) -> list[float | int]:
+ def get_optuna_values(
+ self, pair: str, namespace: OptunaNamespace
+ ) -> list[float | int]:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]: # "label"
values = self._optuna_label_values.get(pair)
else:
return values
def set_optuna_values(
- self, pair: str, namespace: str, values: list[float | int]
+ self, pair: str, namespace: OptunaNamespace, values: list[float | int]
) -> None:
if namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]: # "label"
self._optuna_label_values[pair] = values
def optuna_throttle_callback(
self,
pair: str,
- namespace: str,
+ namespace: OptunaNamespace,
callback: Callable[[], Optional[optuna.study.Study]],
) -> None:
if namespace not in {
)
def _get_multi_objective_study_best_trial(
- self, namespace: str, study: optuna.study.Study
+ self, namespace: OptunaNamespace, study: optuna.study.Study
) -> Optional[optuna.trial.FrozenTrial]:
if namespace not in {
QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]
def optuna_optimize(
self,
pair: str,
- namespace: str,
+ namespace: OptunaNamespace,
objective: ObjectiveFuncType,
direction: Optional[optuna.study.StudyDirection] = None,
directions: Optional[list[optuna.study.StudyDirection]] = None,
else:
return optuna.pruners.NopPruner()
- def optuna_create_sampler(self) -> optuna.samplers.BaseSampler:
- sampler = self._optuna_config.get(
- "sampler", QuickAdapterRegressorV3._OPTUNA_SAMPLERS[0]
- )
- if sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[1]: # "auto"
- return optunahub.load_module("samplers/auto_sampler").AutoSampler(
- seed=self._optuna_config.get("seed")
+ def optuna_create_sampler(
+ self, sampler: Optional[OptunaSampler] = None
+ ) -> optuna.samplers.BaseSampler:
+ if sampler is None:
+ sampler = self._optuna_config.get(
+ "sampler",
)
- elif sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[0]: # "tpe"
+ if sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[0]: # "tpe"
return optuna.samplers.TPESampler(
n_startup_trials=self._optuna_config.get("n_startup_trials"),
multivariate=True,
group=True,
seed=self._optuna_config.get("seed"),
)
+ elif sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[1]: # "auto"
+ return optunahub.load_module("samplers/auto_sampler").AutoSampler(
+ seed=self._optuna_config.get("seed")
+ )
+ elif sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[2]: # "nsgaii"
+ return optuna.samplers.NSGAIISampler(
+ seed=self._optuna_config.get("seed"),
+ )
+ elif sampler == QuickAdapterRegressorV3._OPTUNA_SAMPLERS[3]: # "nsgaiii"
+ return optuna.samplers.NSGAIIISampler(
+ seed=self._optuna_config.get("seed"),
+ )
else:
raise ValueError(
f"Invalid optuna sampler {sampler!r}. "
f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_SAMPLERS)}"
)
+ def optuna_samplers_by_namespace(
+ self, namespace: OptunaNamespace
+ ) -> tuple[tuple[OptunaSampler, ...], OptunaSampler]:
+ if namespace in {
+ QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0], # "hp"
+ QuickAdapterRegressorV3._OPTUNA_NAMESPACES[1], # "train"
+ }:
+ return (
+ QuickAdapterRegressorV3._OPTUNA_HPO_SAMPLERS,
+ self._optuna_config.get(
+ "sampler",
+ ),
+ )
+ elif namespace == QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]: # "label"
+ return (
+ QuickAdapterRegressorV3._OPTUNA_LABEL_SAMPLERS,
+ self._optuna_config.get(
+ "label_sampler",
+ ),
+ )
+ else:
+ raise ValueError(
+ f"Invalid namespace {namespace!r}. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
+ )
+
def optuna_create_study(
self,
pair: str,
- namespace: str,
+ namespace: OptunaNamespace,
direction: Optional[optuna.study.StudyDirection] = None,
directions: Optional[list[optuna.study.StudyDirection]] = None,
) -> Optional[optuna.study.Study]:
pair, namespace, study_name, storage
)
+ samplers, sampler = self.optuna_samplers_by_namespace(namespace)
+ if sampler not in set(samplers):
+ raise ValueError(
+ f"Invalid optuna {namespace} sampler {sampler!r}. "
+ f"Supported: {', '.join(samplers)}"
+ )
+
try:
return optuna.create_study(
study_name=study_name,
- sampler=self.optuna_create_sampler(),
+ sampler=self.optuna_create_sampler(sampler),
pruner=self.optuna_create_pruner(is_study_single_objective),
direction=direction,
directions=directions,
return None
def optuna_validate_params(
- self, pair: str, namespace: str, study: Optional[optuna.study.Study]
+ self, pair: str, namespace: OptunaNamespace, study: Optional[optuna.study.Study]
) -> bool:
if not study:
return False
return self.optuna_validate_value(best_value) is not None
def optuna_enqueue_previous_best_params(
- self, pair: str, namespace: str, study: Optional[optuna.study.Study]
+ self, pair: str, namespace: OptunaNamespace, study: Optional[optuna.study.Study]
) -> None:
if not study:
return
exc_info=True,
)
- def optuna_save_best_params(self, pair: str, namespace: str) -> None:
+ def optuna_save_best_params(self, pair: str, namespace: OptunaNamespace) -> None:
best_params_path = Path(
self.full_path / f"optuna-{namespace}-best-params-{pair.split('/')[0]}.json"
)
raise
def optuna_load_best_params(
- self, pair: str, namespace: str
+ self, pair: str, namespace: OptunaNamespace
) -> Optional[dict[str, Any]]:
best_params_path = Path(
self.full_path / f"optuna-{namespace}-best-params-{pair.split('/')[0]}.json"
@staticmethod
def optuna_delete_study(
- pair: str, namespace: str, study_name: str, storage: optuna.storages.BaseStorage
+ pair: str,
+ namespace: OptunaNamespace,
+ study_name: str,
+ storage: optuna.storages.BaseStorage,
) -> None:
try:
optuna.delete_study(study_name=study_name, storage=storage)
repositories / freqai-strategies.git / commitdiff
