climate_ref.results.outliers

Outlier detection for scalar metric values.

This is a port of the logic previously living in the ref-app backend, and has been hoisted here to ensure all consumers use the same logic.

Outlier detection is read logic, not presentation policy. All consumers should see the same set of outliers. The outlier detection is performed at run-time (instead of a pre-computed flag) as the outlier configuration may depend on the use-case. We may also adopt other outlier detection algorithms in future.

The algorithm is source-id-aware Inter-Quartile Range (IQR). Within each group_by group, IQR bounds are computed on the per-source_id mean value (so each model is weighted equally regardless of ensemble size), then applied to individual values. The values that are outside factor * IQR are flagged as outliers. "Reference" values are never flagged. Non-finite values (NaN/inf) are always flagged.

AnnotatedScalar

A scalar ORM row paired with its outlier verdict.

Source code in packages/climate-ref/src/climate_ref/results/outliers.py

@attrs.frozen(kw_only=True)
class AnnotatedScalar:
    """A scalar ORM row paired with its outlier verdict."""

    value: ScalarMetricValue
    """The underlying scalar metric value row."""

    is_outlier: bool
    """Whether this value was flagged as an outlier."""

    verification_status: str
    """``"verified"`` or ``"unverified"``, mirroring ``is_outlier``."""

is_outlier instance-attribute

Whether this value was flagged as an outlier.

value instance-attribute

The underlying scalar metric value row.

verification_status instance-attribute

"verified" or "unverified", mirroring is_outlier.

OutlierPolicy

Configuration for scalar outlier detection.

The defaults reproduce the behaviour users currently see through the ref-app API (factor=10.0, source-id-aware IQR grouped by ("statistic", "metric")).

Source code in packages/climate-ref/src/climate_ref/results/outliers.py

@attrs.frozen(kw_only=True)
class OutlierPolicy:
    """
    Configuration for scalar outlier detection.

    The defaults reproduce the behaviour users currently see through the ref-app API
    (``factor=10.0``, source-id-aware IQR grouped by ``("statistic", "metric")``).
    """

    method: str = attrs.field(default="iqr", validator=attrs.validators.in_(_SUPPORTED_METHODS))
    """Detection method. ``"off"`` disables detection; ``"iqr"`` enables it."""

    factor: float = 10.0
    """Multiplier on the IQR to set the outlier bounds (``Q1 - factor*IQR``, ``Q3 + factor*IQR``)."""

    min_n: int = attrs.field(default=4, validator=attrs.validators.ge(2))
    """
    Minimum sample size required to run detection.

    On the source-id-aware path this counts distinct non-reference ``source_id`` with a finite mean.
    On the fallback path it counts finite values.
    Non-finite values (NaN/inf) are excluded from the count
    and are flagged unconditionally regardless of whether detection runs.
    """

    group_by: tuple[str, ...] = ("statistic", "metric")
    """CV dimensions to group by before computing bounds. Missing dimensions are ignored."""

    @property
    def enabled(self) -> bool:
        """Whether detection should run."""
        return self.method == "iqr"

enabled property

Whether detection should run.

factor = 10.0 class-attribute instance-attribute

Multiplier on the IQR to set the outlier bounds (Q1 - factor*IQR, Q3 + factor*IQR).

group_by = ('statistic', 'metric') class-attribute instance-attribute

CV dimensions to group by before computing bounds. Missing dimensions are ignored.

method = attrs.field(default='iqr', validator=(attrs.validators.in_(_SUPPORTED_METHODS))) class-attribute instance-attribute

Detection method. "off" disables detection; "iqr" enables it.

min_n = attrs.field(default=4, validator=(attrs.validators.ge(2))) class-attribute instance-attribute

Minimum sample size required to run detection.

On the source-id-aware path this counts distinct non-reference source_id with a finite mean. On the fallback path it counts finite values. Non-finite values (NaN/inf) are excluded from the count and are flagged unconditionally regardless of whether detection runs.

detect_scalar_outliers(scalar_values, policy)

Annotate scalar values with outlier verdicts.

Parameters:

Name	Type	Description	Default
scalar_values	Sequence[ScalarMetricValue]	The full (unpaginated) set of scalar rows for the query scope. Detection must run over the whole set so IQR bounds are globally consistent.	required
policy	OutlierPolicy	Detection configuration.	required

Returns:

Type	Description
tuple[list[AnnotatedScalar], int]	A tuple of (annotated values in input order, total number of outliers).

Source code in packages/climate-ref/src/climate_ref/results/outliers.py

def detect_scalar_outliers(
    scalar_values: Sequence[ScalarMetricValue],
    policy: OutlierPolicy,
) -> tuple[list[AnnotatedScalar], int]:
    """
    Annotate scalar values with outlier verdicts.

    Parameters
    ----------
    scalar_values
        The full (unpaginated) set of scalar rows for the query scope. Detection must run over
        the whole set so IQR bounds are globally consistent.
    policy
        Detection configuration.

    Returns
    -------
    :
        A tuple of (annotated values in input order, total number of outliers).
    """
    if not scalar_values:
        return [], 0

    if not policy.enabled:
        return (
            [
                AnnotatedScalar(value=sv, is_outlier=False, verification_status="verified")
                for sv in scalar_values
            ],
            0,
        )

    df = pd.DataFrame(
        [{"scalar_value": sv, "value": sv.value, **sv.dimensions, "id": sv.id} for sv in scalar_values]
    )
    group_by = [g for g in policy.group_by if g in df.columns]

    verdict_by_id: dict[int, bool] = {}
    groups = df.groupby(list(group_by)) if group_by else [(None, df)]
    for _, group in groups:
        values = group["value"]
        finite = values.map(_is_finite)
        if "source_id" in group.columns:
            # Distinct finite source_ids gate detection (see _iqr_bounds_by_source_id).
            bounds = _iqr_bounds_by_source_id(group, factor=policy.factor, min_n=policy.min_n)
            if bounds is not None:
                lower, upper = bounds
                out_of_bounds = (values < lower) | (values > upper)
                flags = finite & out_of_bounds & (group["source_id"] != "Reference")
            else:
                flags = pd.Series(False, index=group.index)
        else:
            # Finite value count gates detection (see _flag_outliers_iqr).
            flag_list = _flag_outliers_iqr(values.to_list(), factor=policy.factor, min_n=policy.min_n)
            flags = pd.Series(flag_list, index=group.index)

        verdicts = flags | ~finite  # non-finite values are always flagged
        for row_id, verdict in zip(group["id"], verdicts):
            verdict_by_id[row_id] = bool(verdict)

    annotated: list[AnnotatedScalar] = []
    total = 0
    for sv in scalar_values:
        is_outlier = verdict_by_id.get(sv.id, False)
        annotated.append(
            AnnotatedScalar(
                value=sv,
                is_outlier=is_outlier,
                verification_status="unverified" if is_outlier else "verified",
            )
        )
        total += int(is_outlier)
    return annotated, total