Computing a Composite Data Quality Score per Reading #

A quality score earns its place only if it predicts error. Anything else is a number that looks authoritative, gets embedded in a filter, and silently shapes every published figure without ever having been checked. This guide implements the composite score defined in uncertainty quantification and data quality scoring as a single self-contained function, and — more importantly — the validation that shows it works.

What the Score Is For #

The score answers one question: how much should a consumer trust this row relative to the others in the same table? It is deliberately dimensionless and deliberately monotone, because its job is ranking and filtering rather than physical interpretation. The physical claim belongs to the uncertainty column.

That framing settles most design arguments. The score does not need to be calibrated as a probability. It does need to be ordered correctly — a 0.9 reading must be more reliable than a 0.6 one, across the whole fleet and across seasons — and it needs to be explainable, because the first question anyone asks about an excluded station is why.

Four components carry nearly all the signal in practice. Flag state is the strongest and is mostly a veto. Calibration freshness decays predictably and catches the slow failure that nothing else notices. Neighbour agreement catches a sensor that has gone wrong between calibrations. Completeness catches an interval built from too few samples to mean anything.

Production-Ready Implementation #

# python 3.11 · numpy==1.26.4 · pandas==2.2.2
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np
import pandas as pd

# Flag vocabulary shared across the pipeline (CF-convention aligned)
FLAG_TRUST = {1: 1.0, 2: 0.6, 3: 0.4, 4: 0.0, 8: 0.3, 9: 0.0}
DISQUALIFYING = {4, 9}


@dataclass(frozen=True)
class ScoreWeights:
    flag: float = 0.40
    calibration: float = 0.25
    agreement: float = 0.20
    completeness: float = 0.15

    def as_dict(self) -> dict[str, float]:
        return {"flag": self.flag, "calibration": self.calibration,
                "agreement": self.agreement, "completeness": self.completeness}


@dataclass
class QualityScorer:
    """Composite 0..1 quality score with explainable components.

    weights sum is normalised internally, so adjusting one weight does not
    require rebalancing the others — a common source of drift between
    environments where the weights are configuration rather than code.
    """

    weights: ScoreWeights = field(default_factory=ScoreWeights)
    calibration_interval_days: float = 180.0

    def components(
        self,
        *,
        qc_flag: int,
        days_since_calibration: float,
        value: float,
        neighbour_values: list[float],
        neighbour_sigma: float,
        n_samples: int,
        expected_samples: int,
    ) -> dict[str, float]:
        return {
            "flag": FLAG_TRUST.get(qc_flag, 0.5),
            "calibration": float(np.clip(
                1.0 - days_since_calibration / (2 * self.calibration_interval_days), 0.0, 1.0)),
            "agreement": self._agreement(value, neighbour_values, neighbour_sigma),
            "completeness": float(np.clip(n_samples / max(expected_samples, 1), 0.0, 1.0)),
        }

    @staticmethod
    def _agreement(value: float, neighbours: list[float], sigma: float) -> float:
        """Gaussian decay of the z-score against the neighbour median.

        With no neighbours the answer is 0.5 — 'no information', which must not
        be confused with 0.0 ('disagrees') or 1.0 ('agrees').
        """
        usable = [v for v in neighbours if v is not None and np.isfinite(v)]
        if not usable or sigma <= 0:
            return 0.5
        z = abs(value - float(np.median(usable))) / sigma
        return float(np.exp(-0.5 * z * z))

    def score(self, components: dict[str, float], qc_flag: int) -> float:
        if qc_flag in DISQUALIFYING:
            return 0.0                       # veto beats any weighting
        w = self.weights.as_dict()
        return float(np.clip(
            sum(w[k] * components[k] for k in w) / sum(w.values()), 0.0, 1.0))

Applied over a frame, with the components preserved for explanation:

def score_frame(df: pd.DataFrame, scorer: QualityScorer) -> pd.DataFrame:
    """Attach quality_score and quality_components to every row.

    Neighbour values are expected as a list column, prepared upstream by a
    spatial join — computing them per row here would be O(n^2).
    """
    rows = []
    for r in df.itertuples(index=False):
        comps = scorer.components(
            qc_flag=r.qc_flag,
            days_since_calibration=r.days_since_calibration,
            value=r.value,
            neighbour_values=list(r.neighbour_values or []),
            neighbour_sigma=r.neighbour_sigma,
            n_samples=r.n_samples,
            expected_samples=r.expected_samples,
        )
        rows.append({"quality_score": scorer.score(comps, r.qc_flag),
                     "quality_components": comps})
    return df.assign(**pd.DataFrame(rows, index=df.index))

Parameter Tuning Guide #

neighbour_sigma is the parameter people set carelessly. It should be the expected spatial variability between comparable sensors, measured from your own network, not the instrument’s uncertainty — the two differ by a large factor for PM2.5, where genuine street-level variation dwarfs sensor noise.

Verification and Testing #

The validation that matters is the monotonicity check against a reference.

# python 3.11 · pandas==2.2.2 · pytest==8.2.0
import pandas as pd


def score_predicts_error(scored: pd.DataFrame, reference_col: str = "reference_value") -> pd.DataFrame:
    """Mean absolute error by score decile — must fall as the score rises."""
    df = scored.dropna(subset=[reference_col]).copy()
    df["abs_error"] = (df["value"] - df[reference_col]).abs()
    df["decile"] = pd.qcut(df["quality_score"], 10, labels=False, duplicates="drop")
    return df.groupby("decile", as_index=False)["abs_error"].mean()


def test_error_falls_as_score_rises(colocation_frame):
    table = score_predicts_error(colocation_frame)
    top = table.loc[table["decile"] >= 7, "abs_error"].mean()
    bottom = table.loc[table["decile"] <= 2, "abs_error"].mean()
    assert top < 0.6 * bottom, f"score does not predict error: {table}"


def test_disqualifying_flag_vetoes_a_healthy_looking_reading():
    scorer = QualityScorer()
    comps = scorer.components(
        qc_flag=9, days_since_calibration=1, value=20.0,
        neighbour_values=[20.1, 19.8], neighbour_sigma=6.0,
        n_samples=60, expected_samples=60,
    )
    assert comps["calibration"] > 0.99          # every other dimension is perfect
    assert scorer.score(comps, qc_flag=9) == 0.0


def test_no_neighbours_scores_as_no_information():
    scorer = QualityScorer()
    comps = scorer.components(
        qc_flag=1, days_since_calibration=10, value=20.0,
        neighbour_values=[], neighbour_sigma=6.0,
        n_samples=60, expected_samples=60,
    )
    assert comps["agreement"] == 0.5            # not 0.0 and not 1.0

Run score_predicts_error quarterly against your co-location data and keep the table. It is the only evidence that the score means anything, and it is the first thing to check when someone proposes changing a weight.

Gotchas #

Scoring agreement against a single neighbour. With one comparison sensor, a disagreement is equally likely to mean the neighbour is broken. Require at least three usable neighbours before the agreement component carries its full weight, or fall back to 0.5.

Weights that do not sum to one, applied without normalising. Someone raises the flag weight to 0.6 to “make flags matter more” and every score in the table shifts upward, silently moving every filter threshold. Normalise by the weight sum, as above.

Using the score as an input to anomaly detection. The score already contains the anomaly flag, so feeding it back creates a loop where a flagged reading lowers the score which strengthens the flag. Keep the direction of dependency one-way.

Recomputing scores in place on every batch. The components depend on registry state that changes; overwriting yesterday’s score with today’s registry makes historical scores unreproducible. Store the registry version with the score and treat a recompute as a new derivation.

FAQ #

Related #

• Uncertainty Quantification and Data Quality Scoring — the scoring stage this function implements
• Propagating Measurement Uncertainty Through Aggregation — the physical-units companion to this dimensionless score
• Automating QC Flags for Missing Environmental Readings — the flag vocabulary the score’s dominant component reads