Anomaly Detection Python Example

anomalydetection.py

import requests
import json
import pandas as pd
import matplotlib.pyplot as plt

subscription_key = "xxxxxxx"
endpoint = "https://westeurope.api.cognitive.microsoft.com"
batch_detection_url = "/anomalydetector/v1.0/timeseries/entire/detect"
latest_point_detection_url = "/anomalydetector/v1.0/timeseries/last/detect"

source_data = requests.get("https://raw.githubusercontent.com/Azure-Samples/anomalydetector/master/example-data/request-data.json").json()

headers = {'Content-Type': 'application/json',
           'Ocp-Apim-Subscription-Key': subscription_key}

response = requests.post(endpoint + batch_detection_url, data=json.dumps(source_data), headers=headers)
response = json.loads(response.content.decode("utf-8"))

# Aggregate inputs and results
df_reponse = pd.DataFrame.from_dict(response)
df_source = pd.DataFrame(source_data['series'])
df = pd.concat([df_source, df_reponse], axis=1)

# Display results
plt.figure(1, figsize=(20, 8))

plt.plot(df.timestamp, df.value, color='black', label='Actual Values')
plt.plot(df.timestamp, df.expectedValues, linestyle='dashed', color='blue', label='Expected Values')
plt.fill_between(df.timestamp, df.expectedValues - df.lowerMargins, df.expectedValues + df.upperMargins,
                 color='blue', alpha=0.1)

anomalies = df[(df.isAnomaly == True)]
plt.scatter(anomalies.timestamp, anomalies.value, c='r', alpha=0.5, label='Anomalies')

plt.xlabel('Time')
plt.xticks(df.timestamp, rotation='vertical')
plt.ylabel('Values')
plt.legend()

plt.show()