Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file modified .gitignore
Binary file not shown.
3 changes: 3 additions & 0 deletions src/hyperactive/experiment/integrations/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,13 @@
TorchExperiment,
)

from .catboost_cv import CatBoostCvExperiment

__all__ = [
"SklearnCvExperiment",
"SkproProbaRegExperiment",
"SktimeClassificationExperiment",
"SktimeForecastingExperiment",
"TorchExperiment",
"CatBoostCvExperiment",
]
120 changes: 120 additions & 0 deletions src/hyperactive/experiment/integrations/catboost_cv.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
from typing import Dict, Any, Optional, Tuple
import pandas as pd
from catboost import Pool, cv
from hyperactive.base._experiment import BaseExperiment

class CatBoostCvExperiment(BaseExperiment):
"""Cross-validation experiment for CatBoost using `catboost.cv()`.

Wraps CatBoost cross-validation so it can be used directly with Hyperactive optimizers.
Returns the final mean test metric over folds (e.g. test-Logloss-mean, test-AUC-mean).

Parameters
----------
pool : catboost.Pool
Data pool with features, labels, and optional weights/groups.
iterations : int, default=100
Maximum boosting iterations.
fold_count : int, default=5
Number of cross-validation folds.
early_stopping_rounds : int or None, default=None
partition_random_seed : int, default=0
type : str, default="Classical"
CV scheme ('Classical', 'TimeSeries', 'Inverted', etc.).
metric : str, default="Logloss"
Metric to extract as the optimization score.
loss_function : str, default="Logloss"
Training objective. If different from `metric`, added as `custom_metric`.
"""

_tags = {
"object_type": "experiment",
"python_dependencies": "catboost",
"property:randomness": "random",
"property:higher_or_lower_is_better": "lower"
}

def __init__(
self,
pool: Pool,
iterations: int = 100,
fold_count: int = 5,
early_stopping_rounds: Optional[int] = None,
partition_random_seed: int = 0,
type: str = "Classical",
metric: str = "Logloss",
loss_function: str = "Logloss",
):
super().__init__()
self.pool = pool
self.iterations = iterations
self.fold_count = fold_count
self.early_stopping_rounds = early_stopping_rounds
self.partition_random_seed = partition_random_seed
self.type = type
self.metric = metric
self.loss_function = loss_function

direction = "lower"
self.set_tags(**{"property:higher_or_lower_is_better": direction})

def _paramnames(self) -> Optional[list]:
return None

def _evaluate(self, params: Dict[str, Any]) -> Tuple[float, Dict[str, Any]]:
"""Run CatBoost CV and return mean test metric + metadata."""
cv_params = params.copy()

cv_params.setdefault("loss_function", self.loss_function)

cv_params.setdefault("iterations", self.iterations)

if self.metric != self.loss_function:
custom_metrics = cv_params.get("custom_metric", [])
if isinstance(custom_metrics, str):
custom_metrics = [custom_metrics]
if self.metric not in custom_metrics:
custom_metrics.append(self.metric)
cv_params["custom_metric"] = custom_metrics

try:
cv_results: pd.DataFrame = cv(
params=cv_params,
pool=self.pool,
fold_count=self.fold_count,
early_stopping_rounds=self.early_stopping_rounds,
partition_random_seed=self.partition_random_seed,
type=self.type,
verbose=False,
plot=False,
return_models=False,
as_pandas=True,
)
except Exception as e:
raise RuntimeError(
f"CatBoost CV failed with params: {cv_params}\n"
f"Error: {str(e)}"
) from e

target_col = f"test-{self.metric}-mean"

if target_col not in cv_results.columns:
available = ", ".join(cv_results.columns)
raise ValueError(
f"Expected column '{target_col}' not found in cv_results.\n"
f"Available columns: {available}\n"
f"Check that metric='{self.metric}' is computed. "
f"Current loss_function: '{cv_params.get('loss_function')}'"
)

mean_score = float(cv_results[target_col].iloc[-1])

metadata = {
"cv_results": cv_results.to_dict(orient="records"),
"final_iteration": int(cv_results["iterations"].iloc[-1]),
"target_column": target_col,
"used_loss_function": cv_params["loss_function"],
"custom_metrics_used": cv_params.get("custom_metric", None),
}

return mean_score, metadata
70 changes: 70 additions & 0 deletions src/hyperactive/tests/test_integrations/test_catboost_cv.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import numpy as np
import pytest
from catboost import Pool
from hyperactive.experiment.integrations import CatBoostCvExperiment

@pytest.fixture
def dummy_binary_pool():
"""Create a small random binary classification dataset as CatBoost Pool."""
np.random.seed(42)
X = np.random.rand(400, 8)
y = np.random.randint(0, 2, 400)
return Pool(data=X, label=y)

def test_catboost_cv_runs_and_returns_valid_score(dummy_binary_pool):
"""Basic sanity test: ensure CatBoostCvExperiment runs cv() correctly."""
exp = CatBoostCvExperiment(
pool=dummy_binary_pool,
metric="Logloss",
fold_count=4,
iterations=50,
early_stopping_rounds=10,
)

params = {
"learning_rate": 0.03,
"depth": 5,
"l2_leaf_reg": 3.0,
}

raw_score, metadata = exp.evaluate(params)
signed_score, _ = exp.score(params)

assert isinstance(raw_score, float)
assert 0 < raw_score < 1, f"Logloss out of expected range: {raw_score:.4f}"
assert signed_score < 0, "Signed score should be negative (lower is better)"
assert exp.get_tag("property:higher_or_lower_is_better") == "lower"

assert isinstance(metadata, dict)
assert "cv_results" in metadata
assert "target_column" in metadata
assert metadata["target_column"] == "test-Logloss-mean"
assert "final_iteration" in metadata
assert metadata["final_iteration"] > 0
assert "used_loss_function" in metadata
assert metadata["used_loss_function"] == "Logloss"

def test_catboost_cv_with_auc(dummy_binary_pool):
"""Test AUC metric with Logloss objective (shows metric vs loss separation)."""
exp = CatBoostCvExperiment(
pool=dummy_binary_pool,
loss_function="Logloss",
metric="AUC",
fold_count=3,
iterations=30,
)

params = {"learning_rate": 0.05, "depth": 4}

raw_score, metadata = exp.evaluate(params)
signed_score, _ = exp.score(params)

# AUC on random data is noisy, but typically around 0.5 ± 0.2–0.3
assert 0.25 < raw_score < 0.75, f"Unexpected AUC on random data: {raw_score:.4f}"

# Optimizing Logloss (lower-better) → tag is "lower", signed_score is negative
assert signed_score < 0, f"Signed score should be negative, got {signed_score:.4f}"

assert exp.get_tag("property:higher_or_lower_is_better") == "lower"
assert "used_loss_function" in metadata
assert metadata["used_loss_function"] == "Logloss"