Auditing a Conflation Run Before Upload Jump to heading#

from __future__ import annotations

import logging
import random
from dataclasses import dataclass

import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.conflate.audit")

SLICES = ["density_band", "feature_class", "region"]
EXAMPLES_PER_BAND = 4
SEED = 20260917


@dataclass(frozen=True)
class Rules:
    match_floor: float
    review_floor: float
    min_gap: float
    cardinality: str
    precedence: str

    def as_sentences(self) -> list[str]:
        return [
            f"A pair is a confident match when its combined score is at least "
            f"{self.match_floor:.2f} AND it beats the next candidate by at "
            f"least {self.min_gap:.2f}.",
            f"A pair scoring between {self.review_floor:.2f} and "
            f"{self.match_floor:.2f}, or beating its runner-up by less than "
            f"{self.min_gap:.2f}, goes to human review.",
            f"Below {self.review_floor:.2f} the record is recorded as having no "
            f"match, which is an expected outcome rather than a failure.",
            f"Cardinality: {self.cardinality}.",
            f"Precedence when both sources have a value: {self.precedence}.",
        ]


def sliced_counts(pairs: pd.DataFrame) -> pd.DataFrame:
    """Counts by outcome, for the whole run and for every slice."""
    rows = []
    overall = pairs["outcome"].value_counts()
    rows.append({"slice": "ALL", "value": "ALL", **overall.to_dict()})
    for column in SLICES:
        if column not in pairs.columns:
            logger.warning("slice column %r absent; the pack will be weaker", column)
            continue
        for value, group in pairs.groupby(column):
            rows.append({"slice": column, "value": str(value),
                         **group["outcome"].value_counts().to_dict()})
    frame = pd.DataFrame(rows).fillna(0)
    numeric = [c for c in frame.columns if c not in {"slice", "value"}]
    frame[numeric] = frame[numeric].astype(int)
    frame["match_rate"] = frame.get("match", 0) / frame[numeric].sum(axis=1)
    return frame


def threshold_examples(pairs: pd.DataFrame, rules: Rules) -> pd.DataFrame:
    """Examples across the score range, weighted towards the boundary.

    Showing only high-scoring pairs proves nothing: a reviewer needs to see
    what a marginal decision looks like, because that is where errors live.
    """
    rng = random.Random(SEED)
    bands = {
        "well above": pairs[pairs["score"] >= rules.match_floor + 0.10],
        "just above": pairs[(pairs["score"] >= rules.match_floor)
                            & (pairs["score"] < rules.match_floor + 0.03)],
        "just below": pairs[(pairs["score"] < rules.match_floor)
                            & (pairs["score"] >= rules.match_floor - 0.03)],
        "small gap": pairs[(pairs["score"] >= rules.match_floor)
                           & (pairs["gap"] < rules.min_gap)],
        "no match": pairs[pairs["score"] < rules.review_floor],
    }
    chosen = []
    for band, group in bands.items():
        if group.empty:
            logger.warning("no examples in band %r", band)
            continue
        take = min(EXAMPLES_PER_BAND, len(group))
        picks = rng.sample(list(group.index), take)
        sample = group.loc[picks].copy()
        sample.insert(0, "band", band)
        chosen.append(sample)
    return pd.concat(chosen) if chosen else pd.DataFrame()


def limitations(pairs: pd.DataFrame, counts: pd.DataFrame) -> list[str]:
    """Derive the weak spots from the data rather than relying on memory."""
    notes: list[str] = []
    worst = counts[counts["slice"] != "ALL"].nsmallest(1, "match_rate")
    if not worst.empty:
        row = worst.iloc[0]
        notes.append(f"Lowest match rate is {row['match_rate']:.0%} in "
                     f"{row['slice']}={row['value']}; treat results there as weaker.")
    thin = pairs[pairs["evidence_weight"] < 0.6]
    if len(thin):
        notes.append(f"{len(thin)} pair(s) ({len(thin)/len(pairs):.1%}) were decided "
                     f"on less than 60% of the available evidence.")
    reviewed = (pairs["outcome"] == "review").sum()
    notes.append(f"{reviewed} pair(s) require human review before use; no "
                 f"automated decision was made for them.")
    return notes


def build_pack(pairs: pd.DataFrame, rules: Rules, path: str) -> None:
    counts = sliced_counts(pairs)
    examples = threshold_examples(pairs, rules)
    notes = limitations(pairs, counts)

    with open(path, "w", encoding="utf-8") as fh:
        fh.write("# Conflation run audit\n\n## Counts\n\n")
        fh.write(counts.to_markdown(index=False))
        fh.write("\n\n## Decision rules\n\n")
        for sentence in rules.as_sentences():
            fh.write(f"- {sentence}\n")
        fh.write("\n## Examples\n\n")
        fh.write(examples.to_markdown(index=False))
        fh.write("\n\n## Known limitations\n\n")
        for note in notes:
            fh.write(f"- {note}\n")
    logger.info("wrote evidence pack to %s", path)


if __name__ == "__main__":
    logger.info("publish the pack where every changeset comment can link to it")