-
Notifications
You must be signed in to change notification settings - Fork 678
/
Copy pathMINIROCKET.py
178 lines (148 loc) · 8.72 KB
/
MINIROCKET.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
# AUTOGENERATED! DO NOT EDIT! File to edit: ../../nbs/055_models.MINIROCKET.ipynb.
# %% auto 0
__all__ = ['MiniRocketClassifier', 'load_minirocket', 'MiniRocketRegressor', 'MiniRocketVotingClassifier', 'get_minirocket_preds',
'MiniRocketVotingRegressor']
# %% ../../nbs/055_models.MINIROCKET.ipynb 3
import sklearn
from sklearn.ensemble import VotingClassifier, VotingRegressor
from sklearn.linear_model import RidgeClassifierCV, RidgeCV
from sklearn.metrics import make_scorer
from sklearn.preprocessing import StandardScaler
from ..data.external import *
from ..imports import *
from .layers import *
from ..utils import *
# %% ../../nbs/055_models.MINIROCKET.ipynb 4
class MiniRocketClassifier(sklearn.pipeline.Pipeline):
"""Time series classification using MINIROCKET features and a linear classifier"""
def __init__(self, num_features=10_000, max_dilations_per_kernel=32, random_state=None,
alphas=np.logspace(-3, 3, 7), normalize_features=True, memory=None, verbose=False, scoring=None, class_weight=None, **kwargs):
""" MiniRocketClassifier is recommended for up to 10k time series.
For a larger dataset, you can use MINIROCKET (in Pytorch).
scoring = None --> defaults to accuracy.
"""
try:
import sktime
from sktime.transformations.panel.rocket._minirocket_multivariate import MiniRocketMultivariate
except ImportError:
raise ImportError("You need to install sktime to be able to use MiniRocketClassifier")
self.steps = [('minirocketmultivariate', MiniRocketMultivariate(num_kernels=num_features,
max_dilations_per_kernel=max_dilations_per_kernel,
random_state=random_state))]
if normalize_features:
self.steps += [('scalar', StandardScaler(with_mean=False))]
self.steps += [('ridgeclassifiercv', RidgeClassifierCV(alphas=alphas,
scoring=scoring,
class_weight=class_weight,
**kwargs))]
store_attr()
self._validate_steps()
def __repr__(self):
return f'Pipeline(steps={self.steps.copy()})'
def save(self, fname=None, path='./models'):
fname = ifnone(fname, 'MiniRocketClassifier')
path = Path(path)
filename = path/fname
filename.parent.mkdir(parents=True, exist_ok=True)
with open(f'{filename}.pkl', 'wb') as output:
pickle.dump(self, output, pickle.HIGHEST_PROTOCOL)
# %% ../../nbs/055_models.MINIROCKET.ipynb 5
def load_minirocket(fname, path='./models'):
path = Path(path)
filename = path/fname
with open(f'{filename}.pkl', 'rb') as input:
output = pickle.load(input)
return output
# %% ../../nbs/055_models.MINIROCKET.ipynb 6
class MiniRocketRegressor(sklearn.pipeline.Pipeline):
"""Time series regression using MINIROCKET features and a linear regressor"""
def __init__(self, num_features=10000, max_dilations_per_kernel=32, random_state=None,
alphas=np.logspace(-3, 3, 7), *, normalize_features=True, memory=None, verbose=False, scoring=None, **kwargs):
""" MiniRocketRegressor is recommended for up to 10k time series.
For a larger dataset, you can use MINIROCKET (in Pytorch).
scoring = None --> defaults to r2.
"""
try:
import sktime
from sktime.transformations.panel.rocket._minirocket_multivariate import MiniRocketMultivariate
except ImportError:
raise ImportError("You need to install sktime to be able to use MiniRocketRegressor")
self.steps = [('minirocketmultivariate', MiniRocketMultivariate(num_kernels=num_features,
max_dilations_per_kernel=max_dilations_per_kernel,
random_state=random_state))]
if normalize_features:
self.steps += [('scalar', StandardScaler(with_mean=False))]
self.steps += [('ridgecv', RidgeCV(alphas=alphas, scoring=scoring, **kwargs))]
store_attr()
self._validate_steps()
def __repr__(self):
return f'Pipeline(steps={self.steps.copy()})'
def save(self, fname=None, path='./models'):
fname = ifnone(fname, 'MiniRocketRegressor')
path = Path(path)
filename = path/fname
filename.parent.mkdir(parents=True, exist_ok=True)
with open(f'{filename}.pkl', 'wb') as output:
pickle.dump(self, output, pickle.HIGHEST_PROTOCOL)
# %% ../../nbs/055_models.MINIROCKET.ipynb 7
def load_minirocket(fname, path='./models'):
path = Path(path)
filename = path/fname
with open(f'{filename}.pkl', 'rb') as input:
output = pickle.load(input)
return output
# %% ../../nbs/055_models.MINIROCKET.ipynb 8
class MiniRocketVotingClassifier(VotingClassifier):
"""Time series classification ensemble using MINIROCKET features, a linear classifier and majority voting"""
def __init__(self, n_estimators=5, weights=None, n_jobs=-1, num_features=10_000, max_dilations_per_kernel=32, random_state=None,
alphas=np.logspace(-3, 3, 7), normalize_features=True, memory=None, verbose=False, scoring=None, class_weight=None, **kwargs):
store_attr()
try:
import sktime
except ImportError:
raise ImportError("You need to install sktime to be able to use MiniRocketVotingClassifier")
estimators = [(f'est_{i}', MiniRocketClassifier(num_features=num_features, max_dilations_per_kernel=max_dilations_per_kernel,
random_state=random_state, alphas=alphas, normalize_features=normalize_features, memory=memory,
verbose=verbose, scoring=scoring, class_weight=class_weight, **kwargs))
for i in range(n_estimators)]
super().__init__(estimators, voting='hard', weights=weights, n_jobs=n_jobs, verbose=verbose)
def __repr__(self):
return f'MiniRocketVotingClassifier(n_estimators={self.n_estimators}, \nsteps={self.estimators[0][1].steps})'
def save(self, fname=None, path='./models'):
fname = ifnone(fname, 'MiniRocketVotingClassifier')
path = Path(path)
filename = path/fname
filename.parent.mkdir(parents=True, exist_ok=True)
with open(f'{filename}.pkl', 'wb') as output:
pickle.dump(self, output, pickle.HIGHEST_PROTOCOL)
# %% ../../nbs/055_models.MINIROCKET.ipynb 9
def get_minirocket_preds(X, fname, path='./models', model=None):
if X.ndim == 1: X = X[np.newaxis][np.newaxis]
elif X.ndim == 2: X = X[np.newaxis]
if model is None:
model = load_minirocket(fname=fname, path=path)
return model.predict(X)
# %% ../../nbs/055_models.MINIROCKET.ipynb 10
class MiniRocketVotingRegressor(VotingRegressor):
"""Time series regression ensemble using MINIROCKET features, a linear regressor and a voting regressor"""
def __init__(self, n_estimators=5, weights=None, n_jobs=-1, num_features=10_000, max_dilations_per_kernel=32, random_state=None,
alphas=np.logspace(-3, 3, 7), normalize_features=True, memory=None, verbose=False, scoring=None, **kwargs):
store_attr()
try:
import sktime
except ImportError:
raise ImportError("You need to install sktime to be able to use MiniRocketVotingRegressor")
estimators = [(f'est_{i}', MiniRocketRegressor(num_features=num_features, max_dilations_per_kernel=max_dilations_per_kernel,
random_state=random_state, alphas=alphas, normalize_features=normalize_features, memory=memory,
verbose=verbose, scoring=scoring, **kwargs))
for i in range(n_estimators)]
super().__init__(estimators, weights=weights, n_jobs=n_jobs, verbose=verbose)
def __repr__(self):
return f'MiniRocketVotingRegressor(n_estimators={self.n_estimators}, \nsteps={self.estimators[0][1].steps})'
def save(self, fname=None, path='./models'):
fname = ifnone(fname, 'MiniRocketVotingRegressor')
path = Path(path)
filename = path/fname
filename.parent.mkdir(parents=True, exist_ok=True)
with open(f'{filename}.pkl', 'wb') as output:
pickle.dump(self, output, pickle.HIGHEST_PROTOCOL)