-
Notifications
You must be signed in to change notification settings - Fork 89
/
_pyod.py
126 lines (96 loc) · 3.91 KB
/
_pyod.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
"""Implements outlier detection from pyOD."""
import numpy as np
from sklearn.base import clone
from aeon.annotation.base._base import BaseSeriesAnnotator
from aeon.utils.validation._dependencies import _check_soft_dependencies
__author__ = ["mloning", "satya-pattnaik", "fkiraly"]
import pandas as pd
class PyODAnnotator(BaseSeriesAnnotator):
"""Transformer that applies outlier detector from pyOD.
Parameters
----------
estimator : PyOD estimator
See ``https://pyod.readthedocs.io/en/latest/`` documentation for a detailed
description of all options.
fmt : str {"dense", "sparse"}, optional (default="dense")
annotation output format:
* If "sparse", a sub-series of labels for only the outliers in X is returned,
* If "dense", a series of labels for all values in X is returned.
labels : str {"indicator", "score"}, optional (default="indicator")
annotation output labels:
* If "indicator", returned values are boolean, indicating whether a value is an
outlier,
* If "score", returned values are floats, giving the outlier score.
"""
_tags = {"python_dependencies": "pyod"}
def __init__(self, estimator, fmt="dense", labels="indicator"):
self.estimator = estimator # pyod estimator
super(PyODAnnotator, self).__init__(fmt=fmt, labels=labels)
def _fit(self, X, Y=None):
"""Fit to training data.
core logic
Parameters
----------
X : pd.DataFrame
training data to fit model to, time series
Y : pd.Series, optional
ground truth annotations for training if annotator is supervised
Returns
-------
self : returns a reference to self
Notes
-----
Create fitted model that sets attributes ending in "_".
"""
X_np = X.to_numpy()
if len(X_np.shape) == 1:
X_np = X_np.reshape(-1, 1)
self.estimator_ = clone(self.estimator)
self.estimator_.fit(X_np)
return self
def _predict(self, X):
"""Create annotations on test/deployment data.
Parameters
----------
X : pd.DataFrame - data to annotate, time series
Returns
-------
Y : pd.Series - annotations for sequence X
exact format depends on annotation type
"""
fmt = self.fmt
labels = self.labels
X_np = X.to_numpy()
if len(X_np.shape) == 1:
X_np = X_np.reshape(-1, 1)
Y_np = self.estimator_.predict(X_np)
if labels == "score":
Y_val_np = self.estimator_.decision_function(X_np)
elif labels == "indicator":
Y_val_np = Y_np
if fmt == "dense":
Y = pd.Series(Y_val_np, index=X.index)
elif fmt == "sparse":
Y_loc = np.where(Y_np)
Y = pd.Series(Y_val_np[Y_loc], index=X.index[Y_loc])
return Y
@classmethod
def get_test_params(cls, parameter_set="default"):
"""Return testing parameter settings for the estimator.
Parameters
----------
parameter_set : str, default="default"
Name of the set of test parameters to return, for use in tests. If no
special parameters are defined for a value, will return `"default"` set.
There are currently no reserved values for annotators.
Returns
-------
params : dict or list of dict, default = {}
Parameters to create testing instances of the class
Each dict are parameters to construct an "interesting" test instance, i.e.,
`MyClass(**params)` or `MyClass(**params[i])` creates a valid test instance.
`create_test_instance` uses the first (or only) dictionary in `params`
"""
_check_soft_dependencies("pyod", severity="error")
from pyod.models.knn import KNN
return {"estimator": KNN()}