# Aggregating Compliance Metrics Across a Dataset Inventory

To aggregate compliance metrics across a dataset inventory, load the per-dataset check results from SQLite into a pandas DataFrame and reduce them with groupby().mean() — because the pass column is 0 or 1, the mean is the pass rate directly, and a row-wise AND across checks yields an overall compliance score per dataset.

This is the computational core behind Compliance Dashboards for Spatial Catalogs, which covers where these numbers are rendered and gated, and it belongs to the broader Spatial Data Audit Reporting & Compliance Governance program. The script below takes a stored results table and produces the by-check, by-collection, and overall figures a dashboard consumes.

# Automated Python Implementation

The script is self-contained: it seeds a small SQLite table so it runs with no external data, then aggregates it three ways. Point --db at your real check_results store to run it against a production catalog.

#!/usr/bin/env python3
"""Aggregate compliance metrics from a SQLite check_results table.

Usage:
    python aggregate_metrics.py --db compliance.db --out metrics.json

With no pre-populated database, --seed inserts a demo run so the script
is runnable standalone.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import sqlite3

import pandas as pd

SCHEMA = """
CREATE TABLE IF NOT EXISTS check_results (
    dataset_id TEXT NOT NULL,
    collection TEXT,
    check_name TEXT NOT NULL,
    passed     INTEGER NOT NULL,
    run_at     TEXT NOT NULL
);
"""

DEMO_ROWS = [
    ("roads_2025", "transport", "valid_crs", 1),
    ("roads_2025", "transport", "has_license", 1),
    ("roads_2025", "transport", "schema_pass", 1),
    ("parcels_2025", "cadastre", "valid_crs", 1),
    ("parcels_2025", "cadastre", "has_license", 0),
    ("parcels_2025", "cadastre", "schema_pass", 1),
    ("flood_2025", "hazard", "valid_crs", 0),
    ("flood_2025", "hazard", "has_license", 1),
    ("flood_2025", "hazard", "schema_pass", 0),
]

def seed(conn: sqlite3.Connection) -> None:
    """Insert one demo run so the script is runnable with no real data."""
    run_at = dt.datetime.now(dt.timezone.utc).isoformat()
    conn.executescript(SCHEMA)
    conn.executemany(
        "INSERT INTO check_results "
        "(dataset_id, collection, check_name, passed, run_at) "
        "VALUES (?, ?, ?, ?, ?)",
        [(d, c, n, p, run_at) for (d, c, n, p) in DEMO_ROWS],
    )
    conn.commit()

def load_latest(conn: sqlite3.Connection) -> pd.DataFrame:
    """Load the newest run's rows into a DataFrame."""
    latest = conn.execute("SELECT MAX(run_at) FROM check_results").fetchone()[0]
    return pd.read_sql_query(
        "SELECT * FROM check_results WHERE run_at = ?", conn, params=(latest,)
    )

def aggregate(df: pd.DataFrame) -> dict:
    """Compute by-check, by-collection, and overall compliance metrics."""
    by_check = (
        df.groupby("check_name")["passed"].mean().mul(100).round(1)
    ).to_dict()

    by_collection = (
        df.groupby("collection")["passed"].mean().mul(100).round(1)
    ).to_dict()

    # A dataset is fully compliant only if every one of its checks passed.
    wide = df.pivot_table(
        index="dataset_id", columns="check_name",
        values="passed", aggfunc="min",
    )
    fully_compliant = wide.all(axis=1)
    overall = round(float(fully_compliant.mean()) * 100, 1)

    return {
        "run_at": df["run_at"].iloc[0],
        "datasets": int(df["dataset_id"].nunique()),
        "overall_compliant_pct": overall,
        "pass_rate_by_check_pct": by_check,
        "pass_rate_by_collection_pct": by_collection,
        "non_compliant_datasets": sorted(
            fully_compliant[~fully_compliant].index.tolist()
        ),
    }

def main() -> None:
    parser = argparse.ArgumentParser(description="Aggregate compliance metrics.")
    parser.add_argument("--db", default=":memory:")
    parser.add_argument("--out", default="metrics.json")
    parser.add_argument("--seed", action="store_true",
                        help="Insert a demo run before aggregating.")
    args = parser.parse_args()

    conn = sqlite3.connect(args.db)
    try:
        conn.executescript(SCHEMA)
        if args.seed or conn.execute(
            "SELECT COUNT(*) FROM check_results"
        ).fetchone()[0] == 0:
            seed(conn)
        metrics = aggregate(load_latest(conn))
    finally:
        conn.close()

    with open(args.out, "w", encoding="utf-8") as fh:
        json.dump(metrics, fh, indent=2)
    print(json.dumps(metrics, indent=2))

if __name__ == "__main__":
    main()

# Validation and pipeline integration

Run it standalone against the seeded demo, then confirm the arithmetic:

python aggregate_metrics.py --db :memory: --seed --out metrics.json

# valid_crs passes on 2 of 3 datasets -> expect 66.7
python -c "import json; m=json.load(open('metrics.json')); print(m['pass_rate_by_check_pct']['valid_crs'])"

A pytest fixture pins the aggregation logic so a refactor cannot silently change a published number:

import pandas as pd
from aggregate_metrics import aggregate

def test_overall_requires_all_checks():
    df = pd.DataFrame([
        {"dataset_id": "a", "collection": "x", "check_name": "crs", "passed": 1, "run_at": "t"},
        {"dataset_id": "a", "collection": "x", "check_name": "lic", "passed": 1, "run_at": "t"},
        {"dataset_id": "b", "collection": "x", "check_name": "crs", "passed": 1, "run_at": "t"},
        {"dataset_id": "b", "collection": "x", "check_name": "lic", "passed": 0, "run_at": "t"},
    ])
    m = aggregate(df)
    assert m["overall_compliant_pct"] == 50.0
    assert m["non_compliant_datasets"] == ["b"]

Wire the summary into a scheduled job so metrics refresh on every catalog scan, using the same enforcement discipline as CI/CD validation and policy enforcement for spatial data:

# .github/workflows/metrics.yml
name: Compliance metrics
on:
  schedule:
    - cron: '0 5 * * *'
jobs:
  metrics:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install "pandas>=2.1"
      - run: python aggregate_metrics.py --db compliance.db --out metrics.json
      - uses: actions/upload-artifact@v4
        with:
          name: compliance-metrics
          path: metrics.json

# Long-term compliance best practices

• Aggregate a stored table, never live scans. Compute checks once, persist them, and aggregate from the store so the same numbers reproduce exactly and history is preserved.
• Define “compliant” as passing every check. A row-wise AND (wide.all(axis=1)) prevents a dataset with one silent failure from being counted as compliant on the strength of its other passing checks.
• Report the denominator alongside every rate. A pass rate is meaningless without the dataset count; a shrinking inventory can flatter a metric that is actually flat.
• Group by run_at to expose trends. The same aggregation over multiple runs is a time series — the artifact that proves a remediation sprint worked.
• Keep the non-compliant list in the output. A headline percentage tells leadership the state; the explicit list of failing dataset_ids tells engineers what to fix.
• Serialize to JSON, not a rendered view. Emit structured metrics so both the dashboard and a formal report can consume one canonical source.

# Handling Partial and Failed Runs

The arithmetic in an aggregation is straightforward. What makes production aggregation hard is that runs are not clean: a check crashes on one dataset, a worker times out, a run covers half the catalogue before the pipeline is cancelled. How the aggregation treats those cases determines whether its output is trustworthy.

Distinguish “failed the check” from “the check failed”. These are opposite findings and both arrive as an absent pass. A dataset that failed validation is a compliance problem; a dataset whose validation crashed is an engineering problem, and counting it as non-compliant sends the wrong team after it. The results schema needs three states — pass, fail, error — and the aggregation needs to report the third separately rather than folding it into the second.

Treat a partial run as partial. If a run covers 300 of 480 datasets, its coverage figure is 62% of the catalogue, not 100% of what it happened to see. Recording the intended scope at the start of the run, rather than inferring it from the results afterwards, is what makes this detectable. Without it, a run that dies halfway through reports excellent numbers over the subset it managed to reach.

Do not carry forward silently. Filling a missing result with the previous run’s value produces a dashboard that cannot show a regression, because the last known good value persists indefinitely for any dataset whose check stopped running. If carrying forward is necessary — and for slow checks it sometimes is — the age of each result should be visible, and results past an agreed age should be treated as absent rather than as current.

Make the run itself a row. Start time, end time, intended scope, completed count, error count. Most questions about a suspicious metric are answered from that row alone, and it costs one insert per run. Aggregations that store only results, with no record of the runs that produced them, force every such question to become an investigation of the underlying data.

# Related

• Compliance Dashboards for Spatial Catalogs — the parent guide that renders and gates these metrics
• Spatial Data Audit Reporting & Compliance Governance — the governance program these numbers report into
• Automated Compliance Report Generation — turn the aggregated metrics into a formal audit deliverable
• Metadata Schema Validation & Linting — the source of the per-dataset schema and license checks