-
Notifications
You must be signed in to change notification settings - Fork 1.3k
/
Copy pathbase.py
309 lines (241 loc) · 9.49 KB
/
base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
"""Base class for sampling"""
# Authors: Guillaume Lemaitre <[email protected]>
# Christos Aridas
# License: MIT
from abc import ABCMeta, abstractmethod
import numpy as np
from sklearn.base import BaseEstimator
from .dask._support import is_dask_collection
from .utils import check_sampling_strategy, check_target_type
from .utils._validation import (
ArraysTransformer,
_deprecate_positional_args,
get_classes_counts,
)
from .utils.wrapper import (
check_classification_targets,
label_binarize,
)
class SamplerMixin(BaseEstimator, metaclass=ABCMeta):
"""Mixin class for samplers with abstract method.
Warning: This class should not be used directly. Use the derive classes
instead.
"""
_estimator_type = "sampler"
def fit(self, X, y):
"""Check inputs and statistics of the sampler.
You should use ``fit_resample`` in all cases.
Parameters
----------
X : {array-like, dataframe, sparse matrix} of shape \
(n_samples, n_features)
Data array.
y : array-like of shape (n_samples,)
Target array.
Returns
-------
self : object
Return the instance itself.
"""
arrays_transformer = ArraysTransformer(X, y)
dask_collection = any([is_dask_collection(arr) for arr in (X, y)])
if dask_collection:
X, y = arrays_transformer.to_dask_array(X, y)
if (not dask_collection or
(dask_collection and self.validate_if_dask_collection)):
X, y, _ = self._check_X_y(X, y)
self._classes_counts = get_classes_counts(y)
self.sampling_strategy_ = check_sampling_strategy(
self.sampling_strategy, self._classes_counts, self._sampling_type
)
return self
def fit_resample(self, X, y):
"""Resample the dataset.
Parameters
----------
X : {array-like, dataframe, sparse matrix} of shape \
(n_samples, n_features)
Matrix containing the data which have to be sampled.
y : array-like of shape (n_samples,)
Corresponding label for each sample in X.
Returns
-------
X_resampled : {array-like, dataframe, sparse matrix} of shape \
(n_samples_new, n_features)
The array containing the resampled data.
y_resampled : array-like of shape (n_samples_new,)
The corresponding label of `X_resampled`.
"""
arrays_transformer = ArraysTransformer(X, y)
dask_collection = any([is_dask_collection(arr) for arr in (X, y)])
if dask_collection:
X, y = arrays_transformer.to_dask_array(X, y)
if (not dask_collection or
(dask_collection and self.validate_if_dask_collection)):
check_classification_targets(y)
X, y, binarize_y = self._check_X_y(X, y)
else:
binarize_y = False
self._classes_counts = get_classes_counts(y)
self.sampling_strategy_ = check_sampling_strategy(
self.sampling_strategy, self._classes_counts, self._sampling_type
)
output = self._fit_resample(X, y)
if binarize_y:
y_ = label_binarize(
output[1], classes=list(self._classes_counts.keys())
)
else:
y_ = output[1]
X_, y_ = arrays_transformer.transform(output[0], y_)
return (X_, y_) if len(output) == 2 else (X_, y_, output[2])
# define an alias for back-compatibility
fit_sample = fit_resample
@abstractmethod
def _fit_resample(self, X, y):
"""Base method defined in each sampler to defined the sampling
strategy.
Parameters
----------
X : {array-like, sparse matrix} of shape (n_samples, n_features)
Matrix containing the data which have to be sampled.
y : array-like of shape (n_samples,)
Corresponding label for each sample in X.
Returns
-------
X_resampled : {ndarray, sparse matrix} of shape \
(n_samples_new, n_features)
The array containing the resampled data.
y_resampled : ndarray of shape (n_samples_new,)
The corresponding label of `X_resampled`.
"""
pass
class BaseSampler(SamplerMixin):
"""Base class for sampling algorithms.
Warning: This class should not be used directly. Use the derive classes
instead.
"""
def __init__(
self,
sampling_strategy="auto",
validate_if_dask_collection=False,
):
self.sampling_strategy = sampling_strategy
self.validate_if_dask_collection = validate_if_dask_collection
def _check_X_y(self, X, y, accept_sparse=None):
if accept_sparse is None:
accept_sparse = ["csr", "csc"]
y, binarize_y = check_target_type(y, indicate_one_vs_all=True)
X, y = self._validate_data(
X, y, reset=True, accept_sparse=accept_sparse
)
return X, y, binarize_y
def _identity(X, y):
return X, y
class FunctionSampler(BaseSampler):
"""Construct a sampler from calling an arbitrary callable.
Read more in the :ref:`User Guide <function_sampler>`.
Parameters
----------
func : callable, default=None
The callable to use for the transformation. This will be passed the
same arguments as transform, with args and kwargs forwarded. If func is
None, then func will be the identity function.
accept_sparse : bool, default=True
Whether sparse input are supported. By default, sparse inputs are
supported.
kw_args : dict, default=None
The keyword argument expected by ``func``.
validate : bool, default=True
Whether or not to bypass the validation of ``X`` and ``y``. Turning-off
validation allows to use the ``FunctionSampler`` with any type of
data.
See Also
--------
sklearn.preprocessing.FunctionTransfomer : Stateless transformer.
Notes
-----
See
:ref:`sphx_glr_auto_examples_plot_outlier_rejections.py`
Examples
--------
>>> import numpy as np
>>> from sklearn.datasets import make_classification
>>> from imblearn import FunctionSampler
>>> X, y = make_classification(n_classes=2, class_sep=2,
... weights=[0.1, 0.9], n_informative=3, n_redundant=1, flip_y=0,
... n_features=20, n_clusters_per_class=1, n_samples=1000, random_state=10)
We can create to select only the first ten samples for instance.
>>> def func(X, y):
... return X[:10], y[:10]
>>> sampler = FunctionSampler(func=func)
>>> X_res, y_res = sampler.fit_resample(X, y)
>>> np.all(X_res == X[:10])
True
>>> np.all(y_res == y[:10])
True
We can also create a specific function which take some arguments.
>>> from collections import Counter
>>> from imblearn.under_sampling import RandomUnderSampler
>>> def func(X, y, sampling_strategy, random_state):
... return RandomUnderSampler(
... sampling_strategy=sampling_strategy,
... random_state=random_state).fit_resample(X, y)
>>> sampler = FunctionSampler(func=func,
... kw_args={'sampling_strategy': 'auto',
... 'random_state': 0})
>>> X_res, y_res = sampler.fit_resample(X, y)
>>> print('Resampled dataset shape {}'.format(
... sorted(Counter(y_res).items())))
Resampled dataset shape [(0, 100), (1, 100)]
"""
_sampling_type = "bypass"
@_deprecate_positional_args
def __init__(self, *, func=None, accept_sparse=True, kw_args=None,
validate=True):
super().__init__()
self.func = func
self.accept_sparse = accept_sparse
self.kw_args = kw_args
self.validate = validate
def fit_resample(self, X, y):
"""Resample the dataset.
Parameters
----------
X : {array-like, sparse matrix} of shape (n_samples, n_features)
Matrix containing the data which have to be sampled.
y : array-like of shape (n_samples,)
Corresponding label for each sample in X.
Returns
-------
X_resampled : {array-like, sparse matrix} of shape \
(n_samples_new, n_features)
The array containing the resampled data.
y_resampled : array-like of shape (n_samples_new,)
The corresponding label of `X_resampled`.
"""
arrays_transformer = ArraysTransformer(X, y)
if self.validate:
check_classification_targets(y)
X, y, binarize_y = self._check_X_y(
X, y, accept_sparse=self.accept_sparse
)
self._classes_counts = get_classes_counts(y)
self.sampling_strategy_ = check_sampling_strategy(
self.sampling_strategy, self._classes_counts, self._sampling_type
)
output = self._fit_resample(X, y)
if self.validate:
if binarize_y:
y_ = label_binarize(
output[1], classes=list(self._classes_counts.keys())
)
else:
y_ = output[1]
X_, y_ = arrays_transformer.transform(output[0], y_)
return (X_, y_) if len(output) == 2 else (X_, y_, output[2])
return output
def _fit_resample(self, X, y):
func = _identity if self.func is None else self.func
output = func(X, y, **(self.kw_args if self.kw_args else {}))
return output