Python: Azure – Cost Anomaly Alerts from the Cost Management API

Azure Cost Management has budget alerts, but a budget is a monthly ceiling, and by the time you're close to it the anomaly that caused it has usually been running for a couple of weeks. What I actually want to know is "did yesterday look weird compared to the last two weeks," which a fixed budget threshold can't tell you. This script pulls daily actual cost from the Cost Management Query API, builds a rolling baseline, and posts a webhook alert when the spend for a day old enough to have its cost data in is a statistical outlier. It is one piece of the approach in Azure: Cost Governance Without Killing Developer Velocity; the budget-threshold companion is Node.js: Azure – A Slack Bot for Resource Group Budget Alerts.

Before writing your own, know what Azure already does. Cost Management runs its own anomaly detection on every subscription: it compares each day's total against a forecast trained on the last 60 days (a WaveNet-based model that accounts for patterns like Monday spikes), runs 36 hours after the end of the UTC day, and can email an anomaly alert when it finds something. Turn that on first (the API call is under Notes). This script earns its place when you want a different scope (a resource group, or one per team), a threshold you control, a chat channel instead of an inbox, or coverage where built-in anomaly alerts aren't offered: they exist only at subscription scope, allow five alert rules per subscription, and aren't available in Azure Government or other sovereign clouds.

Requirements

Python 3.10 or later with the azure-identity and requests packages (pip install azure-identity requests).
Cost Management Reader (or Cost Management Contributor) on the scope being watched. The role can be assigned at management group, subscription or resource group scope.
An identity DefaultAzureCredential can find: a managed identity when running in Azure, AZURE_CLIENT_ID / AZURE_TENANT_ID / AZURE_CLIENT_SECRET for a service principal, or an az login session for testing.
A Slack or Microsoft Teams incoming webhook URL that accepts a JSON body with a text field.
Environment variable COST_ALERT_WEBHOOK_URL (not needed with --dry-run), and AZURE_SUBSCRIPTION_ID unless you pass --scope.

Parameters

Name	Type	Required	Description
`--scope`	String	No	Cost Management scope to query, such as `/subscriptions/<id>/resourceGroups/<rg>`. Defaults to `/subscriptions/$AZURE_SUBSCRIPTION_ID`.
`--lookback-days`	Int	No	Settled days before the evaluated day used for the baseline. Default `14`.
`--threshold-stddev`	Float	No	Standard deviations from the baseline mean that count as an anomaly. Default `2.5`.
`--settle-days`	Int	No	Most recent days to skip because their data is still arriving. Default `2`.
`--min-delta`	Float	No	Ignore anomalies smaller than this absolute amount in the billing currency. Default `0`.
`--dry-run`	Switch	No	Print the result without posting to the webhook.

Usage

Run with defaults: a 14-day baseline, a 2.5 standard-deviation threshold, and the day two days ago as the evaluated day.

bash

export AZURE_SUBSCRIPTION_ID="<subscription-id>"
export COST_ALERT_WEBHOOK_URL="<https://hooks.example.com/incoming-webhook>"
python azure_cost_anomaly_alert.py --dry-run

text

Scope: /subscriptions/<subscription-id>
Evaluated 2025-12-22: 612.40 USD (baseline mean 431.18, stddev 38.92, z = 4.66)
Anomaly detected (above baseline).
Dry run: skipping webhook post.

Watch one resource group with a wider baseline, a lower threshold, and a floor so a jump from 3 to 9 dollars doesn't page anyone:

bash

python azure_cost_anomaly_alert.py \
  --scope "/subscriptions/<subscription-id>/resourceGroups/rg-prod-data" \
  --lookback-days 21 --threshold-stddev 2.0 --min-delta 50

Schedule it once a day. Cost data refreshes about every four hours and Microsoft recommends calling the cost APIs no more than once per day, so a morning cron entry is plenty:

bash

30 9 * * * cd /opt/cost-alerts && /opt/cost-alerts/.venv/bin/python azure_cost_anomaly_alert.py >> /var/log/cost-anomaly.log 2>&1

Script

python

"""
azure_cost_anomaly_alert.py

Pulls daily actual cost for an Azure scope from the Cost Management Query API,
compares the most recent settled day against the mean and standard deviation of
the preceding days, and posts a webhook alert if that day is an outlier.

Reads:
    POST {scope}/providers/Microsoft.CostManagement/query (api-version 2025-03-01)

Writes:
    An HTTP POST with a JSON "text" body to a Slack- or Teams-style incoming
    webhook when an anomaly is detected. No other side effects.

Environment variables:
    AZURE_SUBSCRIPTION_ID   Subscription used for the default scope.
    COST_ALERT_WEBHOOK_URL  Incoming webhook URL to post alerts to.
"""

import argparse
import os
import statistics
import sys
import time
from datetime import date, timedelta

import requests
from azure.identity import DefaultAzureCredential

ARM = "https://management.azure.com"
API_VERSION = "2025-03-01"
TOKEN_SCOPE = "https://management.azure.com/.default"
QPU_RETRY_HEADER = "x-ms-ratelimit-microsoft.costmanagement-qpu-retry-after"
MAX_ATTEMPTS = 4


def query_daily_costs(scope: str, start: date, end: date) -> tuple[dict, str]:
    """Return ({iso_date: cost}, currency) for every day from start to end inclusive."""
    token = DefaultAzureCredential().get_token(TOKEN_SCOPE).token
    url = f"{ARM}{scope}/providers/Microsoft.CostManagement/query?api-version={API_VERSION}"
    body = {
        "type": "ActualCost",
        "timeframe": "Custom",
        "timePeriod": {"from": start.isoformat(), "to": end.isoformat()},
        "dataset": {
            "granularity": "Daily",
            "aggregation": {"totalCost": {"name": "PreTaxCost", "function": "Sum"}},
        },
    }
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

    for attempt in range(1, MAX_ATTEMPTS + 1):
        response = requests.post(url, json=body, headers=headers, timeout=60)
        if response.status_code not in (429, 503) or attempt == MAX_ATTEMPTS:
            break
        # Throttled on query processing units (429) or service busy (503): back off as told.
        wait = response.headers.get(QPU_RETRY_HEADER) or response.headers.get("Retry-After") or "30"
        print(f"Throttled ({response.status_code}), retrying in {wait}s...", file=sys.stderr)
        time.sleep(float(wait))

    response.raise_for_status()
    properties = response.json()["properties"]

    if properties.get("nextLink"):
        print("Warning: response was paged; only the first page was read.", file=sys.stderr)

    columns = [col["name"] for col in properties["columns"]]
    date_index = columns.index("UsageDate")
    currency_index = columns.index("Currency") if "Currency" in columns else None
    cost_index = next(
        i for i, col in enumerate(properties["columns"])
        if col["type"] == "Number" and col["name"] != "UsageDate"
    )

    # Days with no usage are absent from the rows, so start every day at zero.
    costs = {(start + timedelta(days=n)).isoformat(): 0.0 for n in range((end - start).days + 1)}
    currency = ""
    for row in properties["rows"]:
        raw = str(int(row[date_index]))  # UsageDate is a number such as 20251222.
        day = date(int(raw[0:4]), int(raw[4:6]), int(raw[6:8])).isoformat()
        costs[day] = costs.get(day, 0.0) + float(row[cost_index])
        if currency_index is not None:
            currency = row[currency_index]

    return costs, currency


def evaluate(costs: dict, lookback_days: int, threshold: float, min_delta: float):
    """Compare the last day against the preceding lookback_days. Returns (result, is_anomaly)."""
    days = sorted(costs)
    if len(days) < lookback_days + 1:
        raise ValueError(f"Need {lookback_days + 1} days of data, got {len(days)}.")

    latest = days[-1]
    baseline = [costs[d] for d in days[-(lookback_days + 1):-1]]
    mean = statistics.mean(baseline)
    stddev = statistics.stdev(baseline)
    cost = costs[latest]

    result = {"date": latest, "cost": cost, "mean": mean, "stddev": stddev, "z": None}
    if stddev == 0:
        # A perfectly flat baseline: any change at all is worth a look.
        return result, cost != mean and abs(cost - mean) >= min_delta

    result["z"] = (cost - mean) / stddev
    return result, abs(result["z"]) >= threshold and abs(cost - mean) >= min_delta


def post_alert(webhook_url: str, scope: str, result: dict, currency: str) -> None:
    direction = "above" if result["cost"] > result["mean"] else "below"
    z_text = f"{result['z']:.2f}" if result["z"] is not None else "n/a (flat baseline)"
    text = (
        f"Azure cost anomaly on {result['date']} for `{scope}`\n"
        f"Cost: {result['cost']:,.2f} {currency} ({direction} baseline)\n"
        f"Baseline mean: {result['mean']:,.2f} {currency}, stddev {result['stddev']:,.2f}\n"
        f"Z-score: {z_text}"
    )
    response = requests.post(webhook_url, json={"text": text}, timeout=15)
    response.raise_for_status()


def main() -> int:
    parser = argparse.ArgumentParser(description="Detect Azure daily cost anomalies.")
    parser.add_argument("--scope", help="Cost Management scope, e.g. /subscriptions/<id>/resourceGroups/<rg>.")
    parser.add_argument("--lookback-days", type=int, default=14)
    parser.add_argument("--threshold-stddev", type=float, default=2.5)
    parser.add_argument("--settle-days", type=int, default=2)
    parser.add_argument("--min-delta", type=float, default=0.0)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    if args.lookback_days < 2:
        parser.error("--lookback-days must be at least 2.")

    scope = args.scope
    if not scope:
        subscription_id = os.environ.get("AZURE_SUBSCRIPTION_ID")
        if not subscription_id:
            print("Set AZURE_SUBSCRIPTION_ID or pass --scope.", file=sys.stderr)
            return 1
        scope = f"/subscriptions/{subscription_id}"
    scope = "/" + scope.strip("/")

    webhook_url = os.environ.get("COST_ALERT_WEBHOOK_URL")
    if not webhook_url and not args.dry_run:
        print("COST_ALERT_WEBHOOK_URL is not set.", file=sys.stderr)
        return 1

    # Evaluate a settled day, not today: recent days are still filling in.
    end = date.today() - timedelta(days=args.settle_days)
    start = end - timedelta(days=args.lookback_days)

    costs, currency = query_daily_costs(scope, start, end)
    result, is_anomaly = evaluate(costs, args.lookback_days, args.threshold_stddev, args.min_delta)

    z_text = f"{result['z']:.2f}" if result["z"] is not None else "n/a"
    print(f"Scope: {scope}")
    print(
        f"Evaluated {result['date']}: {result['cost']:,.2f} {currency} "
        f"(baseline mean {result['mean']:,.2f}, stddev {result['stddev']:,.2f}, z = {z_text})"
    )

    if not is_anomaly:
        print("No anomaly detected.")
        return 0

    print(f"Anomaly detected ({'above' if result['cost'] > result['mean'] else 'below'} baseline).")
    if args.dry_run:
        print("Dry run: skipping webhook post.")
        return 0

    post_alert(webhook_url, scope, result, currency)
    print("Webhook alert posted.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Notes

Don't evaluate today. An earlier draft compared "the most recent row" against the baseline, and the most recent row is a partial day, so it raised a false "drop" alert on most runs. Microsoft documents that cost data for EA and MCA subscriptions typically lands within 8 to 24 hours and can take up to 72 hours for pay-as-you-go, and its own anomaly detection waits 36 hours after the end of the UTC day. --settle-days 2 mirrors that; raise it to 3 for pay-as-you-go subscriptions.
Days with no usage don't come back as rows, so the script seeds every day in the window with zero. Without that, a quiet weekend silently shortens the baseline instead of pulling the mean down.
The Query API throttles by query processing units per tenant (currently 12 per 10 seconds, 60 per minute, 600 per hour, with one QPU per month of data queried). A throttled call returns 429 and the x-ms-ratelimit-microsoft.costmanagement-qpu-retry-after header, which the script honours. One run per scope per day is far below those limits; hundreds of resource-group scopes in a tight loop are not.
A threshold of 2.5 standard deviations flags about 1.2% of days (both sides combined) if daily cost were normally distributed. It usually isn't: weekly batch jobs make Mondays expensive on purpose. Either use --lookback-days in multiples of seven and compare same-weekday baselines, or rely on the built-in detector, which models weekly patterns for you.
The script alerts on drops as well as spikes. A sudden drop often means a workload stopped, or a backup job failed, all worth knowing.
Costs in Cost Management are estimates until the invoice is generated, and the billing period is finalized up to 72 hours after it ends. Small day-over-day revisions near month end are normal.
DefaultAzureCredential picks a managed identity automatically in Azure Automation, Functions or a VM; grant that identity Cost Management Reader and no secrets are needed. The same package exposes a management SDK (pip install azure-mgmt-costmanagement, CostManagementClient) if you prefer typed models over raw REST.

Turn on the built-in anomaly alert too

Microsoft's anomaly alerts are a scheduled action of kind InsightAlert. Creating one needs Cost Management Contributor (or Microsoft.CostManagement/scheduledActions/write). Because Azure checks the creator's permissions each time it sends the email, Microsoft suggests creating it as a service principal if your users hold elevated roles only through PIM. The viewId must point at the ms:DailyAnomalyByResourceGroup view or the rule won't appear in the portal:

http

PUT https://management.azure.com/subscriptions/<subscription-id>/providers/Microsoft.CostManagement/scheduledActions/daily-anomaly-alert?api-version=2025-03-01
Content-Type: application/json

{
  "kind": "InsightAlert",
  "properties": {
    "displayName": "Daily cost anomaly alert",
    "status": "Enabled",
    "viewId": "/subscriptions/<subscription-id>/providers/Microsoft.CostManagement/views/ms:DailyAnomalyByResourceGroup",
    "notification": {
      "to": ["finops@example.com"],
      "subject": "Cost anomaly detected"
    },
    "schedule": {
      "frequency": "Daily",
      "startDate": "2025-12-24T00:00:00Z",
      "endDate": "2026-12-24T00:00:00Z"
    }
  }
}

The email is sent once per anomaly, at detection, and includes the top resource group changes against the previous 60 days. When the endDate passes, the alert expires and stops sending, so put a renewal on the calendar.

Python: Azure – Cost Anomaly Alerts from the Cost Management API

$ grep -n '^#' post.md

Requirements

Parameters

Usage

Script

Notes

Turn on the built-in anomaly alert too

References