Add cost anomaly detection (day-over-day spend spike) — resolves #17

cloudcost.cost-anomaly-real.yml

@@ -0,0 +1,49 @@
+version: "1"
+
+pipeline:
+  name: cost-anomaly-real-test
+  mode: incremental
+  timezone: UTC
+
+sources:
+  - name: azure_billing
+    type: azure.cost_export
+    config:
+      account_url: "https://cloudcosttest36034.blob.core.windows.net"
+      container: "cost-exports"
+      prefix: "exports/"
+
+transforms:
+  - name: normalize_costs
+    type: focus.normalize
+    input: azure_billing
+    config:
+      provider: azure
+      column_map:
+        ServiceName: service_name
+        PreTaxCost: billed_cost
+        InstanceId: resource_id
+        ResourceGroup: resource_group
+        UsageDateTime: usage_date
+
+destinations:
+  - name: local_duckdb
+    type: duckdb
+    config:
+      path: ./data/cost-anomaly-real.duckdb
+
+loads:
+  - input: normalize_costs
+    destination: local_duckdb
+    table: fact_cost
+    mode: overwrite
+
+observability:
+  log_level: info
+  emit_run_summary: true
+
+policies:
+  - name: cost-anomaly
+    type: sql
+    query_file: policies/cost_anomaly.sql
+    severity: high

policies/cost_anomaly.sql

@@ -0,0 +1,61 @@
+-- Cost anomaly detection: flags a provider/service/resource whose daily
+-- spend jumps to >= 3x its own trailing 7-day average. Every other check
+-- in this project finds *persistent* waste; this finds *change* -- a
+-- misconfiguration or runaway process that started yesterday looks nothing
+-- like steady-state waste in a cost table, so a steady-state check will
+-- never catch it.
+--
+-- Reads from fact_cost (the normalized billing export produced by
+-- azure.cost_export + focus.normalize -- see cloudcost.cost-anomaly-real.yml),
+-- not from live inventory or metrics like most checks here.
+WITH daily AS (
+    SELECT
+        provider,
+        COALESCE(NULLIF(resource_id, ''), 'unknown') AS resource_id,
+        service_name,
+        CAST(usage_date AS DATE) AS usage_date,
+        SUM(billed_cost) AS daily_cost
+    FROM fact_cost
+    WHERE usage_date IS NOT NULL
+    GROUP BY
+        provider,
+        COALESCE(NULLIF(resource_id, ''), 'unknown'),
+        service_name,
+        CAST(usage_date AS DATE)
+),
+with_baseline AS (
+    SELECT
+        provider,
+        resource_id,
+        service_name,
+        usage_date,
+        daily_cost,
+        AVG(daily_cost) OVER (
+            PARTITION BY provider, resource_id, service_name
+            ORDER BY usage_date
+            ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
+        ) AS trailing_avg_7d,
+        COUNT(daily_cost) OVER (
+            PARTITION BY provider, resource_id, service_name
+            ORDER BY usage_date
+            ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
+        ) AS baseline_days
+    FROM daily
+)
+SELECT
+    provider,
+    resource_id,
+    service_name,
+    ROUND(daily_cost, 2) AS billed_cost,
+    usage_date,
+    ROUND(trailing_avg_7d, 2) AS trailing_avg_7d,
+    ROUND(daily_cost / trailing_avg_7d, 1) AS spike_multiplier,
+    service_name || ' spend spiked to $' || ROUND(daily_cost, 2)
+        || ' on ' || usage_date
+        || ' vs $' || ROUND(trailing_avg_7d, 2)
+        || ' trailing 7-day average ('
+        || ROUND(daily_cost / trailing_avg_7d, 1) || 'x)' AS evidence_reason
+FROM with_baseline
+WHERE baseline_days >= 7
+  AND trailing_avg_7d > 0
+  AND daily_cost > 3.0 * trailing_avg_7d;

tests/policies/test_cost_anomaly.py

@@ -0,0 +1,110 @@
+import pyarrow as pa
+
+
+def fact_cost(rows: list[dict]) -> pa.Table:
+    """Each row is provider, resource_id, service_name, billed_cost, usage_date."""
+    return pa.table(
+        {
+            "provider": pa.array([r["provider"] for r in rows], type=pa.string()),
+            "resource_id": pa.array([r["resource_id"] for r in rows], type=pa.string()),
+            "service_name": pa.array([r["service_name"] for r in rows], type=pa.string()),
+            "billed_cost": pa.array([r["billed_cost"] for r in rows], type=pa.float64()),
+            "usage_date": pa.array([r["usage_date"] for r in rows], type=pa.string()),
+        }
+    )
+
+
+def baseline_days(resource_id: str, n_days: int = 7, daily_cost: float = 10.0) -> list[dict]:
+    """n_days of steady $daily_cost spend for one resource, starting 2026-01-01."""
+    return [
+        {
+            "provider": "azure",
+            "resource_id": resource_id,
+            "service_name": "Virtual Machines",
+            "billed_cost": daily_cost,
+            "usage_date": f"2026-01-{day:02d}",
+        }
+        for day in range(1, n_days + 1)
+    ]
+
+
+def test_spend_spike_above_3x_trailing_average_is_flagged(run_policy) -> None:
+    rows = baseline_days("vm-a", n_days=7, daily_cost=10.0)
+    rows.append(
+        {
+            "provider": "azure",
+            "resource_id": "vm-a",
+            "service_name": "Virtual Machines",
+            "billed_cost": 100.0,
+            "usage_date": "2026-01-08",
+        }
+    )
+
+    findings = run_policy("cost_anomaly", {"fact_cost": fact_cost(rows)})
+
+    assert len(findings) == 1
+    finding = findings[0]
+    assert finding.provider == "azure"
+    assert finding.resource_id == "vm-a"
+    assert finding.service_name == "Virtual Machines"
+    assert finding.estimated_impact == 100.0
+    assert finding.evidence["spike_multiplier"] == 10.0
+    assert finding.evidence["trailing_avg_7d"] == 10.0
+
+
+def test_spike_below_3x_is_not_flagged(run_policy) -> None:
+    rows = baseline_days("vm-a", n_days=7, daily_cost=10.0)
+    rows.append(
+        {
+            "provider": "azure",
+            "resource_id": "vm-a",
+            "service_name": "Virtual Machines",
+            "billed_cost": 25.0,
+            "usage_date": "2026-01-08",
+        }
+    )
+
+    assert run_policy("cost_anomaly", {"fact_cost": fact_cost(rows)}) == []
+
+
+def test_steady_spend_is_not_flagged(run_policy) -> None:
+    assert run_policy("cost_anomaly", {"fact_cost": fact_cost(baseline_days("vm-a", n_days=10))}) == []
+
+
+def test_requires_a_full_7_day_baseline(run_policy) -> None:
+    # Only 2 days of history: a day-2 spike must not be flagged against a
+    # single-day "average" -- a real 7-day rolling average needs 7 prior days.
+    rows = baseline_days("vm-a", n_days=1, daily_cost=10.0)
+    rows.append(
+        {
+            "provider": "azure",
+            "resource_id": "vm-a",
+            "service_name": "Virtual Machines",
+            "billed_cost": 100.0,
+            "usage_date": "2026-01-02",
+        }
+    )
+
+    assert run_policy("cost_anomaly", {"fact_cost": fact_cost(rows)}) == []
+
+
+def test_anomaly_is_isolated_per_resource(run_policy) -> None:
+    rows = baseline_days("vm-a", n_days=8, daily_cost=10.0)
+    rows += baseline_days("vm-b", n_days=8, daily_cost=10.0)
+    rows.append(
+        {
+            "provider": "azure",
+            "resource_id": "vm-a",
+            "service_name": "Virtual Machines",
+            "billed_cost": 100.0,
+            "usage_date": "2026-01-09",
+        }
+    )
+
+    findings = run_policy("cost_anomaly", {"fact_cost": fact_cost(rows)})
+
+    assert [f.resource_id for f in findings] == ["vm-a"]
+
+
+def test_empty_table_returns_no_findings(run_policy) -> None:
+    assert run_policy("cost_anomaly", {"fact_cost": fact_cost([])}) == []