快速入門:使用異常偵測器 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 two 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.

雖然此應用程式是以 Python 撰寫的,但 API 是一種與大多數程式設計語言都相容的 RESTful Web 服務。While this application is written in Python, the API is a RESTful web service compatible with most programming languages.

必要條件Prerequisites

注意

使用異常偵測器 API 時若要取得最佳結果,您的 JSON 格式時間序列資料應該包含:For best results when using the Anomaly Detector API, your JSON-formatted time series data should include:

  • 依據相同間隔分隔的資料點,遺漏的點數目不超過預期的 10%。data points separated by the same interval, with no more than 10% of the expected number of points missing.
  • 如果您的資料沒有清楚的季節模式,則至少有 12 個資料點。at least 12 data points if your data doesn't have a clear seasonal pattern.
  • 如果您的資料有清楚的季節模式,則至少會發生 4 個模式。at least 4 pattern occurrences if your data does have a clear seasonal pattern.

您必須有具備 Anomaly Detector API 存取權的認知服務 API 帳戶You must have a Cognitive Services API account with access to the Anomaly Detector API. 您可以在建立帳戶後從 Azure 入口網站取得訂用帳戶金鑰。You can get your subscription key from the Azure portal after creating your account.

建立新的應用程式Create a new application

  1. 在您最愛的文字編輯器或 IDE 中建立新的 Python 檔案。In your favorite text editor or IDE, create a new python file. 新增下列匯入。Add the following imports.

    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. 這些稍後將會附加至您的服務端點,以建立 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
    batch_detection_url = "/anomalydetector/v1.0/timeseries/entire/detect"
    latest_point_detection_url = "/anomalydetector/v1.0/timeseries/last/detect"
    
    endpoint = "[YOUR_ENDPOINT_URL]"
    subscription_key = "[YOUR_SUBSCRIPTION_KEY]"
    data_location = "[PATH_TO_TIME_SERIES_DATA]"
    
  3. 開啟檔案並使用 json.load() 來讀取 JSON 資料檔案。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")
    result = send_request(endpoint, batch_detection_url,
                          subscription_key, request_data)
    print(json.dumps(result, indent=4))

    if result.get('code') != 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] == True:
                print(x)

偵測最新資料點的異常狀態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))

載入時間序列資料,並傳送要求Load your time series data and send the request

  1. 載入 JSON 時間序列資料,開啟檔案處理常式,並在其上使用 json.load()Load your JSON time series data opening a file handler, and using json.load() on it. 接著呼叫上面所建立的異常偵測方法。Then call the anomaly detection methods created above.
file_handler = open(data_location)
json_data = json.load(file_handler)

detect_batch(json_data)
detect_latest(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:

後續步驟Next steps