Regressor,
calculate_min_extrema,
calculate_n_extrema,
+ eval_set_and_weights,
fit_regressor,
format_number,
get_config_value,
get_label_defaults,
get_min_max_label_period_candles,
- get_optuna_callbacks,
get_optuna_study_model_parameters,
soft_extremum,
zigzag,
https://github.com/sponsors/robcaulk
"""
- version = "3.8.1"
+ version = "3.8.2"
_TEST_SIZE: Final[float] = 0.1
X_test,
y_test,
test_weights,
+ self.data_split_parameters.get(
+ "test_size", QuickAdapterRegressorV3._TEST_SIZE
+ ),
self.get_optuna_params(
dk.pair, QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]
), # "hp"
f"[{dk.pair}] Optuna {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[1]} RMSE {format_number(optuna_train_value)} is not better than {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[0]} RMSE {format_number(optuna_hp_value)}, skipping training sets sizing optimization"
)
- eval_set, eval_weights = QuickAdapterRegressorV3.eval_set_and_weights(
+ eval_set, eval_weights = eval_set_and_weights(
X_test,
y_test,
test_weights,
def optuna_validate_value(value: Any) -> Optional[float]:
return value if isinstance(value, (int, float)) and np.isfinite(value) else None
- @staticmethod
- def eval_set_and_weights(
- X_test: pd.DataFrame,
- y_test: pd.DataFrame,
- test_weights: NDArray[np.floating],
- test_size: float,
- ) -> tuple[
- Optional[list[tuple[pd.DataFrame, pd.DataFrame]]],
- Optional[list[NDArray[np.floating]]],
- ]:
- if test_size == 0:
- eval_set = None
- eval_weights = None
- else:
- eval_set = [(X_test, y_test)]
- eval_weights = [test_weights]
-
- return eval_set, eval_weights
-
def min_max_pred(
self,
pred_df: pd.DataFrame,
if not test_ok or not train_ok:
return np.inf
+ eval_set, eval_weights = eval_set_and_weights(
+ X_test, y_test, test_weights, test_size
+ )
+
model = fit_regressor(
regressor=regressor,
X=X,
y=y,
train_weights=train_weights,
- eval_set=[(X_test, y_test)],
- eval_weights=[test_weights],
+ eval_set=eval_set,
+ eval_weights=eval_weights,
model_training_parameters=model_training_parameters,
- callbacks=get_optuna_callbacks(trial, regressor),
trial=trial,
)
y_pred = model.predict(X_test)
X_test: pd.DataFrame,
y_test: pd.DataFrame,
test_weights: NDArray[np.floating],
+ test_size: float,
model_training_best_parameters: dict[str, Any],
model_training_parameters: dict[str, Any],
space_reduction: bool,
)
model_training_parameters = {**model_training_parameters, **study_model_parameters}
+ eval_set, eval_weights = eval_set_and_weights(
+ X_test, y_test, test_weights, test_size
+ )
+
model = fit_regressor(
regressor=regressor,
X=X,
y=y,
train_weights=train_weights,
- eval_set=[(X_test, y_test)],
- eval_weights=[test_weights],
+ eval_set=eval_set,
+ eval_weights=eval_weights,
model_training_parameters=model_training_parameters,
- callbacks=get_optuna_callbacks(trial, regressor),
trial=trial,
)
y_pred = model.predict(X_test)
from enum import IntEnum
from functools import lru_cache
from logging import Logger
-from typing import Any, Callable, Final, Literal, Optional, TypeVar, Union
+from typing import (
+ TYPE_CHECKING,
+ Any,
+ Callable,
+ Final,
+ Literal,
+ Optional,
+ TypeVar,
+ Union,
+)
import numpy as np
import optuna
from scipy.stats import gmean, percentileofscore
from technical import qtpylib
+if TYPE_CHECKING:
+ from xgboost.callback import TrainingCallback as XGBoostTrainingCallback
+else:
+ XGBoostTrainingCallback = object
+
T = TypeVar("T", pd.Series, float)
"histgradientboostingregressor",
)
+RegressorCallback = Union[Callable[..., Any], XGBoostTrainingCallback]
-def get_optuna_callbacks(
- trial: optuna.trial.Trial, regressor: Regressor
-) -> list[
- Union[
- optuna.integration.XGBoostPruningCallback,
- optuna.integration.LightGBMPruningCallback,
- ]
-]:
- callbacks: list[
- Union[
- optuna.integration.XGBoostPruningCallback,
- optuna.integration.LightGBMPruningCallback,
- ]
- ]
- if regressor == REGRESSORS[0]: # "xgboost"
- callbacks = [
- optuna.integration.XGBoostPruningCallback(trial, "validation_0-rmse")
- ]
- elif regressor == REGRESSORS[1]: # "lightgbm"
- callbacks = [
- optuna.integration.LightGBMPruningCallback(
- trial, "rmse", valid_name="valid_0"
- )
- ]
- elif regressor == REGRESSORS[2]: # "histgradientboostingregressor"
- callbacks = []
- else:
- raise ValueError(
- f"Invalid regressor {regressor!r}. Supported: {', '.join(REGRESSORS)}"
- )
- return callbacks
+
+class HistGradientBoostingPruningCallback:
+ """Optuna pruning callback for HistGradientBoostingRegressor.
+
+ Uses warm_start to train incrementally since sklearn doesn't support
+ training callbacks.
+
+ Args:
+ trial: Optuna trial.
+ metric: Evaluation metric ("rmse").
+ n_iterations_per_step: Iterations between pruning checks.
+ """
+
+ def __init__(
+ self,
+ trial: optuna.trial.Trial,
+ metric: str = "rmse",
+ n_iterations_per_step: int = 10,
+ ) -> None:
+ self._trial = trial
+ self._metric = metric
+ self._n_iterations_per_step = n_iterations_per_step
+ self._is_higher_better = False # RMSE is minimized
+
+ def _validate_study_direction(self) -> None:
+ """Raise ValueError if study direction doesn't match metric direction."""
+ if len(self._trial.study.directions) > 1:
+ return
+
+ direction = self._trial.study.direction
+ if self._is_higher_better:
+ if direction != optuna.study.StudyDirection.MAXIMIZE:
+ raise ValueError(
+ "Metric is higher-better but study direction is MINIMIZE."
+ )
+ else:
+ if direction != optuna.study.StudyDirection.MINIMIZE:
+ raise ValueError(
+ "Metric is lower-better but study direction is MAXIMIZE."
+ )
+
+ def __call__(
+ self,
+ model: Any,
+ X: np.ndarray | pd.DataFrame,
+ y: np.ndarray | pd.DataFrame,
+ X_val: np.ndarray | pd.DataFrame,
+ y_val: np.ndarray | pd.DataFrame,
+ sample_weight: Optional[NDArray[np.floating]] = None,
+ sample_weight_val: Optional[NDArray[np.floating]] = None,
+ ) -> Any:
+ """Train model incrementally with pruning checks.
+
+ Raises optuna.TrialPruned if trial should be pruned.
+ """
+ from sklearn.metrics import root_mean_squared_error
+
+ self._validate_study_direction()
+
+ if isinstance(y, pd.DataFrame):
+ y = y.to_numpy().ravel()
+ if isinstance(y_val, pd.DataFrame):
+ y_val = y_val.to_numpy().ravel()
+
+ max_iter = model.max_iter
+ current_iter = 0
+
+ # Enable warm_start for incremental training
+ model.warm_start = True
+ model.early_stopping = False # Incompatible with warm_start pruning approach
+
+ while current_iter < max_iter:
+ # Set the target iteration for this step
+ next_iter = min(current_iter + self._n_iterations_per_step, max_iter)
+ model.max_iter = next_iter
+
+ model.fit(X, y, sample_weight=sample_weight)
+
+ y_pred = model.predict(X_val)
+ if self._metric == "rmse":
+ current_score = root_mean_squared_error(
+ y_val, y_pred, sample_weight=sample_weight_val
+ )
+ else:
+ raise ValueError(
+ f"Unsupported metric: {self._metric!r}. Supported metrics: 'rmse'."
+ )
+
+ self._trial.report(current_score, step=next_iter)
+
+ if self._trial.should_prune():
+ message = f"Trial was pruned at iteration {next_iter}."
+ raise optuna.TrialPruned(message)
+
+ current_iter = next_iter
+
+ return model
+
+
+_EARLY_STOPPING_ROUNDS_DEFAULT: Final[int] = 50
def fit_regressor(
eval_weights: Optional[list[NDArray[np.floating]]],
model_training_parameters: dict[str, Any],
init_model: Any = None,
- callbacks: Optional[
- list[
- Union[
- optuna.integration.XGBoostPruningCallback,
- optuna.integration.LightGBMPruningCallback,
- ]
- ]
- ] = None,
+ callbacks: Optional[list[RegressorCallback]] = None,
trial: Optional[optuna.trial.Trial] = None,
) -> Any:
+ """Fit a regressor model.
+
+ Args:
+ regressor: Type of regressor.
+ model_training_parameters: Copied internally to avoid side effects.
+ callbacks: Additional callbacks (pruning callbacks added automatically when trial is set).
+ trial: Optuna trial for pruning and random state offset.
+ """
+ model_training_parameters = model_training_parameters.copy()
+ fit_callbacks = list(callbacks) if callbacks else []
+
+ has_eval_set = eval_set is not None and len(eval_set) > 0
+ if not has_eval_set:
+ eval_set = None
+ eval_weights = None
+
if regressor == REGRESSORS[0]: # "xgboost"
from xgboost import XGBRegressor
model_training_parameters.setdefault("random_state", 1)
+ if has_eval_set:
+ model_training_parameters.setdefault(
+ "early_stopping_rounds", _EARLY_STOPPING_ROUNDS_DEFAULT
+ )
+ else:
+ model_training_parameters.pop("early_stopping_rounds", None)
+
if trial is not None:
model_training_parameters["random_state"] = (
model_training_parameters["random_state"] + trial.number
)
+ if has_eval_set:
+ fit_callbacks.append(
+ optuna.integration.XGBoostPruningCallback(
+ trial, "validation_0-rmse"
+ )
+ )
model = XGBRegressor(
objective="reg:squarederror",
eval_metric="rmse",
- callbacks=callbacks,
+ callbacks=fit_callbacks if fit_callbacks else None,
**model_training_parameters,
)
model.fit(
xgb_model=init_model,
)
elif regressor == REGRESSORS[1]: # "lightgbm"
- from lightgbm import LGBMRegressor
+ from lightgbm import LGBMRegressor, early_stopping
model_training_parameters.setdefault("seed", 1)
+ 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 = None
+
if trial is not None:
model_training_parameters["seed"] = (
model_training_parameters["seed"] + trial.number
)
+ if has_eval_set:
+ fit_callbacks.append(
+ optuna.integration.LightGBMPruningCallback(
+ trial, "rmse", valid_name="valid_0"
+ )
+ )
+
+ if early_stopping_rounds is not None:
+ fit_callbacks.append(
+ early_stopping(
+ stopping_rounds=early_stopping_rounds,
+ first_metric_only=True,
+ verbose=False,
+ )
+ )
model = LGBMRegressor(objective="regression", **model_training_parameters)
model.fit(
eval_sample_weight=eval_weights,
eval_metric="rmse",
init_model=init_model,
- callbacks=callbacks,
+ callbacks=fit_callbacks if fit_callbacks else None,
)
elif regressor == REGRESSORS[2]: # "histgradientboostingregressor"
from sklearn.ensemble import HistGradientBoostingRegressor
model_training_parameters.setdefault("random_state", 1)
model_training_parameters.setdefault("loss", "squared_error")
-
- if trial is not None:
- model_training_parameters["random_state"] = (
- model_training_parameters["random_state"] + trial.number
- )
-
model_training_parameters.pop("early_stopping", None)
model_training_parameters.pop("n_jobs", None)
model_training_parameters.pop("l2_regularization_zero", None)
+ early_stopping_rounds = model_training_parameters.pop(
+ "early_stopping_rounds", None
+ )
+ if early_stopping_rounds is not None:
+ model_training_parameters.setdefault(
+ "n_iter_no_change", early_stopping_rounds
+ )
+ else:
+ model_training_parameters.setdefault(
+ "n_iter_no_change", _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
+ if trial is not None:
+ model_training_parameters["random_state"] = (
+ model_training_parameters["random_state"] + trial.number
+ )
+
X_val = None
y_val = None
if eval_set is not None and len(eval_set) > 0:
**model_training_parameters,
)
- model.fit(
- X=X,
- y=y.to_numpy().ravel(),
- sample_weight=train_weights,
- X_val=X_val,
- y_val=y_val,
- sample_weight_val=sample_weight_val,
- )
+ if trial is not None and X_val is not None and y_val is not None:
+ pruning_callback = HistGradientBoostingPruningCallback(trial, metric="rmse")
+ model.early_stopping = False
+ model = pruning_callback(
+ model=model,
+ X=X,
+ y=y,
+ X_val=X_val,
+ y_val=y_val,
+ sample_weight=train_weights,
+ sample_weight_val=sample_weight_val,
+ )
+ else:
+ model.fit(
+ X=X,
+ y=y.to_numpy().ravel(),
+ sample_weight=train_weights,
+ X_val=X_val,
+ y_val=y_val,
+ sample_weight_val=sample_weight_val,
+ )
else:
raise ValueError(
f"Invalid regressor {regressor!r}. Supported: {', '.join(REGRESSORS)}"
return model
+def eval_set_and_weights(
+ X_test: pd.DataFrame,
+ y_test: pd.DataFrame,
+ test_weights: NDArray[np.floating],
+ test_size: float,
+) -> tuple[
+ Optional[list[tuple[pd.DataFrame, pd.DataFrame]]],
+ Optional[list[NDArray[np.floating]]],
+]:
+ if test_size <= 0:
+ return None, None
+
+ return [(X_test, y_test)], [test_weights]
+
+
def _build_int_range(
frange: tuple[float, float],
min_val: int = 1,
repositories / freqai-strategies.git / commitdiff
