diff --git a/aeon/classification/dictionary_based/_redcomets.py b/aeon/classification/dictionary_based/_redcomets.py index 2efa220e90..2e05d0abf6 100644 --- a/aeon/classification/dictionary_based/_redcomets.py +++ b/aeon/classification/dictionary_based/_redcomets.py @@ -160,7 +160,7 @@ def _fit(self, X, y): self.sax_clfs, ) = self._build_univariate_ensemble(X_concat, y) - elif self.variant in [4, 5, 6, 7, 8, 9]: # Ensemble + else: # Ensemble (variants 4-9) ( self.sfa_transforms, self.sfa_clfs, @@ -396,8 +396,8 @@ def _predict_proba(self, X) -> np.ndarray: if self.variant in [1, 2, 3]: # Concatenate X_concat = X.reshape(*X.shape[:-2], -1) return self._predict_proba_unvivariate(X_concat) - elif self.variant in [4, 5, 6, 7, 8, 9]: - return self._predict_proba_dimension_ensemble(X) # Ensemble + else: # Ensemble (variants 4-9) + return self._predict_proba_dimension_ensemble(X) def _predict_proba_unvivariate(self, X) -> np.ndarray: """Predicts labels probabilities for sequences in univariate X. @@ -473,8 +473,7 @@ def _predict_proba_dimension_ensemble(self, X) -> np.ndarray: if self.variant in [6, 7, 8, 9]: dimension_pred_mats = None for sfa, (rf, _) in zip(sfa_transforms, sfa_clfs): - sfa_dics = sfa.transform_words(X_d) - X_sfa = sfa_dics[:, 0, :] + X_sfa = sfa.transform_words(X_d)[0] rf_pred_mat = rf.predict_proba(X_sfa) @@ -486,7 +485,7 @@ def _predict_proba_dimension_ensemble(self, X) -> np.ndarray: (ensemble_pred_mats, [rf_pred_mat]) ) - elif self.variant in [6, 7, 8, 9]: + else: # variants 6-9 if dimension_pred_mats is None: dimension_pred_mats = [rf_pred_mat] else: @@ -507,7 +506,7 @@ def _predict_proba_dimension_ensemble(self, X) -> np.ndarray: (ensemble_pred_mats, [rf_pred_mat]) ) - elif self.variant in [6, 7, 8, 9]: + else: # variants 6-9 if dimension_pred_mats is None: dimension_pred_mats = [rf_pred_mat] else: @@ -518,7 +517,7 @@ def _predict_proba_dimension_ensemble(self, X) -> np.ndarray: if self.variant in [6, 7, 8, 9]: if self.variant in [6, 7]: fused_dimension_pred_mat = np.sum(dimension_pred_mats, axis=0) - elif self.variant in [8, 9]: + else: # variants 8, 9 weights = np.array( [np.mean(mat.max(axis=1)) for mat in dimension_pred_mats] ).reshape(-1, 1) @@ -535,7 +534,7 @@ def _predict_proba_dimension_ensemble(self, X) -> np.ndarray: if self.variant in [4, 6, 7]: pred_mat = np.sum(np.array(ensemble_pred_mats), axis=0) - elif self.variant in [5, 8, 9]: + else: # variants 5, 8, 9 weights = np.array( [np.mean(mat.max(axis=1)) for mat in ensemble_pred_mats] ).reshape(-1, 1) diff --git a/aeon/classification/dictionary_based/tests/test_redcomets.py b/aeon/classification/dictionary_based/tests/test_redcomets.py index 8db7253a1c..a13c87292d 100644 --- a/aeon/classification/dictionary_based/tests/test_redcomets.py +++ b/aeon/classification/dictionary_based/tests/test_redcomets.py @@ -1,107 +1,157 @@ """REDCOMETS test code.""" -__maintainer__ = [] - -# from sys import platform -# -# import numpy as np -# import pytest -# -# from aeon.classification.dictionary_based import REDCOMETS -# from aeon.datasets import load_basic_motions, load_unit_test -# from aeon.utils.validation._dependencies import _check_soft_dependencies -# -# -# @pytest.mark.skipif( -# not _check_soft_dependencies( -# "imbalanced-learn", -# package_import_alias={"imbalanced-learn": "imblearn"}, -# severity="none", -# ), -# reason="skip test if required soft dependency imbalanced-learn not available", -# ) -# def test_redcomets_score_univariate(): -# """Test of REDCOMETS train estimate on unit test data.""" -# # load unit test data -# X_train, y_train = load_unit_test(split="train") -# X_test, y_test = load_unit_test(split="test") -# -# def test_variant(v, expected_result): -# # train REDCOMETS- -# redcomets = REDCOMETS(variant=v, n_trees=3, random_state=0) -# redcomets.fit(X_train, y_train) -# -# score = redcomets.score(X_test, y_test) -# -# assert isinstance(score, float) -# -# # We cannot guarantee same results on ARM macOS -# if platform != "darwin": -# np.testing.assert_almost_equal(score, expected_result, decimal=4) -# -# test_variant(1, 0.7272) -# test_variant(2, 0.6818) -# test_variant(3, 0.7272) -# -# -# @pytest.mark.skipif( -# not _check_soft_dependencies( -# "imbalanced-learn", -# package_import_alias={"imbalanced-learn": "imblearn"}, -# severity="none", -# ), -# reason="skip test if required soft dependency imbalanced-learn not available", -# ) -# def test_redcomets_score_multivariate(): -# """Test of REDCOMETS train estimate on unit test data.""" -# # load unit test data -# X_train, y_train = load_basic_motions(split="train") -# X_test, y_test = load_basic_motions(split="test") -# -# def test_variant(v, expected_result): -# # train REDCOMETS- -# redcomets = REDCOMETS(variant=v, n_trees=3, random_state=0) -# redcomets.fit(X_train, y_train) -# -# score = redcomets.score(X_test, y_test) -# -# assert isinstance(score, float) -# -# # We cannot guarantee same results on ARM macOS -# if platform != "darwin": -# np.testing.assert_almost_equal(score, expected_result, decimal=4) -# -# test_variant(1, 0.95) -# test_variant(2, 0.975) -# test_variant(3, 0.975) -# test_variant(4, 0.875) -# test_variant(5, 0.875) -# test_variant(6, 0.875) -# test_variant(7, 0.875) -# test_variant(8, 0.875) -# test_variant(9, 0.875) -# -# -# @pytest.mark.skipif( -# not _check_soft_dependencies( -# "imbalanced-learn", -# package_import_alias={"imbalanced-learn": "imblearn"}, -# severity="none", -# ), -# reason="skip test if required soft dependency imbalanced-learn not available", -# ) -# def test_redcomets_lens_generation(): -# """Test of REDCOMETS random lens generation.""" -# # load unit test data -# X, y = load_unit_test() -# -# # Generate 10 random lenses -# redcomets = REDCOMETS(random_state=0) -# lenses = redcomets._get_random_lenses(np.squeeze(X), 10) -# -# assert len(lenses) == 10 -# assert isinstance(lenses, list) -# -# for w, a in lenses: -# assert isinstance(w, int) -# assert isinstance(a, int) +import numpy as np +import pytest +from sklearn.utils import check_random_state + +from aeon.classification.dictionary_based import REDCOMETS + +N_PER_CLASS = 10 +N_TIMEPOINTS = 48 # long enough to yield >=2 SFA and >=2 SAX lenses per view +N_CLASSES = 2 + + +def _labelled_panel(n_channels, n_per_class=N_PER_CLASS, random_state=0): + """Return a balanced random panel with ``N_CLASSES`` classes. + + Values are random: the tests assert output structure (shape, valid labels, + normalised probabilities), not classification accuracy. + """ + rng = check_random_state(random_state) + n_cases = n_per_class * N_CLASSES + X = rng.standard_normal((n_cases, n_channels, N_TIMEPOINTS)) + y = np.repeat(np.arange(N_CLASSES), n_per_class) + return X, y + + +def _assert_valid_output(clf, X): + """Check output structure: normalised probabilities and in-vocabulary labels.""" + proba = clf.predict_proba(X) + pred = clf.predict(X) + + assert proba.shape == (X.shape[0], clf.n_classes_) + np.testing.assert_allclose(proba.sum(axis=1), 1.0) + assert pred.shape == (X.shape[0],) + assert set(pred).issubset(set(clf.classes_)) + + +@pytest.mark.parametrize("variant", [1, 2, 3]) +def test_redcomets_univariate_variants(variant): + """Univariate variants 1-3 fit and produce well-formed predictions.""" + X, y = _labelled_panel(n_channels=1) + clf = REDCOMETS(variant=variant, n_trees=3, random_state=0) + clf.fit(X, y) + _assert_valid_output(clf, X) + + +@pytest.mark.parametrize("variant", [1, 2, 3]) +def test_redcomets_multivariate_concatenate_variants(variant): + """Variants 1-3 handle multivariate input by concatenating channels.""" + X, y = _labelled_panel(n_channels=3) + clf = REDCOMETS(variant=variant, n_trees=3, random_state=0) + clf.fit(X, y) + _assert_valid_output(clf, X) + + +@pytest.mark.parametrize("variant", [4, 5, 6, 7, 8, 9]) +def test_redcomets_dimension_ensemble_variants(variant): + """Variants 4-9 build and fuse a per-channel ensemble on multivariate input. + + These variants exercise the dimension-ensemble build and the variant-specific + fusion in ``_predict_proba_dimension_ensemble`` (plain sum vs. confidence + weighting, at both the per-channel and cross-channel stages). + """ + X, y = _labelled_panel(n_channels=3) + clf = REDCOMETS(variant=variant, n_trees=3, random_state=0) + clf.fit(X, y) + _assert_valid_output(clf, X) + + +def test_redcomets_deterministic(): + """A fixed random_state gives identical predictions across fits.""" + X, y = _labelled_panel(n_channels=3) + + pred1 = REDCOMETS(variant=5, n_trees=3, random_state=0).fit(X, y).predict(X) + pred2 = REDCOMETS(variant=5, n_trees=3, random_state=0).fit(X, y).predict(X) + + np.testing.assert_array_equal(pred1, pred2) + + +def test_redcomets_balanced_input_needs_no_oversampling(): + """Already-balanced classes fit without invoking the oversampling branch.""" + X, y = _labelled_panel(n_channels=1) # N_PER_CLASS each, balanced + assert np.unique(y, return_counts=True)[1].tolist() == [N_PER_CLASS, N_PER_CLASS] + + clf = REDCOMETS(variant=1, n_trees=3, random_state=0) + clf.fit(X, y) + _assert_valid_output(clf, X) + + +def test_redcomets_imbalanced_input_uses_smote(): + """An imbalanced class large enough for neighbour search is SMOTE-oversampled. + + The minority class has more than five samples, exercising the capped + neighbour-count SMOTE path rather than the fallback. + """ + X, _ = _labelled_panel(n_channels=1, n_per_class=14) + y = np.array([0] * 20 + [1] * 8) # minority > 5 -> capped SMOTE neighbours + + clf = REDCOMETS(variant=1, n_trees=3, random_state=0) + clf.fit(X, y) + assert set(clf.classes_) == {0, 1} + _assert_valid_output(clf, X) + + +def test_redcomets_tiny_minority_uses_random_oversampler(): + """A minority class too small for SMOTE falls back to random oversampling. + + With two minority samples the SMOTE neighbour count drops below one, so + REDCOMETS must fall back to RandomOverSampler and still fit on both classes. + """ + X, _ = _labelled_panel(n_channels=1, n_per_class=10) + y = np.array([0] * 18 + [1] * 2) + + clf = REDCOMETS(variant=1, n_trees=3, random_state=0) + clf.fit(X, y) + assert set(clf.classes_) == {0, 1} + _assert_valid_output(clf, X) + + +@pytest.mark.parametrize("bad_variant", [0, 10]) +def test_redcomets_rejects_invalid_variant(bad_variant): + """Variants outside 1-9 are rejected at construction.""" + with pytest.raises(AssertionError): + REDCOMETS(variant=bad_variant) + + +@pytest.mark.parametrize("bad_perc", [0, 101]) +def test_redcomets_rejects_invalid_perc_length(bad_perc): + """perc_length must lie in (0, 100].""" + with pytest.raises(AssertionError): + REDCOMETS(perc_length=bad_perc) + + +def test_redcomets_univariate_rejects_ensemble_variant(): + """Dimension-ensemble variants 4-9 require multivariate input.""" + X, y = _labelled_panel(n_channels=1) + clf = REDCOMETS(variant=4, n_trees=3, random_state=0) + with pytest.raises(AssertionError): + clf.fit(X, y) + + +def test_redcomets_test_params_are_valid(): + """The documented test parameters construct a valid REDCOMETS instance.""" + params = REDCOMETS._get_test_params() + assert params["variant"] in range(1, 10) + REDCOMETS(**params) # construction asserts pass + + +def test_redcomets_declares_no_imbalanced_learn_dependency(): + """REDCOMETS no longer depends on imbalanced-learn (gh-3654).""" + deps = REDCOMETS(random_state=0).get_tag("python_dependencies", None) + if deps is None: + deps = [] + if isinstance(deps, str): + deps = [deps] + assert "imblearn" not in deps + assert "imbalanced-learn" not in deps diff --git a/aeon/classification/dictionary_based/tests/test_redcomets_oversampling.py b/aeon/classification/dictionary_based/tests/test_redcomets_oversampling.py deleted file mode 100644 index ed2acb09d5..0000000000 --- a/aeon/classification/dictionary_based/tests/test_redcomets_oversampling.py +++ /dev/null @@ -1,116 +0,0 @@ -"""Tests for REDCOMETS oversampling without imbalanced-learn.""" - -from collections import Counter - -import numpy as np -from sklearn.neighbors import NearestNeighbors -from sklearn.utils import check_random_state - -from aeon.classification.dictionary_based import REDCOMETS -from aeon.transformations.collection import Normalizer -from aeon.transformations.collection.imbalance import SMOTE, RandomOverSampler -from aeon.utils.validation._dependencies import _check_soft_dependencies - - -def _normalise_2d(X2d): - """Apply aeon collection Normalizer to a 2D panel and return 2D.""" - return Normalizer().fit_transform(X2d).squeeze(1) - - -def _redcomets_resample_aeon(X2d, y, random_state=0, n_jobs=1): - """Mirror REDCOMETS oversampling with aeon transformers.""" - X = _normalise_2d(X2d) - min_neighbours = min(Counter(y).values()) - max_neighbours = max(Counter(y).values()) - if min_neighbours == max_neighbours: - return X, y - if min_neighbours > 5: - min_neighbours = 6 - n_neighbors = min_neighbours - 2 - X_3d = X[:, np.newaxis, :] - try: - if n_neighbors < 1: - raise ValueError("not enough neighbours") - X_smote, y_smote = SMOTE( - n_neighbors=n_neighbors, - random_state=random_state, - distance="euclidean", - n_jobs=n_jobs, - ).fit_transform(X_3d, y) - except ValueError: - X_smote, y_smote = RandomOverSampler(random_state=random_state).fit_transform( - X_3d, y - ) - return np.squeeze(X_smote, 1), y_smote - - -def test_redcomets_smote_class_counts_match_imblearn(): - """Class counts and samples after aeon SMOTE match the previous imblearn path.""" - if not _check_soft_dependencies( - "imbalanced-learn", - package_import_alias={"imbalanced-learn": "imblearn"}, - severity="none", - ): - return - - from imblearn.over_sampling import SMOTE as ImbSMOTE - - rng = check_random_state(0) - X2d = rng.randn(40, 30) - y = np.array([0] * 24 + [1] * 16) - Xn = _normalise_2d(X2d) - min_neighbours = 6 - X_imb, y_imb = ImbSMOTE( - sampling_strategy="all", - k_neighbors=NearestNeighbors(n_neighbors=min_neighbours - 1), - random_state=0, - ).fit_resample(Xn, y) - X_aeon, y_aeon = _redcomets_resample_aeon(X2d, y, random_state=0) - - assert Counter(map(str, y_imb)) == Counter(map(str, y_aeon)) - assert X_imb.shape == X_aeon.shape - - def sort_xy(X, y): - y = np.asarray(y).astype(str) - idx = np.lexsort((X[:, 0], y)) - return X[idx], y[idx] - - Xi, yi = sort_xy(X_imb, y_imb) - Xa, ya = sort_xy(X_aeon, y_aeon) - assert np.array_equal(yi, ya) - assert np.allclose(Xi, Xa) - - -def test_redcomets_fits_without_imblearn_tag(): - """REDCOMETS no longer declares an imblearn soft dependency.""" - clf = REDCOMETS(variant=1, n_trees=2, random_state=0) - deps = clf.get_tag("python_dependencies", None) - if deps is None: - deps = [] - if isinstance(deps, str): - deps = [deps] - assert "imblearn" not in deps - assert "imbalanced-learn" not in deps - - -def test_redcomets_ros_fallback_on_tiny_minority(): - """When SMOTE cannot run, RandomOverSampler balances class counts.""" - rng = check_random_state(0) - # one class with a single sample forces SMOTE neighbour failure - X2 = rng.randn(10, 20) - y = np.array([0] * 8 + [1] * 1 + [2] * 1) - X_res, y_res = _redcomets_resample_aeon(X2, y, random_state=0) - counts = Counter(map(int, y_res)) - assert len(set(counts.values())) == 1 - assert counts[0] == 8 - - -def test_redcomets_end_to_end_fit_predict(): - """REDCOMETS fits and predicts on a small imbalanced panel.""" - rng = check_random_state(0) - X = rng.randn(30, 1, 24) - y = np.array([0] * 20 + [1] * 10) - clf = REDCOMETS(variant=1, n_trees=5, random_state=0) - clf.fit(X, y) - pred = clf.predict(X) - assert pred.shape == (30,) diff --git a/aeon/transformations/collection/imbalance/_esmote.py b/aeon/transformations/collection/imbalance/_esmote.py index cced6c4574..6159942c62 100644 --- a/aeon/transformations/collection/imbalance/_esmote.py +++ b/aeon/transformations/collection/imbalance/_esmote.py @@ -159,7 +159,6 @@ def _make_samples( nn_ts, distance=self.distance, step=steps[count], - return_bias=False, **self._distance_params, ) @@ -185,7 +184,6 @@ def _generate_sample_use_elastic_distance( transformation_precomputed: bool = False, transformed_x: np.ndarray | None = None, transformed_y: np.ndarray | None = None, - return_bias=False, ): """ Generate a single synthetic sample using soft distance. @@ -227,8 +225,5 @@ def _generate_sample_use_elastic_distance( empty_of_array[:, k] = curr_ts[:, k] - nn_ts[:, key] bias = step * empty_of_array - if return_bias: - return bias - new_ts = new_ts - bias return new_ts diff --git a/aeon/transformations/collection/imbalance/_smote.py b/aeon/transformations/collection/imbalance/_smote.py index 7cc862fa01..e2330f1951 100644 --- a/aeon/transformations/collection/imbalance/_smote.py +++ b/aeon/transformations/collection/imbalance/_smote.py @@ -150,7 +150,7 @@ def _transform(self, X, y=None): return X_resampled, y_resampled def _make_samples( - self, X, y_dtype, y_type, nn_data, nn_num, n_samples, step_size=1.0, y=None + self, X, y_dtype, y_type, nn_data, nn_num, n_samples, step_size=1.0 ): """Make artificial samples constructed based on nearest neighbours. @@ -180,10 +180,6 @@ def _make_samples( step_size : float, default=1.0 The step size to create samples. - y : ndarray of shape (n_samples_all,), default=None - The true target associated with `nn_data`. Used by Borderline SMOTE-2 to - weight the distances in the sample generation process. - Returns ------- X_new : ndarray @@ -201,13 +197,11 @@ def _make_samples( rows = np.floor_divide(samples_indices, nn_num.shape[1]) cols = np.mod(samples_indices, nn_num.shape[1]) - X_new = self._generate_samples(X, nn_data, nn_num, rows, cols, steps, y_type, y) + X_new = self._generate_samples(X, nn_data, nn_num, rows, cols, steps) y_new = np.full(n_samples, fill_value=y_type, dtype=y_dtype) return X_new, y_new - def _generate_samples( - self, X, nn_data, nn_num, rows, cols, steps, y_type=None, y=None - ): + def _generate_samples(self, X, nn_data, nn_num, rows, cols, steps): r"""Generate a synthetic sample. The rule for the generation is: @@ -237,12 +231,6 @@ def _generate_samples( will be used when creating new samples. steps : ndarray of shape (n_samples,), dtype=float Step sizes for new samples. - y_type : str, int or None, default=None - Class label of the current target classes for which we want to generate - samples. - y : ndarray of shape (n_samples_all,), default=None - The true target associated with `nn_data`. Used by Borderline SMOTE-2 to - weight the distances in the sample generation process. Returns ------- @@ -250,11 +238,6 @@ def _generate_samples( Synthetically generated samples. """ diffs = nn_data[nn_num[rows, cols]] - X[rows] - if y is not None: - mask_pair_samples = y[nn_num[rows, cols]] != y_type - diffs[mask_pair_samples] *= self._random_state.uniform( - low=0.0, high=0.5, size=(mask_pair_samples.sum(), 1) - ) X_new = X[rows] + steps * diffs return X_new.astype(X.dtype) diff --git a/aeon/transformations/collection/imbalance/tests/test_adasyn.py b/aeon/transformations/collection/imbalance/tests/test_adasyn.py index 0bb5c62ea6..5f61f3ee8e 100644 --- a/aeon/transformations/collection/imbalance/tests/test_adasyn.py +++ b/aeon/transformations/collection/imbalance/tests/test_adasyn.py @@ -1,61 +1,96 @@ -"""Test ADASYN oversampler ported from imblearn.""" +"""Test ADASYN oversampler.""" import numpy as np import pytest from aeon.testing.data_generation import make_example_3d_numpy from aeon.transformations.collection.imbalance import ADASYN -from aeon.utils.validation._dependencies import _check_soft_dependencies -def test_adasyn(): - """Test the ADASYN class. +def test_adasyn_balances_classes_approximately(): + """ADASYN oversamples the minority class to roughly the majority count. - This function creates a 3D numpy array, applies - ADASYN using the ADASYN class, and asserts that the - transformed data has a balanced number of samples. - ADASYN is a variant of SMOTE that generates synthetic samples, - but it focuses on generating samples near the decision boundary. - Therefore, sometimes, it may generate more or less samples than SMOTE, - which is why we only check if the number of samples is nearly balanced. + Unlike SMOTE, ADASYN allocates samples by local difficulty and rounds the + per-sample counts, so the minority class is only approximately balanced. """ - n_samples = 100 # Total number of labels - majority_num = 90 # number of majority class - minority_num = n_samples - majority_num # number of minority class - - X = np.random.rand(n_samples, 1, 10) - y = np.array([0] * majority_num + [1] * minority_num) - - transformer = ADASYN() - transformer.fit(X, y) - res_X, res_y = transformer.transform(X, y) - _, res_count = np.unique(res_y, return_counts=True) - - assert np.abs(len(res_X) - 2 * majority_num) < minority_num - assert np.abs(len(res_y) - 2 * majority_num) < minority_num - assert res_count[0] == majority_num - assert np.abs(res_count[0] - res_count[1]) < minority_num - - -@pytest.mark.skipif( - not _check_soft_dependencies( - "imbalanced-learn", - package_import_alias={"imbalanced-learn": "imblearn"}, - severity="none", - ), - reason="skip test if required soft dependency imbalanced-learn not available", -) -def test_equivalence_imbalance(): - """Test ported ADASYN code produces the same as imblearn version.""" - from imblearn.over_sampling import ADASYN as imbADASYN - - X, y = make_example_3d_numpy(n_cases=20, n_channels=1) - y = np.array([0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]) - X = X.squeeze() - s1 = imbADASYN(random_state=49) - X2, y2 = s1.fit_resample(X, y) - s2 = ADASYN(random_state=49) - X3, y3 = s2.fit_transform(X, y) - X3 = X3.squeeze() - assert np.array_equal(y2, y3) - assert np.allclose(X2, X3, atol=1e-4) + majority, minority = 90, 10 + X = np.random.RandomState(0).rand(majority + minority, 1, 10) + y = np.array([0] * majority + [1] * minority) + + res_X, res_y = ADASYN(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert counts[0] == majority # majority class untouched + assert abs(counts[0] - counts[1]) < minority # minority near-balanced + + +def test_adasyn_matches_imblearn_reference(): + """ADASYN reproduces the imbalanced-learn ADASYN output for a fixed seed. + + The expected synthetic samples were captured from + ``imblearn.over_sampling.ADASYN(n_neighbors=1, random_state=49)`` on this input; + this port reproduces them exactly, so the hardcoded array replaces the former + runtime dependency on imbalanced-learn as a parity oracle. Locks the density + ratio allocation and the synthetic-sample construction. + """ + X, _ = make_example_3d_numpy( + n_cases=8, n_channels=1, n_timepoints=4, random_state=0 + ) + y = np.array([0, 0, 0, 1, 1, 1, 1, 1]) # class 0 minority (3), class 1 majority (5) + n_original = len(y) + + res_X, res_y = ADASYN(n_neighbors=1, random_state=49).fit_transform(X, y) + + expected = np.array( + [ + [0.60192891, 1.77745026, 1.36949884, 1.83535333], + [2.45821348, 2.32420838, 2.10032461, 3.20847871], + [2.13662262, 2.50624231, 1.85469891, 3.46016654], + ] + ) + synthetic = res_X[n_original:].squeeze(axis=1) + np.testing.assert_allclose(synthetic, expected, atol=1e-6) + assert np.all(res_y[n_original:] == 0) + + +def test_adasyn_skips_already_balanced_class(): + """A non-majority class already at the majority count gets no synthetic samples.""" + y = np.array([0] * 5 + [1] * 5 + [2] * 3) + X = np.random.RandomState(1).rand(len(y), 1, 4) + + _, res_y = ADASYN(n_neighbors=1, random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert counts[0] == 5 and counts[1] == 5 # tied majorities untouched + assert counts[2] > 3 # smaller class was oversampled + + +def test_adasyn_raises_when_no_majority_neighbours(): + """ADASYN raises if no minority neighbour is from the majority class. + + When the minority class is tightly clustered and well separated from the + majority, every neighbour of a minority point is itself a minority point, so the + density ratio is all zero and ADASYN cannot allocate samples. + """ + X_min = 0.01 * np.random.RandomState(2).rand(4, 1, 4) + X_maj = 100 + np.random.RandomState(3).rand(8, 1, 4) + X = np.vstack([X_min, X_maj]) + y = np.array([0] * 4 + [1] * 8) + + with pytest.raises(RuntimeError, match="majority"): + ADASYN(n_neighbors=3, random_state=0).fit_transform(X, y) + + +def test_adasyn_raises_when_rounding_yields_no_samples(): + """ADASYN raises if the density-ratio rounding allocates zero samples. + + With a sampling target of a single sample spread across many minority points, + every per-point allocation rounds down to zero, so no synthetic samples can be + produced and ADASYN reports this rather than returning an unchanged set. + """ + rs = np.random.RandomState(0) + X = np.vstack([rs.rand(8, 1, 6), rs.rand(9, 1, 6) + 0.3]) # target = 9 - 8 = 1 + y = np.array([0] * 8 + [1] * 9) + + with pytest.raises(ValueError, match="No samples will be generated"): + ADASYN(n_neighbors=5, random_state=0).fit_transform(X, y) diff --git a/aeon/transformations/collection/imbalance/tests/test_esmote.py b/aeon/transformations/collection/imbalance/tests/test_esmote.py index 3bd4b52115..a56f28b4ee 100644 --- a/aeon/transformations/collection/imbalance/tests/test_esmote.py +++ b/aeon/transformations/collection/imbalance/tests/test_esmote.py @@ -5,26 +5,40 @@ from aeon.transformations.collection.imbalance import ESMOTE -def test_smote(): - """Test the ESMOTE class. +def test_esmote_balances_classes(): + """ESMOTE oversamples the minority class up to the majority count.""" + majority, minority = 20, 6 + X = np.random.RandomState(0).rand(majority + minority, 1, 10) + y = np.array([0] * majority + [1] * minority) - This function creates a 3D numpy array, applies - ESMOTE using the ESMOTE class, and asserts that the - transformed data has a balanced number of samples. + res_X, res_y = ESMOTE(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert res_y.shape == (2 * majority,) + assert set(counts) == {majority} + + +def test_esmote_deterministic(): + """A fixed random_state gives identical synthetic series across fits. + + ESMOTE draws neighbours, step sizes and alignment tie-breaks from its random + state; seeding it must make the whole elastic generation reproducible. """ - n_samples = 100 # Total number of labels - majority_num = 90 # number of majority class - minority_num = n_samples - majority_num # number of minority class - - X = np.random.rand(n_samples, 1, 10) - y = np.array([0] * majority_num + [1] * minority_num) - - transformer = ESMOTE() - transformer.fit(X, y) - res_X, res_y = transformer.transform(X, y) - _, res_count = np.unique(res_y, return_counts=True) - - assert len(res_X) == 2 * majority_num - assert len(res_y) == 2 * majority_num - assert res_count[0] == majority_num - assert res_count[1] == majority_num + X = np.random.RandomState(1).rand(26, 1, 10) + y = np.array([0] * 20 + [1] * 6) + + res1 = ESMOTE(random_state=7).fit_transform(X, y)[0] + res2 = ESMOTE(random_state=7).fit_transform(X, y)[0] + + np.testing.assert_array_equal(res1, res2) + + +def test_esmote_skips_already_balanced_class(): + """A non-majority class already at the majority count gets no synthetic samples.""" + y = np.array([0] * 8 + [1] * 8 + [2] * 6) + X = np.random.RandomState(2).rand(len(y), 1, 10) + + _, res_y = ESMOTE(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert set(counts) == {8} # smaller class raised to the tied-majority count diff --git a/aeon/transformations/collection/imbalance/tests/test_ohit.py b/aeon/transformations/collection/imbalance/tests/test_ohit.py index 6b21f018e6..b06860016e 100644 --- a/aeon/transformations/collection/imbalance/tests/test_ohit.py +++ b/aeon/transformations/collection/imbalance/tests/test_ohit.py @@ -5,29 +5,17 @@ from aeon.transformations.collection.imbalance import OHIT -def test_ohit(): - """Test the OHIT class. +def test_ohit_balances_classes(): + """OHIT oversamples the minority class up to the majority count.""" + majority, minority = 90, 10 + X = np.random.RandomState(0).rand(majority + minority, 1, 10) + y = np.array([0] * majority + [1] * minority) - This function creates a 3D numpy array, applies - OHIT using the OHIT class, and asserts that the - transformed data has a balanced number of samples. - """ - n_samples = 100 # Total number of labels - majority_num = 90 # number of majority class - minority_num = n_samples - majority_num # number of minority class - - X = np.random.rand(n_samples, 1, 10) - y = np.array([0] * majority_num + [1] * minority_num) - - transformer = OHIT() - transformer.fit(X, y) - res_X, res_y = transformer.transform(X, y) - _, res_count = np.unique(res_y, return_counts=True) + res_X, res_y = OHIT(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) - assert len(res_X) == 2 * majority_num - assert len(res_y) == 2 * majority_num - assert res_count[0] == majority_num - assert res_count[1] == majority_num + assert res_X.shape == (2 * majority, 1, 10) + assert set(counts) == {majority} def test_ohit_random_state_reproducible(): @@ -51,3 +39,50 @@ def test_ohit_does_not_mutate_params(): assert transformer.get_params()["k"] is None assert transformer.get_params()["kapa"] is None + + +def test_ohit_single_sample_minority_class(): + """A minority class with a single sample is oversampled by replication. + + DRSNN clustering needs more than one point, so OHIT falls back to tiling the + lone minority series up to the required count. + """ + y = np.array([0] * 1 + [1] * 10 + [2] * 6) + X = np.random.RandomState(5).rand(len(y), 1, 8) + lone = X[0] + + res_X, res_y = OHIT(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert set(counts) == {10} + # every class-0 sample is a copy of the single original minority series + class0 = res_X[res_y == 0] + assert np.all(class0 == lone) + + +def test_ohit_no_cluster_fallback(): + """When DRSNN finds no core points, all minority samples form one cluster. + + Forcing the density-ratio threshold above any achievable value leaves no core + points, so OHIT must fall back to treating the whole minority class as a single + cluster and still return a balanced set. + """ + majority, minority = 20, 6 + X = np.random.RandomState(4).rand(majority + minority, 1, 8) + y = np.array([0] * majority + [1] * minority) + + _, res_y = OHIT(drT=1e9, random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert set(counts) == {majority} + + +def test_ohit_skips_already_balanced_class(): + """A non-majority class already at the majority count gets no synthetic samples.""" + y = np.array([0] * 20 + [1] * 20 + [2] * 10) + X = np.random.RandomState(6).rand(len(y), 1, 8) + + _, res_y = OHIT(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert set(counts) == {20} diff --git a/aeon/transformations/collection/imbalance/tests/test_random_over_sampler.py b/aeon/transformations/collection/imbalance/tests/test_random_over_sampler.py index f3216bca55..816a97e1c8 100644 --- a/aeon/transformations/collection/imbalance/tests/test_random_over_sampler.py +++ b/aeon/transformations/collection/imbalance/tests/test_random_over_sampler.py @@ -1,11 +1,9 @@ """Tests for RandomOverSampler.""" import numpy as np -import pytest from sklearn.utils import check_random_state from aeon.transformations.collection.imbalance import RandomOverSampler -from aeon.utils.validation._dependencies import _check_soft_dependencies def test_random_over_sampler_balances_classes(): @@ -45,34 +43,13 @@ def test_random_over_sampler_multivariate(): assert np.array_equal(y_res[:9], y) -@pytest.mark.skipif( - not _check_soft_dependencies( - "imbalanced-learn", - package_import_alias={"imbalanced-learn": "imblearn"}, - severity="none", - ), - reason="skip test if required soft dependency imbalanced-learn not available", -) -def test_random_over_sampler_matches_imblearn(): - """Match imblearn RandomOverSampler with sampling_strategy='all'.""" - from imblearn.over_sampling import RandomOverSampler as ImbROS - +def test_random_over_sampler_multiclass_balances_to_majority(): + """Every minority class is raised to the majority count in a multi-class panel.""" rng = check_random_state(0) - X2 = rng.randn(20, 15) + X = rng.randn(20, 1, 15) y = np.array([0] * 12 + [1] * 5 + [2] * 3) - X_imb, y_imb = ImbROS(sampling_strategy="all", random_state=0).fit_resample(X2, y) - X_aeon, y_aeon = RandomOverSampler(random_state=0).fit_transform( - X2[:, np.newaxis, :], y - ) - X_aeon = X_aeon.squeeze(1) - def sort_xy(X, y): - y = np.asarray(y) - idx = np.lexsort((X[:, 0], y.astype(str))) - return X[idx], y[idx] + _, y_res = RandomOverSampler(random_state=0).fit_transform(X, y) + _, counts = np.unique(y_res, return_counts=True) - Xi, yi = sort_xy(X_imb, y_imb) - Xa, ya = sort_xy(X_aeon, y_aeon) - assert Xi.shape == Xa.shape - assert np.array_equal(yi.astype(str), ya.astype(str)) - assert np.allclose(Xi, Xa) + assert counts.tolist() == [12, 12, 12] diff --git a/aeon/transformations/collection/imbalance/tests/test_smote.py b/aeon/transformations/collection/imbalance/tests/test_smote.py index 70189633d0..ce94e801b3 100644 --- a/aeon/transformations/collection/imbalance/tests/test_smote.py +++ b/aeon/transformations/collection/imbalance/tests/test_smote.py @@ -1,57 +1,70 @@ """Test function for SMOTE.""" import numpy as np -import pytest from aeon.testing.data_generation import make_example_3d_numpy from aeon.transformations.collection.imbalance import SMOTE -from aeon.utils.validation._dependencies import _check_soft_dependencies -def test_smote(): - """Test the SMOTE class. +def test_smote_balances_classes(): + """SMOTE oversamples the minority class up to the majority count.""" + majority, minority = 90, 10 + X = np.random.RandomState(0).rand(majority + minority, 1, 10) + y = np.array([0] * majority + [1] * minority) - This function creates a 3D numpy array, applies - SMOTE using the SMOTE class, and asserts that the - transformed data has a balanced number of samples. + res_X, res_y = SMOTE(random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + assert res_X.shape == (2 * majority, 1, 10) + assert res_y.shape == (2 * majority,) + # every class is raised to the majority count + assert set(counts) == {majority} + + +def test_smote_matches_imblearn_reference(): + """SMOTE reproduces the imbalanced-learn SMOTE output for a fixed seed. + + The expected synthetic samples were captured from + ``imblearn.over_sampling.SMOTE(k_neighbors=1, random_state=49)`` on this input; + this port reproduces them exactly, so the hardcoded array replaces the former + runtime dependency on imbalanced-learn as a parity oracle. Locks the SMOTE + interpolation ``s_i + u * (s_nn - s_i)`` and the labelling of new samples. + """ + X, _ = make_example_3d_numpy( + n_cases=8, n_channels=1, n_timepoints=4, random_state=0 + ) + y = np.array([0, 0, 0, 1, 1, 1, 1, 1]) # class 0 minority (3), class 1 majority (5) + n_original = len(y) + n_synthetic = 5 - 3 # majority count - minority count + + res_X, res_y = SMOTE(n_neighbors=1, random_state=49).fit_transform(X, y) + + expected = np.array( + [ + [3.39643756, 1.79313415, 2.81692439, 2.47419297], + [3.71802842, 1.61110021, 3.06255009, 2.22250514], + ] + ) + synthetic = res_X[n_original:].squeeze(axis=1) + assert synthetic.shape == (n_synthetic, 4) + np.testing.assert_allclose(synthetic, expected, atol=1e-6) + # synthetic samples all belong to the oversampled minority class + assert np.all(res_y[n_original:] == 0) + + +def test_smote_skips_already_balanced_class(): + """A non-majority class already at the majority count gets no synthetic samples. + + With two classes tied for the largest count, the tied non-majority class has a + sampling target of zero and must be skipped, while the genuinely smaller class is + still oversampled to the majority count. """ - n_samples = 100 # Total number of labels - majority_num = 90 # number of majority class - minority_num = n_samples - majority_num # number of minority class - - X = np.random.rand(n_samples, 1, 10) - y = np.array([0] * majority_num + [1] * minority_num) - - transformer = SMOTE() - transformer.fit(X, y) - res_X, res_y = transformer.transform(X, y) - _, res_count = np.unique(res_y, return_counts=True) - - assert len(res_X) == 2 * majority_num - assert len(res_y) == 2 * majority_num - assert res_count[0] == majority_num - assert res_count[1] == majority_num - - -@pytest.mark.skipif( - not _check_soft_dependencies( - "imbalanced-learn", - package_import_alias={"imbalanced-learn": "imblearn"}, - severity="none", - ), - reason="skip test if required soft dependency imbalanced-learn not available", -) -def test_equivalence_imbalance(): - """Test ported SMOTE code produces the same as imblearn version.""" - from imblearn.over_sampling import SMOTE as imbSMOTE - - X, y = make_example_3d_numpy(n_cases=20, n_channels=1) - y = np.array([0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]) - X = X.squeeze() - s1 = imbSMOTE(random_state=49) - X2, y2 = s1.fit_resample(X, y) - s2 = SMOTE(random_state=49) - X3, y3 = s2.fit_transform(X, y) - X3 = X3.squeeze() - assert np.array_equal(y2, y3) - assert np.allclose(X2, X3, atol=1e-4) + counts_in = {0: 5, 1: 5, 2: 3} + y = np.array([0] * 5 + [1] * 5 + [2] * 3) + X = np.random.RandomState(1).rand(len(y), 1, 4) + + _, res_y = SMOTE(n_neighbors=1, random_state=0).fit_transform(X, y) + _, counts = np.unique(res_y, return_counts=True) + + majority = max(counts_in.values()) + assert set(counts) == {majority} diff --git a/pyproject.toml b/pyproject.toml index ec68e12f5f..19ff00b1a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,7 +58,6 @@ dependencies = [ [project.optional-dependencies] all_extras = [ "huggingface-hub>=0.20.0", - "imbalanced-learn", "matplotlib>=3.3.2", "pycatch22>=0.4.5", "pyod>=1.1.3",