from scipy.stats import pearsonr, t
from technical.pivots_points import pivots_points
+from ExtremaWeightingTransformer import COMBINED_AGGREGATIONS
from Utils import (
DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES,
DEFAULTS_EXTREMA_SMOOTHING,
EXTREMA_COLUMN,
+ MAXIMA_COLUMN,
MAXIMA_THRESHOLD_COLUMN,
+ MINIMA_COLUMN,
MINIMA_THRESHOLD_COLUMN,
+ SMOOTHED_EXTREMA_COLUMN,
SMOOTHING_METHODS,
SMOOTHING_MODES,
TRADE_PRICE_TARGETS,
zigzag,
zlema,
)
-from ExtremaWeightingTransformer import COMBINED_AGGREGATIONS
TradeDirection = Literal["long", "short"]
InterpolationDirection = Literal["direct", "inverse"]
EXTREMA_COLUMN: {"color": "orange", "type": "line"},
},
"min_max": {
- "smoothed-extrema": {"color": "wheat", "type": "line"},
- "maxima": {"color": "red", "type": "bar"},
- "minima": {"color": "green", "type": "bar"},
+ SMOOTHED_EXTREMA_COLUMN: {"color": "wheat", "type": "line"},
+ MAXIMA_COLUMN: {"color": "red", "type": "bar"},
+ MINIMA_COLUMN: {"color": "green", "type": "bar"},
},
},
}
minutes=len(dataframe) * self.get_timeframe_minutes()
)
dataframe[EXTREMA_COLUMN] = 0.0
- dataframe["minima"] = 0.0
- dataframe["maxima"] = 0.0
+ dataframe[MINIMA_COLUMN] = 0.0
+ dataframe[MAXIMA_COLUMN] = 0.0
if len(pivots_indices) == 0:
logger.warning(
if not np.isfinite(plot_eps):
plot_eps = 0.0
plot_eps = max(float(plot_eps) * 0.5, QuickAdapterV3._PLOT_EXTREMA_MIN_EPS)
- dataframe["maxima"] = (
+ dataframe[MAXIMA_COLUMN] = (
weighted_extrema.where(extrema_direction.gt(0), 0.0)
.clip(lower=0.0)
.mask(extrema_direction.gt(0) & weighted_extrema.eq(0.0), plot_eps)
)
- dataframe["minima"] = (
+ dataframe[MINIMA_COLUMN] = (
weighted_extrema.where(extrema_direction.lt(0), 0.0)
.clip(upper=0.0)
.mask(extrema_direction.lt(0) & weighted_extrema.eq(0.0), -plot_eps)
)
dataframe[EXTREMA_COLUMN] = smoothed_extrema
- dataframe["smoothed-extrema"] = smoothed_extrema
+ dataframe[SMOOTHED_EXTREMA_COLUMN] = smoothed_extrema
return dataframe
MAXIMA_THRESHOLD_COLUMN: Final = "&s-maxima_threshold"
MINIMA_THRESHOLD_COLUMN: Final = "&s-minima_threshold"
+MAXIMA_COLUMN: Final = "maxima"
+MINIMA_COLUMN: Final = "minima"
+SMOOTHED_EXTREMA_COLUMN: Final = "smoothed_extrema"
+
SmoothingKernel = Literal["gaussian", "kaiser", "triang"]
SMOOTHING_KERNELS: Final[tuple[SmoothingKernel, ...]] = (
"gaussian",
)
-Regressor = Literal["xgboost", "lightgbm", "histgradientboostingregressor"]
+Regressor = Literal["xgboost", "lightgbm", "histgradientboostingregressor", "catboost"]
REGRESSORS: Final[tuple[Regressor, ...]] = (
"xgboost",
"lightgbm",
"histgradientboostingregressor",
+ "catboost",
)
RegressorCallback = Union[Callable[..., Any], XGBoostTrainingCallback]
y_val=y_val,
sample_weight_val=sample_weight_val,
)
+ elif regressor == REGRESSORS[3]: # "catboost"
+ from catboost import CatBoostRegressor, Pool
+
+ model_training_parameters.setdefault("random_seed", 1)
+ model_training_parameters.setdefault("loss_function", "RMSE")
+
+ task_type = model_training_parameters.get("task_type", "CPU")
+ if task_type == "GPU":
+ model_training_parameters.setdefault("max_ctr_complexity", 4)
+ model_training_parameters.pop("n_jobs", None)
+ else:
+ n_jobs = model_training_parameters.pop("n_jobs", None)
+ if n_jobs is not None:
+ 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)
+
+ verbosity = model_training_parameters.pop("verbosity", None)
+ if "verbose" not in model_training_parameters and verbosity is not None:
+ model_training_parameters["verbose"] = verbosity
+
+ if trial is not None:
+ model_training_parameters["random_seed"] = (
+ model_training_parameters["random_seed"] + trial.number
+ )
+
+ pruning_callback = None
+ if trial is not None and has_eval_set:
+ pruning_callback = optuna.integration.CatBoostPruningCallback(trial, "RMSE")
+ fit_callbacks.append(pruning_callback)
+
+ model = CatBoostRegressor(**model_training_parameters)
+
+ model.fit(
+ Pool(data=X, label=y, weight=train_weights),
+ eval_set=Pool(
+ data=eval_set[0][0],
+ label=eval_set[0][1],
+ weight=eval_weights[0] if eval_weights else None,
+ )
+ if has_eval_set
+ else None,
+ early_stopping_rounds=early_stopping_rounds
+ if early_stopping_rounds is not None and has_eval_set
+ else None,
+ use_best_model=True
+ if early_stopping_rounds is not None and has_eval_set
+ else False,
+ callbacks=fit_callbacks if fit_callbacks else None,
+ )
+
+ if pruning_callback is not None:
+ pruning_callback.check_pruned()
else:
raise ValueError(
f"Invalid regressor value {regressor!r}: supported values are {', '.join(REGRESSORS)}"
trial: optuna.trial.Trial,
regressor: Regressor,
model_training_best_parameters: dict[str, Any],
+ model_training_parameters: dict[str, Any],
space_reduction: bool,
space_fraction: float,
) -> dict[str, Any]:
),
}
+ elif regressor == REGRESSORS[3]: # "catboost"
+ # Parameter order: boosting -> tree structure -> regularization -> sampling
+ task_type = model_training_parameters.get("task_type", "CPU")
+ if task_type == "GPU":
+ default_ranges: dict[str, tuple[float, float]] = {
+ # Boosting/Training
+ "iterations": (100, 2000),
+ "learning_rate": (0.001, 0.3),
+ # Tree structure
+ "depth": (4, 12),
+ "min_data_in_leaf": (1, 20),
+ "border_count": (32, 254),
+ "max_ctr_complexity": (2, 6),
+ # Regularization
+ "l2_leaf_reg": (1, 10),
+ "model_size_reg": (0.0, 1.0),
+ # Sampling/Randomization
+ "bagging_temperature": (0, 10),
+ "random_strength": (1, 20),
+ "rsm": (0.5, 1.0),
+ "subsample": (0.6, 1.0),
+ }
+ bootstrap_options = ["Bayesian", "Bernoulli"]
+ else: # CPU
+ default_ranges: dict[str, tuple[float, float]] = {
+ # Boosting/Training
+ "iterations": (100, 2000),
+ "learning_rate": (0.001, 0.3),
+ # Tree structure
+ "depth": (4, 10),
+ "min_data_in_leaf": (1, 20),
+ # Regularization
+ "l2_leaf_reg": (1, 10),
+ "model_size_reg": (0.0, 1.0),
+ # Sampling/Randomization
+ "bagging_temperature": (0, 10),
+ "random_strength": (1, 20),
+ "rsm": (0.5, 1.0),
+ "subsample": (0.6, 1.0),
+ }
+ bootstrap_options = ["Bayesian", "Bernoulli", "MVS"]
+
+ log_scaled_params = {
+ "iterations",
+ "learning_rate",
+ }
+
+ ranges = _build_ranges(default_ranges, log_scaled_params)
+
+ bootstrap_type = trial.suggest_categorical("bootstrap_type", bootstrap_options)
+
+ params = {
+ # Boosting/Training
+ "iterations": _optuna_suggest_int_from_range(
+ trial, "iterations", ranges["iterations"], min_val=1, log=True
+ ),
+ "learning_rate": trial.suggest_float(
+ "learning_rate",
+ ranges["learning_rate"][0],
+ ranges["learning_rate"][1],
+ log=True,
+ ),
+ # Tree structure
+ "depth": _optuna_suggest_int_from_range(
+ trial, "depth", ranges["depth"], min_val=1
+ ),
+ "min_data_in_leaf": _optuna_suggest_int_from_range(
+ trial, "min_data_in_leaf", ranges["min_data_in_leaf"], min_val=1
+ ),
+ "grow_policy": trial.suggest_categorical(
+ "grow_policy", ["SymmetricTree", "Depthwise", "Lossguide"]
+ ),
+ # Regularization
+ "l2_leaf_reg": trial.suggest_float(
+ "l2_leaf_reg", ranges["l2_leaf_reg"][0], ranges["l2_leaf_reg"][1]
+ ),
+ "model_size_reg": trial.suggest_float(
+ "model_size_reg",
+ ranges["model_size_reg"][0],
+ ranges["model_size_reg"][1],
+ ),
+ # Sampling/Randomization
+ "bootstrap_type": bootstrap_type,
+ "random_strength": trial.suggest_float(
+ "random_strength",
+ ranges["random_strength"][0],
+ ranges["random_strength"][1],
+ ),
+ "rsm": trial.suggest_float(
+ "rsm",
+ ranges["rsm"][0],
+ ranges["rsm"][1],
+ ),
+ }
+
+ if bootstrap_type == "Bayesian":
+ params["bagging_temperature"] = trial.suggest_float(
+ "bagging_temperature",
+ ranges["bagging_temperature"][0],
+ ranges["bagging_temperature"][1],
+ )
+
+ if bootstrap_type in ["Bernoulli", "MVS"]:
+ params["subsample"] = trial.suggest_float(
+ "subsample",
+ ranges["subsample"][0],
+ ranges["subsample"][1],
+ )
+
+ if task_type == "GPU":
+ params["border_count"] = _optuna_suggest_int_from_range(
+ trial, "border_count", ranges["border_count"], min_val=1
+ )
+ params["max_ctr_complexity"] = _optuna_suggest_int_from_range(
+ trial,
+ "max_ctr_complexity",
+ ranges["max_ctr_complexity"],
+ min_val=1,
+ )
+
+ return params
+
else:
raise ValueError(
f"Invalid regressor value {regressor!r}: supported values are {', '.join(REGRESSORS)}"
repositories / freqai-strategies.git / commitdiff
