Callbacks

deepgboost.callbacks.base_callback.TrainingCallback

Abstract base class for DeepGBoost training callbacks.

Subclass this and override any of the four hook methods you need. All methods have safe default implementations (no-op / False).

Source code in src/deepgboost/callbacks/base_callback.py

class TrainingCallback:
    """
    Abstract base class for DeepGBoost training callbacks.

    Subclass this and override any of the four hook methods you need.
    All methods have safe default implementations (no-op / False).
    """

    def before_training(
        self,
        model,
    ) -> None:
        """Called once before the layer loop begins."""

    def after_training(
        self,
        model,
    ) -> None:
        """Called once after the layer loop ends (or early-stopping)."""

    def before_iteration(
        self,
        model,
        epoch: int,
        evals_log: dict,
    ) -> bool:
        """
        Called at the start of each boosting layer.

        Parameters
        ----------
        model : DGBFModel
            The model being trained.
        epoch : int
            Zero-based layer index.
        evals_log : dict
            ``{dataset_name: {metric_name: latest_value}}`` for all eval sets.

        Returns
        -------
        bool
            ``True`` to stop training before fitting this layer.
        """
        return False

    def after_iteration(
        self,
        model,
        epoch: int,
        evals_log: dict,
    ) -> bool:
        """
        Called at the end of each boosting layer.

        Parameters
        ----------
        model : DGBFModel
            The model being trained.
        epoch : int
            Zero-based layer index.
        evals_log : dict
            ``{dataset_name: {metric_name: latest_value}}`` for all eval sets.

        Returns
        -------
        bool
            ``True`` to stop training after this layer.
        """
        return False

after_iteration(model, epoch, evals_log)

Called at the end of each boosting layer.

Parameters:

Name	Type	Description	Default
model	DGBFModel	The model being trained.	required
epoch	int	Zero-based layer index.	required
evals_log	dict	{dataset_name: {metric_name: latest_value}} for all eval sets.	required

Returns:

Type	Description
bool	True to stop training after this layer.

Source code in src/deepgboost/callbacks/base_callback.py

def after_iteration(
    self,
    model,
    epoch: int,
    evals_log: dict,
) -> bool:
    """
    Called at the end of each boosting layer.

    Parameters
    ----------
    model : DGBFModel
        The model being trained.
    epoch : int
        Zero-based layer index.
    evals_log : dict
        ``{dataset_name: {metric_name: latest_value}}`` for all eval sets.

    Returns
    -------
    bool
        ``True`` to stop training after this layer.
    """
    return False

after_training(model)

Called once after the layer loop ends (or early-stopping).

Source code in src/deepgboost/callbacks/base_callback.py

def after_training(
    self,
    model,
) -> None:
    """Called once after the layer loop ends (or early-stopping)."""

before_iteration(model, epoch, evals_log)

Called at the start of each boosting layer.

Parameters:

Name	Type	Description	Default
model	DGBFModel	The model being trained.	required
epoch	int	Zero-based layer index.	required
evals_log	dict	{dataset_name: {metric_name: latest_value}} for all eval sets.	required

Returns:

Type	Description
bool	True to stop training before fitting this layer.

Source code in src/deepgboost/callbacks/base_callback.py

def before_iteration(
    self,
    model,
    epoch: int,
    evals_log: dict,
) -> bool:
    """
    Called at the start of each boosting layer.

    Parameters
    ----------
    model : DGBFModel
        The model being trained.
    epoch : int
        Zero-based layer index.
    evals_log : dict
        ``{dataset_name: {metric_name: latest_value}}`` for all eval sets.

    Returns
    -------
    bool
        ``True`` to stop training before fitting this layer.
    """
    return False

before_training(model)

Called once before the layer loop begins.

Source code in src/deepgboost/callbacks/base_callback.py

def before_training(
    self,
    model,
) -> None:
    """Called once before the layer loop begins."""

deepgboost.callbacks.early_stopping_callback.EarlyStoppingCallback

Bases: TrainingCallback

Stop training when a monitored metric stops improving.

Parameters:

Name	Type	Description	Default
patience	int	Number of layers with no improvement before stopping.	10
metric	str	Metric key to monitor inside evals_log values. The key is looked up in the first eval set in evals_log.	'train_loss'
data	str or None	Name of the eval dataset to monitor. If None, uses the first dataset found in evals_log.	None
restore_best	bool	If True, restores the model to the best-seen state when stopping.	True
min_delta	float	Minimum change to qualify as an improvement.	1e-06

Source code in src/deepgboost/callbacks/early_stopping_callback.py

class EarlyStoppingCallback(TrainingCallback):
    """
    Stop training when a monitored metric stops improving.

    Parameters
    ----------
    patience : int
        Number of layers with no improvement before stopping.
    metric : str
        Metric key to monitor inside ``evals_log`` values.
        The key is looked up in the *first* eval set in ``evals_log``.
    data : str or None
        Name of the eval dataset to monitor.  If ``None``, uses the first
        dataset found in ``evals_log``.
    restore_best : bool
        If ``True``, restores the model to the best-seen state when stopping.
    min_delta : float
        Minimum change to qualify as an improvement.
    """

    def __init__(
        self,
        patience: int = 10,
        metric: str = "train_loss",
        data: str | None = None,
        restore_best: bool = True,
        min_delta: float = 1e-6,
    ):
        self.patience = patience
        self.metric = metric
        self.data = data
        self.restore_best = restore_best
        self.min_delta = min_delta

        self._best_score: float | None = None
        self._best_epoch: int = 0
        self._best_graph: Any = None
        self._best_weights: Any = None
        self._best_linear: Any = None
        self._wait: int = 0

    def before_training(self, model) -> None:
        self._best_score = None
        self._best_epoch = 0
        self._wait = 0

    def after_iteration(
        self,
        model,
        epoch: int,
        evals_log: dict,
    ) -> bool:
        if not evals_log:
            return False

        # Pick dataset to monitor
        dataset = self.data or next(iter(evals_log))
        if dataset not in evals_log:
            return False

        score = evals_log[dataset].get(self.metric)
        if score is None:
            return False

        # Determine if improvement (lower is better for loss metrics)
        improved = self._best_score is None or score < self._best_score - self.min_delta

        if improved:
            self._best_score = score
            self._best_epoch = epoch
            self._wait = 0
            if self.restore_best:
                self._best_graph = copy.deepcopy(model.graph_)
                self._best_weights = copy.deepcopy(model.weights_)
                self._best_linear = copy.deepcopy(model.linear_models_)
        else:
            self._wait += 1

        if self._wait >= self.patience:
            if self.restore_best and self._best_graph is not None:
                model.graph_ = self._best_graph
                model.weights_ = self._best_weights
                model.linear_models_ = self._best_linear
            return True  # stop

        return False

deepgboost.callbacks.evaluation_monitor_callback.EvaluationMonitorCallback

Bases: TrainingCallback

Print evaluation metrics to stdout after each layer.

Parameters:

Name	Type	Description	Default
period	int	Print every period layers (default 1 = every layer).	1

Source code in src/deepgboost/callbacks/evaluation_monitor_callback.py

class EvaluationMonitorCallback(TrainingCallback):
    """
    Print evaluation metrics to stdout after each layer.

    Parameters
    ----------
    period : int
        Print every ``period`` layers (default 1 = every layer).
    """

    def __init__(
        self,
        period: int = 1,
    ):
        self.period = period

    def after_iteration(
        self,
        model,
        epoch: int,
        evals_log: dict,
    ) -> bool:
        if (epoch + 1) % self.period == 0 and evals_log:
            parts = []
            for dataset, metrics in evals_log.items():
                for metric, val in metrics.items():
                    parts.append(f"{dataset}-{metric}: {val:.6f}")
            print(f"[{epoch + 1}]\t" + "\t".join(parts))
        return False

deepgboost.callbacks.learning_rate_scheduler_callback.LearningRateSchedulerCallback

Bases: TrainingCallback

Adjust model.learning_rate before each boosting layer.

Parameters:

Name	Type	Description	Default
schedule_fn	callable	A function f(epoch: int) -> float that returns the new learning rate for that layer.	required

Examples:

>>> scheduler = LearningRateScheduler(lambda epoch: 0.1 * 0.95**epoch)

Source code in src/deepgboost/callbacks/learning_rate_scheduler_callback.py

class LearningRateSchedulerCallback(TrainingCallback):
    """
    Adjust ``model.learning_rate`` before each boosting layer.

    Parameters
    ----------
    schedule_fn : callable
        A function ``f(epoch: int) -> float`` that returns the new
        learning rate for that layer.

    Examples
    --------
    >>> scheduler = LearningRateScheduler(lambda epoch: 0.1 * 0.95**epoch)
    """

    def __init__(self, schedule_fn):
        self.schedule_fn = schedule_fn

    def before_iteration(
        self,
        model,
        epoch: int,
        evals_log: dict,
    ) -> bool:
        model.learning_rate = float(self.schedule_fn(epoch))
        return False