Quickstart: Detect anomalies in your time series data using the Anomaly Detector REST API and Python

Use this quickstart to start using the Anomaly Detector API's two detection modes to detect anomalies in your time series data. This Python application sends two API requests containing JSON-formatted time series data, and gets the responses.

API request Application output
Detect anomalies as a batch The JSON response containing the anomaly status (and other data) for each data point in the time series data, and the positions of any detected anomalies.
Detect the anomaly status of the latest data point The JSON response containing the anomaly status (and other data) for the latest data point in the time series data.

While this application is written in Python, the API is a RESTful web service compatible with most programming languages. You can find the source code for this quickstart on GitHub.

Prerequisites

  • Python 2.x or 3.x

  • An Anomaly detector key and endpoint

  • The Requests library for python

  • A JSON file containing time series data points. The example data for this quickstart can be found on GitHub.

Create an Anomaly Detector resource

Start using the Anomaly Detector service by creating one of the Azure resources below.

Create an environment variable

Note

The endpoints for non-trial resources created after July 1, 2019 use the custom subdomain format shown below. For more information and a complete list of regional endpoints, see Custom subdomain names for Cognitive Services.

Using your key and endpoint from the resource you created, create two environment variables for authentication:

  • ANOMALY_DETECTOR_KEY - The resource key for authenticating your requests.
  • ANOMALY_DETECTOR_ENDPOINT - The resource endpoint for sending API requests. It will look like this:
    • https://<your-custom-subdomain>.api.cognitive.microsoft.com

Use the instructions for your operating system.

setx ANOMALY_DETECTOR_KEY <replace-with-your-anomaly-detector-key>
setx ANOMALY_DETECTOR_ENDPOINT <replace-with-your-anomaly-detector-endpoint>

After you add the environment variable, restart the console window.

Create a new application

  1. Create a new python file and add the following imports.

    import os
    import requests
    import json
    
  2. Create variables for your subscription key and your endpoint. Below are the URIs you can use for anomaly detection. These will be appended to your service endpoint later to create the API request URLs.

    Detection method URI
    Batch detection /anomalydetector/v1.0/timeseries/entire/detect
    Detection on the latest data point /anomalydetector/v1.0/timeseries/last/detect
    # URLs for anomaly detection with the Anomaly Detector API
    batch_detection_url = "/anomalydetector/v1.0/timeseries/entire/detect"
    latest_point_detection_url = "/anomalydetector/v1.0/timeseries/last/detect"
    
    # This sample assumes you have created an environment variable for your key and endpoint
    endpoint = os.environ["ANOMALY_DETECTOR_ENDPOINT"]
    subscription_key = os.environ["ANOMALY_DETECTOR_KEY"]
    
    # Replace with a path to the JSON formatted time series data.
    data_location = "[PATH_TO_TIME_SERIES_DATA]"
    
  3. Read in the JSON data file by opening it, and using json.load().

    file_handler = open(data_location)
    json_data = json.load(file_handler)
    

Create a function to send requests

  1. Create a new function called send_request() that takes the variables created above. Then perform the following steps.

  2. Create a dictionary for the request headers. Set the Content-Type to application/json, and add your subscription key to the Ocp-Apim-Subscription-Key header.

  3. Send the request using requests.post(). Combine your endpoint and anomaly detection URL for the full request URL, and include your headers, and json request data. And then return the response.

    def send_request(endpoint, url, subscription_key, request_data):
        headers = {'Content-Type': 'application/json', 'Ocp-Apim-Subscription-Key': subscription_key}
        response = requests.post(endpoint+url, data=json.dumps(request_data), headers=headers)
        return json.loads(response.content.decode("utf-8"))
    

Detect anomalies as a batch

  1. Create a method called detect_batch() to detect anomalies throughout the data as a batch. Call the send_request() method created above with your endpoint, url, subscription key, and json data.

  2. Call json.dumps() on the result to format it, and print it to the console.

  3. If the response contains code field, print the error code and error message.

  4. Otherwise, find the positions of anomalies in the data set. The response's isAnomaly field contains a boolean value relating to whether a given data point is an anomaly. Iterate through the list, and print the index of any True values. These values correspond to the index of anomalous data points, if any were found.

    def detect_batch(request_data):
        print("Detecting anomalies as a batch")
        # Send the request, and print the JSON result
        result = send_request(endpoint, batch_detection_url, subscription_key, request_data)
        print(json.dumps(result, indent=4))
    
        if result.get('code') is not None:
            print("Detection failed. ErrorCode:{}, ErrorMessage:{}".format(result['code'], result['message']))
        else:
            # Find and display the positions of anomalies in the data set
            anomalies = result["isAnomaly"]
            print("Anomalies detected in the following data positions:")
    
            for x in range(len(anomalies)):
                if anomalies[x]:
                    print (x, request_data['series'][x]['value'])
    

Detect the anomaly status of the latest data point

  1. Create a method called detect_latest() to determine if the latest data point in your time series is an anomaly. Call the send_request() method above with your endpoint, url, subscription key, and json data.

  2. Call json.dumps() on the result to format it, and print it to the console.

    def detect_latest(request_data):
        print("Determining if latest data point is an anomaly")
        # send the request, and print the JSON result
        result = send_request(endpoint, latest_point_detection_url, subscription_key, request_data)
        print(json.dumps(result, indent=4))
    

Send the request

Call the anomaly detection methods created above.

detect_batch(json_data)
detect_latest(json_data)

Example response

A successful response is returned in JSON format. Click the links below to view the JSON response on GitHub:

Clean up resources

If you want to clean up and remove a Cognitive Services subscription, you can delete the resource or resource group. Deleting the resource group also deletes any other resources associated with the resource group.

Next steps

Concepts:

Tutorials: