geointel/backend/app/services/reviewed_metrics_service.py

"""Detection metrics after an operator has adjudicated the evidence.

The review vocabulary already separates a model error from a reference gap,
because the product's position is that official footprints are not
automatically perfect ground truth. Until now those verdicts were only counted:
an operator who established that twelve of forty false positives are buildings
the reference simply lacks still saw a precision counting all forty against the
model — a number they had personally disproved.

Applying the verdicts gives an adjudicated score. Because part of the evidence
is usually still unreviewed, the honest form is an interval: pessimistic
assumes every unreviewed item is a model error, optimistic assumes none is. The
raw score stays reported alongside, so nothing is quietly improved.
"""

from __future__ import annotations

from typing import Any

# A verdict that the finding was not the model's fault. The detection (or the
# missing detection) was right; the reference or the matching rule was not.
EXONERATING_DECISIONS = frozenset({"reference_gap_or_change", "qa_alignment_mismatch"})

# Verdicts that confirm the finding, and verdicts that reach no conclusion.
# Both keep counting: being unable to judge is not evidence in the model's
# favour, and treating it as such is how a score drifts upward unearned.
CONFIRMING_DECISIONS = frozenset(
    {"confirmed_model_false_positive", "confirmed_model_false_negative"}
)
INCONCLUSIVE_DECISIONS = frozenset({"uncertain", "imagery_obscured_or_uncertain"})


def _score(matches: int, false_positives: int, false_negatives: int) -> dict[str, Any]:
    precision = matches / (matches + false_positives) if matches + false_positives > 0 else None
    recall = matches / (matches + false_negatives) if matches + false_negatives > 0 else None
    f1_score = None
    if precision is not None and recall is not None:
        f1_score = (2 * precision * recall / (precision + recall)) if precision + recall > 0 else 0.0
    return {
        "matches": matches,
        "false_positives": false_positives,
        "false_negatives": false_negatives,
        "precision": precision,
        "recall": recall,
        "f1_score": f1_score,
    }


class ReviewedMetricsService:
    @staticmethod
    def _adjudicate_role(total: int, decisions: dict[str, int]) -> tuple[int, int, int]:
        """Split a finding count into exonerated, confirmed and unreviewed.

        A decision the runtime does not recognise — one from a later release —
        counts as no judgement rather than as an exoneration.
        """

        exonerated = sum(count for name, count in decisions.items() if name in EXONERATING_DECISIONS)
        judged = sum(
            count
            for name, count in decisions.items()
            if name in EXONERATING_DECISIONS | CONFIRMING_DECISIONS | INCONCLUSIVE_DECISIONS
        )
        exonerated = min(exonerated, total)
        judged = min(judged, total)
        return exonerated, judged - exonerated, max(0, total - judged)

    @staticmethod
    def adjudicate(
        *,
        matches: int,
        false_positives: int,
        false_negatives: int,
        false_positive_decisions: dict[str, int],
        false_negative_decisions: dict[str, int],
    ) -> dict[str, Any]:
        """Apply operator verdicts to a quality check's counts."""

        fp_exonerated, _fp_confirmed, fp_unreviewed = ReviewedMetricsService._adjudicate_role(
            false_positives, false_positive_decisions
        )
        fn_exonerated, _fn_confirmed, fn_unreviewed = ReviewedMetricsService._adjudicate_role(
            false_negatives, false_negative_decisions
        )

        adjudicated_fp = false_positives - fp_exonerated
        adjudicated_fn = false_negatives - fn_exonerated

        # The interval covers what the unreviewed remainder could still turn
        # out to be, so a partly reviewed check never reads as a settled one.
        pessimistic = _score(matches, adjudicated_fp, adjudicated_fn)
        optimistic = _score(matches, adjudicated_fp - fp_unreviewed, adjudicated_fn - fn_unreviewed)

        return {
            "raw": _score(matches, false_positives, false_negatives),
            "adjudicated": pessimistic,
            "pessimistic": pessimistic,
            "optimistic": optimistic,
            "review_complete": fp_unreviewed == 0 and fn_unreviewed == 0,
            "false_positive_breakdown": {
                "total": false_positives,
                "exonerated": fp_exonerated,
                "confirmed_or_inconclusive": false_positives - fp_exonerated - fp_unreviewed,
                "unreviewed": fp_unreviewed,
            },
            "false_negative_breakdown": {
                "total": false_negatives,
                "exonerated": fn_exonerated,
                "confirmed_or_inconclusive": false_negatives - fn_exonerated - fn_unreviewed,
                "unreviewed": fn_unreviewed,
            },
        }