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."""
        return super().set_params(params)

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))