您现在访问的是微软AZURE全球版技术文档网站,若需要访问由世纪互联运营的MICROSOFT AZURE中国区技术文档网站,请访问 https://docs.azure.cn.

快速入门:使用异常检测器 REST API 和 Python 检测时序数据的异常Quickstart: Detect anomalies in your time series data using the Anomaly Detector REST API and Python

参考本快速入门可以开始使用异常检测器 API 的两种检测模式来检测时序数据的异常。Use this quickstart to start using the Anomaly Detector API's two detection modes to detect anomalies in your time series data. 此 Python 应用程序发送包含 JSON 格式的时序数据的 API 请求,并获取响应。This Python application sends API requests containing JSON-formatted time series data, and gets the responses.

API 请求API request 应用程序输出Application output
以批的形式检测异常Detect anomalies as a batch JSON 响应包含时序数据中每个数据点的异常状态(和其他数据),以及检测到的任何异常所在的位置。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 JSON 响应包含时序数据中最新数据点的异常状态(和其他数据)。The JSON response containing the anomaly status (and other data) for the latest data point in the time series data.
检测标记新数据趋势的更改点Detect change points that mark new data trends 包含时序数据中检测到的更改点的 JSON 响应。The JSON response containing the detected change points in the time series data.

虽然此应用程序是使用 Python 编写的,但 API 是一种 RESTful Web 服务,与大多数编程语言兼容。While this application is written in Python, the API is a RESTful web service compatible with most programming languages. 可在 GitHub 上找到本快速入门的源代码。You can find the source code for this quickstart on GitHub.

先决条件Prerequisites

  • Azure 订阅 - 免费创建订阅Azure subscription - Create one for free

  • 拥有 Azure 订阅后,可在 Azure 门户中创建异常检测器资源来获取密钥和终结点。Once you have your Azure subscription, create an Anomaly Detector resource in the Azure portal to get your key and endpoint. 等待其部署并单击“转到资源”按钮。Wait for it to deploy and click the Go to resource button.

    • 需要从创建的资源获取密钥和终结点,以便将应用程序连接到异常检测器 API。You will need the key and endpoint from the resource you create to connect your application to the Anomaly Detector API. 你稍后会在快速入门中将密钥和终结点粘贴到下方的代码中。You'll paste your key and endpoint into the code below later in the quickstart. 可以使用免费定价层 (F0) 试用该服务,然后再升级到付费层进行生产。You can use the free pricing tier (F0) to try the service, and upgrade later to a paid tier for production.
  • Python 2.x 或 3.xPython 2.x or 3.x

  • Python 的请求库The Requests library for python

  • 包含时序数据点的 JSON 文件。A JSON file containing time series data points. 可在 GitHub 上找到本快速入门的示例数据。The example data for this quickstart can be found on GitHub.

创建环境变量Create an environment variable

备注

在 2019 年 7 月 1 日之后创建的非试用资源的终结点使用如下所示的自定义子域格式。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 - 用于验证请求的资源密钥。ANOMALY_DETECTOR_KEY - The resource key for authenticating your requests.
  • ANOMALY_DETECTOR_ENDPOINT - 用于发送 API 请求的资源终结点。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. 创建新的 python 文件并添加以下导入。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. 下面是可用于异常检测的 URI。Below are the URIs you can use for anomaly detection. 稍后会将这些 URL 追加到服务终结点,以创建 API 请求 URL。These will be appended to your service endpoint later to create the API request URLs.

    检测方法Detection method URIURI
    批量检测Batch detection /anomalydetector/v1.0/timeseries/entire/detect
    对最新数据点进行检测Detection on the latest data point /anomalydetector/v1.0/timeseries/last/detect
    更改点检测Change point detection /anomalydetector/v1.0/timeseries/changepoint/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"
    change_point_detection_url = "/anomalydetector/v1.0/timeseries/changepoint/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. 打开 JSON 数据文件,并使用 json.load() 读入该文件。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. 创建名为 send_request() 的新函数,该函数使用上面创建的变量。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. Content-Type 设置为 application/json,并将订阅密钥添加到 Ocp-Apim-Subscription-Key 标头。Set the Content-Type to application/json, and add your subscription key to the Ocp-Apim-Subscription-Key header.

  3. 使用 requests.post() 发送请求。Send the request using requests.post(). 合并终结点和异常检测 URL 以构成完整的请求 URL,并包含标头和 JSON 请求数据。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. 创建名为 detect_batch() 的方法,以批的形式检测整个数据中的异常。Create a method called detect_batch() to detect anomalies throughout the data as a batch. 调用在上面使用终结点、URL、订阅密钥和 JSON 数据创建的 send_request() 方法。Call the send_request() method created above with your endpoint, URL, subscription key, and json data.

  2. 针对结果调用 json.dumps() 以设置其格式,然后将结果输出到控制台。Call json.dumps() on the result to format it, and print it to the console.

  3. 如果响应包含 code 字段,请输出错误代码和错误消息。If the response contains code field, print the error code and error message.

  4. 否则,请查找异常在数据集中的位置。Otherwise, find the positions of anomalies in the data set. 响应的 isAnomaly 字段包含一个布尔值,该值与给定的数据点是否为异常相关。The response's isAnomaly field contains a boolean value relating to whether a given data point is an anomaly. 循环访问该列表,并输出任何 True 值的索引。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. 创建名为 detect_latest() 的方法,以确定时序中的最新数据点是否为异常。Create a method called detect_latest() to determine if the latest data point in your time series is an anomaly. 结合终结点、URL、订阅密钥和 JSON 数据调用上述 send_request() 方法。Call the send_request() method above with your endpoint, URL, subscription key, and json data.

  2. 针对结果调用 json.dumps() 以设置其格式,然后将结果输出到控制台。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))
    

检测数据中的更改点Detect change points in the data

  1. 创建名为 detect_change_point() 的方法,以批的形式检测整个数据中的异常。Create a method called detect_change_point() to detect anomalies throughout the data as a batch. 调用在上面使用终结点、URL、订阅密钥和 JSON 数据创建的 send_request() 方法。Call the send_request() method created above with your endpoint, URL, subscription key, and json data.

  2. 针对结果调用 json.dumps() 以设置其格式,然后将结果输出到控制台。Call json.dumps() on the result to format it, and print it to the console.

  3. 如果响应包含 code 字段,请输出错误代码和错误消息。If the response contains a code field, print the error code and error message.

  4. 否则,请查找异常在数据集中的位置。Otherwise, find the positions of anomalies in the data set. 响应的 isChangePoint 字段包含一个布尔值,该值指示给定的数据点是否为异常。The response's isChangePoint field contains a boolean value indicating whether a given data point is an anomaly. 循环访问该列表,并输出任何 True 值的索引。Iterate through the list, and print the index of any True values. 如果找到任何此类值,这些值对应于趋势更改点的索引。These values correspond to the indices of trend change points points, if any were found.

    def detect_change_point(request_data):
        print("Detecting change point")
        # send the request, and print the JSON result
        result = send_request(endpoint, change_point_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 changePoint in the data set
            change_points = result["isChangePoint"]
            print("changePoints detected in the following data positions:")
    
            for x in range(len(change_points)):
                if change_points[x]:
                    print(x, request_data['series'][x]['value'])
    

发送请求Send the request

调用上面创建的异常检测方法。Call the anomaly detection methods created above.

detect_batch(json_data)
detect_latest(json_data)
detect_change_point(json_data)

示例响应Example response

成功的响应以 JSON 格式返回。A successful response is returned in JSON format. 单击以下链接在 GitHub 上查看 JSON 响应: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: