2.8 KiB
| title | sidebar_label |
|---|---|
| Anomaly Detection Algorithms | Anomaly Detection Algorithms |
Input Limitations
execute is the core method of anomaly detection algorithms. Before calling this method, the framework configures the historical time-series data used for anomaly detection in the self.list object attribute.
Output Limitations
The execute method returns an array of the same length as self.list. A value of -1 in the array indicates an anomaly.
For example, in the series [2, 2, 2, 2, 100], assuming that 100 is an anomaly, the method returns [1, 1, 1, 1, -1].
Sample Code
This section describes an example anomaly detection algorithm that returns the final data point in a time series as an anomaly.
from taosanalytics.service import AbstractAnomalyDetectionService
# Algorithm files must start with an underscore ("_") and end with "Service".
class _MyAnomalyDetectionService(AbstractAnomalyDetectionService):
""" Define a class inheriting from AbstractAnomalyDetectionService and implementing the abstract method of that class. """
# Name the algorithm using only lowercase ASCII characters.
name = 'myad'
# Include a description of the algorithm (recommended)
desc = """return the last value as the anomaly data"""
def __init__(self):
"""Method to initialize the class"""
super().__init__()
def execute(self):
""" Implementation of algorithm logic"""
"""Create an array with length len(self.list) whose results are all 1, then set the final value in the array to -1 to indicate an anomaly"""
res = [1] * len(self.list)
res[-1] = -1
"""Return results"""
return res
def set_params(self, params):
"""This algorithm does not take any parameters, so this logic is not included."""
Save this file to the ./lib/taosanalytics/algo/ad/ directory and restart the taosanode service. In the TDengine CLI, run SHOW ANODES FULL to see your new algorithm. Your applications can now invoke this algorithm via SQL.
--- Detect anomalies in the `col` column using the newly added `myad` algorithm
SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col, 'algo=myad')
If you have never started the anode, see Installation to add the anode to your TDengine cluster.
Unit Testing
You can add unit test cases to the anomaly_test.py file in the taosanalytics/test directory or create a file for unit tests. The framework uses the Python unittest module.
def test_myad(self):
""" Test the _IqrService class """
s = loader.get_service("myad")
# Configure the data to test
s.set_input_list(AnomalyDetectionTest.input_list, None)
r = s.execute()
# The final value is an anomaly
self.assertEqual(r[-1], -1)
self.assertEqual(len(r), len(AnomalyDetectionTest.input_list))