File: test_anomaly_detector.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (83 lines) | stat: -rw-r--r-- 2,369 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
#!/usr/bin/env python
# coding=utf-8

import functools
import pytest
import time
import os
from datetime import datetime

from devtools_testutils import (
    AzureRecordedTestCase,
    PowerShellPreparer,
    recorded_by_proxy,
)
from azure.ai.anomalydetector import AnomalyDetectorClient
from azure.ai.anomalydetector.models import *
from azure.core.credentials import AzureKeyCredential

from test_data import get_test_data

AnomalyDetectorEnvPreparer = functools.partial(
    PowerShellPreparer,
    "anomaly_detector",
    anomaly_detector_endpoint="https://fake_ad_resource.cognitiveservices.azure.com/",
    anomaly_detector_key="00000000000000000000000000000000",
)


class TestAnomalyDetector(AzureRecordedTestCase):
    @AnomalyDetectorEnvPreparer()
    @recorded_by_proxy
    def test_entire_detect(self, anomaly_detector_endpoint, anomaly_detector_key):

        ad_client = AnomalyDetectorClient(anomaly_detector_endpoint, AzureKeyCredential(anomaly_detector_key))
        assert ad_client is not None

        data = get_test_data()
        series = []
        time_format = "%Y-%m-%dT%H:%M:%SZ"
        for i in range(len(data["timestamp"])):
            series.append(
                TimeSeriesPoint(timestamp=datetime.strptime(data["timestamp"][i], time_format), value=data["value"][i])
            )

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

        response = ad_client.detect_univariate_entire_series(request)
        detect_index = []
        for i, value in enumerate(response.is_anomaly):
            if value:
                detect_index.append(i)

        assert detect_index == [
            3,
            18,
            21,
            22,
            23,
            24,
            25,
            28,
            29,
            30,
            31,
            32,
            35,
            44,
        ]

    @AnomalyDetectorEnvPreparer()
    @recorded_by_proxy
    def test_multi_ad_list_model(self, anomaly_detector_endpoint, anomaly_detector_key):

        ad_client = AnomalyDetectorClient(anomaly_detector_endpoint, AzureKeyCredential(anomaly_detector_key))
        assert ad_client is not None

        models = ad_client.list_multivariate_models(skip=0, top=10)
        model_count = len(list(models))

        assert model_count >= 0