-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathexample3a_anomaly_detection_3d.py
116 lines (82 loc) · 4.68 KB
/
example3a_anomaly_detection_3d.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
## -------------------------------------------------------------------------------------------------
## -- Paper : MLPro 2.0 - Online machine learning in Python
## -- Journal : ScienceDirect, Machine Learning with Applications (MLWA)
## -- Authors : Detlef Arend, Laxmikant Shrikant Baheti, Steve Yuwono,
## -- Syamraj Purushamparambil Satheesh Kumar, Andreas Schwung
## -- Module : example3a_anomaly_detection_3d.py
## -------------------------------------------------------------------------------------------------
"""
Ver. 1.0.0 (2024-12-12)
This module demonstrates the use of anomaly detector based on local outlier factor algorithm with MLPro.
To this regard, a stream of a stream provider is combined with a stream workflow to a stream scenario.
The workflow consists of a standard task 'Aanomaly Detector'.
You will learn:
1) How to set up a stream workflow based on stream tasks.
2) How to set up a stream scenario based on a stream and a processing stream workflow.
3) How to add a task anomalydetector.
4) How to reuse an anomaly detector algorithm from scikitlearn (https://scikit-learn.org/), specifically
Local Outlier Factor
"""
from mlpro.bf.streams.streams import StreamMLProPOutliers
from mlpro.bf.various import Log
from mlpro.bf.mt import Range
from mlpro.bf.ops import Mode
from mlpro.bf.plot import PlotSettings
from mlpro.oa.streams import OAStreamScenario, OAStreamWorkflow
from mlpro_int_sklearn.wrappers.anomalydetectors.lof import WrSklearnLOF2MLPro
## -------------------------------------------------------------------------------------------------
## -------------------------------------------------------------------------------------------------
class AdScenario4ADlof (OAStreamScenario):
C_NAME = 'AdScenario4ADlof'
## -------------------------------------------------------------------------------------------------
def _setup(self, p_mode, p_ada: bool, p_visualize: bool, p_logging):
# 1 Get the native stream from MLPro stream provider
mystream = StreamMLProPOutliers( p_functions = ['sin', 'cos', 'const'],
p_outlier_rate = 0.022,
p_seed = 6,
p_visualize = p_visualize,
p_logging = p_logging )
# 2 Creation of a workflow
workflow = OAStreamWorkflow( p_name = 'Anomaly detection using LOF@scikit-learn',
p_range_max = Range.C_RANGE_NONE,
p_ada = p_ada,
p_visualize = p_visualize,
p_logging = p_logging )
# 3 Initiailise the lof anomaly detector class
anomalydetector =WrSklearnLOF2MLPro( p_group_anomaly_det = False,
p_neighbours = 3,
p_delay = 3,
p_data_buffer = 5,
p_visualize = p_visualize,
p_logging = p_logging )
# 4 Add anomaly detection task to workflow
workflow.add_task( p_task = anomalydetector )
# 5 Return stream and workflow
return mystream, workflow
# 1 Demo setup
# 1.1 Default values
cycle_limit = 360
logging = Log.C_LOG_ALL
visualize = True
step_rate = 2
# 1.2 Welcome message
print('\n\n-----------------------------------------------------------------------------------------')
print('Publication: "MLPro 2.0 - Online machine learning in Python"')
print('Journal : ScienceDirect, Machine Learning with Applications (MLWA)')
print('Authors : D. Arend, L.S. Baheti, S. Yuwono, S.P.S. Kumar, A. Schwung')
print('Affiliation: South Westphalia University of Applied Sciences, Germany')
print('Sample : 3a Anomaly detection (3D)')
print('-----------------------------------------------------------------------------------------\n')
# 2 Instantiate the stream scenario
myscenario = AdScenario4ADlof( p_mode = Mode.C_MODE_REAL,
p_cycle_limit = cycle_limit,
p_visualize = visualize,
p_logging = logging )
# 3 Reset and run own stream scenario
myscenario.reset()
myscenario.init_plot( p_plot_settings=PlotSettings( p_view = PlotSettings.C_VIEW_ND,
p_view_autoselect = True,
p_step_rate = step_rate ) )
input('\nPlease arrange all windows and press ENTER to start stream processing...')
myscenario.run()
input('Press ENTER to exit...')