Cognitive Services Anomaly Detector client library for Python

Anomaly Detector 是一个具有一组API的AI服务，它允许您在几乎不需要机器学习（ML）知识的情况下，通过批量验证或实时推理来监控和检测时间序列数据中的异常。

入门

先决条件

• 要使用此包，需要Python 3.7或更高版本。
• 您需要一个 Azure订阅 来使用此包。
• 现有的Cognitive Services Anomaly Detector实例。

安装包

python -m pip install azure-ai-anomalydetector

注意：此客户端库版本的默认服务版本为 3.0.0b6。

此表显示了SDK版本与服务支持的API版本之间的关系

验证客户端

获取端点

您可以使用 Azure Portal 或 Azure CLI 查找您的异常检测服务资源端点。

# Get the endpoint for the Anomaly Detector service resource
az cognitiveservices account show --name "resource-name" --resource-group "resource-group-name" --query "properties.endpoint"

获取API密钥

您可以从Azure Portal中的异常检测服务资源中获取 API密钥。或者，您可以使用以下 Azure CLI 脚本片段来获取您的资源API密钥。

az cognitiveservices account keys list --resource-group <your-resource-group-name> --name <your-resource-name>

使用API密钥凭据创建AnomalyDetectorClient

一旦您获得了API密钥的值，您可以将其作为字符串传递给AzureKeyCredential的实例。使用该密钥作为凭据参数来验证客户端。

from azure.core.credentials import AzureKeyCredential
from azure.ai.anomalydetector import AnomalyDetectorClient

credential = AzureKeyCredential("<api_key>")
client = AnomalyDetectorClient(endpoint="https://<resource-name>.cognitiveservices.azure.com/", credential=credential)

关键概念

使用异常检测器，您可以使用 单变量异常检测 在一个变量中检测异常，或者使用 多变量异常检测 在多个变量中检测异常。

单变量异常检测

单变量异常检测API允许您监控和检测时间序列数据中的异常，而无需了解机器学习。算法会自动识别并应用最适合您数据的模型，无论行业、场景或数据量。使用您的时间序列数据，API确定异常检测的边界、期望值以及哪些数据点是异常。

使用异常检测器不需要任何机器学习经验，REST API允许您轻松地将服务集成到您的应用程序和流程中。

使用单变量异常检测，您可以自动检测整个时间序列数据中的异常，或者实时发生时的异常。

多变量异常检测

多变量异常检测 API进一步通过轻松集成用于检测组指标异常的高级AI来帮助开发者，而无需机器学习知识或标记数据。现在自动计算多达300个不同信号的依赖性和互相关性作为关键因素。这项新功能帮助您积极保护您的复杂系统，如软件应用程序、服务器、工厂机器、航天器，甚至您的业务，免受故障的影响。

使用多变量异常检测，您可以自动检测整个时间序列数据中的异常，或实时发生时的异常。使用多变量异常检测有三个步骤。

• 训练：使用Train Model API创建和训练模型，然后使用Get Model Status API获取状态和模型元数据。
• 推理:
  • 使用Async Inference API触发异步推理过程，然后使用Get Inference results API获取一批数据的检测结果。
  • 您还可以使用Sync Inference API在每次触发检测时针对每个时间戳。
• 其他操作：在多元异常检测模型中，支持列表模型API和删除模型API进行模型管理。

线程安全性

我们保证所有客户端实例方法都是线程安全的，并且彼此独立（指南）。这确保了重用客户端实例的建议总是安全的，即使在跨线程的情况下。

示例

以下部分提供了几个代码片段，涵盖了Anomaly Detector服务的一些最常见任务，包括

• 单变量异常检测 - 批量检测
• 单变量异常检测 - 流式检测
• 单变量异常检测 - 检测变化点
• 多变量异常检测

批量检测

from azure.ai.anomalydetector import AnomalyDetectorClient
from azure.core.credentials import AzureKeyCredential
from azure.ai.anomalydetector.models import *


SUBSCRIPTION_KEY = os.environ["ANOMALY_DETECTOR_KEY"]
ANOMALY_DETECTOR_ENDPOINT = os.environ["ANOMALY_DETECTOR_ENDPOINT"]
TIME_SERIES_DATA_PATH = os.path.join("sample_data", "request-data.csv")
client = AnomalyDetectorClient(ANOMALY_DETECTOR_ENDPOINT, AzureKeyCredential(SUBSCRIPTION_KEY))

series = []
data_file = pd.read_csv(TIME_SERIES_DATA_PATH, header=None, encoding="utf-8", parse_dates=[0])
for index, row in data_file.iterrows():
    series.append(TimeSeriesPoint(timestamp=row[0], value=row[1]))

request = UnivariateDetectionOptions(
    series=series,
    granularity=TimeGranularity.DAILY,
)


if any(response.is_anomaly):
    print("An anomaly was detected at index:")
    for i, value in enumerate(response.is_anomaly):
        if value:
            print(i)
else:
    print("No anomalies were detected in the time series.")

流式检测

from azure.ai.anomalydetector import AnomalyDetectorClient
from azure.core.credentials import AzureKeyCredential
from azure.ai.anomalydetector.models import *


SUBSCRIPTION_KEY = os.environ["ANOMALY_DETECTOR_KEY"]
ANOMALY_DETECTOR_ENDPOINT = os.environ["ANOMALY_DETECTOR_ENDPOINT"]
TIME_SERIES_DATA_PATH = os.path.join("sample_data", "request-data.csv")
client = AnomalyDetectorClient(ANOMALY_DETECTOR_ENDPOINT, AzureKeyCredential(SUBSCRIPTION_KEY))

series = []
data_file = pd.read_csv(TIME_SERIES_DATA_PATH, header=None, encoding="utf-8", parse_dates=[0])
for index, row in data_file.iterrows():
    series.append(TimeSeriesPoint(timestamp=row[0], value=row[1]))

request = UnivariateDetectionOptions(
    series=series,
    granularity=TimeGranularity.DAILY,
)
print("Detecting the anomaly status of the latest data point.")

if response.is_anomaly:
    print("The latest point is detected as anomaly.")
else:
    print("The latest point is not detected as anomaly.")

检测变化点

from azure.ai.anomalydetector import AnomalyDetectorClient
from azure.core.credentials import AzureKeyCredential
from azure.ai.anomalydetector.models import *


SUBSCRIPTION_KEY = os.environ["ANOMALY_DETECTOR_KEY"]
ANOMALY_DETECTOR_ENDPOINT = os.environ["ANOMALY_DETECTOR_ENDPOINT"]
TIME_SERIES_DATA_PATH = os.path.join("sample_data", "request-data.csv")
client = AnomalyDetectorClient(ANOMALY_DETECTOR_ENDPOINT, AzureKeyCredential(SUBSCRIPTION_KEY))

series = []
data_file = pd.read_csv(TIME_SERIES_DATA_PATH, header=None, encoding="utf-8", parse_dates=[0])
for index, row in data_file.iterrows():
    series.append(TimeSeriesPoint(timestamp=row[0], value=row[1]))

request = UnivariateChangePointDetectionOptions(
    series=series,
    granularity=TimeGranularity.DAILY,
)


if any(response.is_change_point):
    print("An change point was detected at index:")
    for i, value in enumerate(response.is_change_point):
        if value:
            print(i)
else:
    print("No change point were detected in the time series.")

多元异常检测示例

要了解如何使用Anomaly Detector库进行多元异常检测，请参阅此示例。

要获取Anomaly Detector包的更多详细信息，请参阅此azure.ai.anomalydetector包。

故障排除

一般

Anomaly Detector客户端库将抛出在Azure Core中定义的异常。

日志记录

此库使用标准的logging库进行日志记录。

HTTP会话的基本信息（URL、标题等）将在INFO级别进行记录。

详细的DEBUG级别日志记录，包括请求/响应体和unredacted标题，可以通过在客户端或每个操作中启用logging_enable关键字参数来启用。

有关示例，请参阅完整的SDK日志记录文档此处。

可选配置

可以在客户端和每个操作级别传递可选关键字参数。azure-core 参考文档描述了重试、日志记录、传输协议等可用配置。

下一步

这些代码示例显示了Azure Anomaly Detector库的常见场景操作。更多示例可以在samples目录下找到。

• 单变量异常检测 - 批量检测: sample_detect_entire_series_anomaly.py

• 单变量异常检测 - 流式检测: sample_detect_last_point_anomaly.py

• 单变量异常检测 - 变化点检测: sample_detect_change_point.py

• 多元异常检测: sample_multivariate_detect.py

其他文档

有关Azure Anomaly Detector的更详细文档，请参阅docs.microsoft.com上的Anomaly Detector文档。

贡献

此项目欢迎贡献和建议。大多数贡献都需要您同意贡献者许可协议（CLA），声明您有权并且实际上确实授予我们使用您的贡献的权利。有关详细信息，请访问CLA主页。

当您提交拉取请求时，CLA-bot将自动确定您是否需要提供CLA，并适当装饰PR（例如，标签、评论）。只需遵循bot提供的说明即可。您只需在整个使用我们的CLA的所有存储库中这样做一次。

本项目采用了微软开源行为准则。如需更多信息，请参阅行为准则常见问题解答或联系opencode@microsoft.com，如有任何其他问题或意见。