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
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,11 @@
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):
Expand Down Expand Up @@ -44,41 +42,27 @@ def _redcomets_resample_aeon(X2d, y, random_state=0, n_jobs=1):
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
def test_redcomets_smote_oversampling_balances_classes():
"""The REDCOMETS SMOTE path balances classes and preserves the original cases.

Mirrors the REDCOMETS wiring: normalise, cap the neighbour count at six, use
``min_class_count - 2`` neighbours, then SMOTE. SMOTE's numerical parity with
imbalanced-learn is locked separately in the SMOTE tests; here the contract is
that the oversampling balances the classes to the majority count and leaves the
original (normalised) cases unchanged at the front of the output.
"""
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)
n_original = len(y)

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]
X_aeon, y_aeon = _redcomets_resample_aeon(X2d, y, random_state=0)

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)
assert X_aeon.shape == (2 * 24, 30)
assert Counter(map(int, y_aeon)) == {0: 24, 1: 24}
# original normalised cases are preserved, unchanged, before the synthetic ones
assert np.allclose(X_aeon[:n_original], _normalise_2d(X2d))
assert np.array_equal(y_aeon[:n_original], y)


def test_redcomets_fits_without_imblearn_tag():
Expand Down
5 changes: 0 additions & 5 deletions aeon/transformations/collection/imbalance/_esmote.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,6 @@ def _make_samples(
nn_ts,
distance=self.distance,
step=steps[count],
return_bias=False,
**self._distance_params,
)

Expand All @@ -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.
Expand Down Expand Up @@ -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
23 changes: 3 additions & 20 deletions aeon/transformations/collection/imbalance/_smote.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -237,24 +231,13 @@ 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
-------
X_new : {ndarray, sparse matrix} of shape (n_samples, n_features)
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)

Expand Down
137 changes: 86 additions & 51 deletions aeon/transformations/collection/imbalance/tests/test_adasyn.py
Original file line number Diff line number Diff line change
@@ -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)
56 changes: 35 additions & 21 deletions aeon/transformations/collection/imbalance/tests/test_esmote.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading
Loading