Finding Statistical Outliers in OSM Tag Values Jump to heading#

from __future__ import annotations

import difflib
import logging
import re
import statistics
from collections import Counter, defaultdict
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass

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

MIN_SAMPLE = 200          # below this a distribution means little
NUMERIC_K = 6.0           # generous: this is a review queue, not a gate
MIN_RARE = 50             # a value below this count may be a typo
TYPO_RATIO = 0.001        # ...if it is this much rarer than its neighbour

UNIT_SUFFIX = re.compile(r"^\s*(-?\d+(?:\.\d+)?)\s*([a-z']*)\s*$", re.I)
# Plausible ranges in the tag's DOCUMENTED unit, used to name the mistake.
PLAUSIBLE = {"height": (0.5, 900.0), "width": (0.3, 100.0),
             "maxspeed": (5.0, 200.0), "capacity": (1.0, 20_000.0)}
UNIT_FACTORS = {"cm": 0.01, "mm": 0.001, "ft": 0.3048, "'": 0.3048,
                "km": 1000.0, "mph": 1.609344}


@dataclass(frozen=True)
class Outlier:
    osm_id: int
    key: str
    value: str
    category: str
    reason: str
    suspected: str | None = None


def parse_number(raw: str) -> tuple[float, str] | None:
    match = UNIT_SUFFIX.match(raw)
    if not match:
        return None
    try:
        return float(match.group(1)), match.group(2).lower()
    except ValueError:
        return None


def suspect_unit(key: str, value: float) -> str | None:
    """An implausible metre value that is plausible in another unit is not a
    random outlier; it is a specific, nameable mistake."""
    low, high = PLAUSIBLE.get(key, (float("-inf"), float("inf")))
    if low <= value <= high:
        return None
    for unit, factor in UNIT_FACTORS.items():
        if low <= value * factor <= high:
            return f"plausible if the value is {unit}: {value * factor:,.2f}"
    return None


def robust_spread(values: Sequence[float]) -> float:
    centre = statistics.median(values)
    return statistics.median([abs(v - centre) for v in values]) * 1.4826


def numeric_outliers(features: Iterable[tuple[int, Mapping[str, str]]],
                     key: str, category_key: str) -> list[Outlier]:
    """Group by category first: a house and a skyscraper are different
    distributions, and pooling them hides both ends."""
    groups: dict[str, list[tuple[int, float, str]]] = defaultdict(list)
    for osm_id, tags in features:
        raw = tags.get(key)
        if raw is None:
            continue
        parsed = parse_number(raw)
        if parsed is None:
            continue
        value, unit = parsed
        if unit in UNIT_FACTORS:            # explicit units are not errors
            value *= UNIT_FACTORS[unit]
        groups[tags.get(category_key, "unknown")].append((osm_id, value, raw))

    found: list[Outlier] = []
    for category, rows in groups.items():
        if len(rows) < MIN_SAMPLE:
            continue
        values = [v for _, v, _ in rows]
        centre = statistics.median(values)
        spread = robust_spread(values) or max(abs(centre) * 0.05, 0.1)
        low, high = centre - NUMERIC_K * spread, centre + NUMERIC_K * spread
        for osm_id, value, raw in rows:
            if low <= value <= high:
                continue
            found.append(Outlier(
                osm_id, key, raw, category,
                f"{value:,.2f} against a {category} median of {centre:,.2f} "
                f"(band {low:,.2f} to {high:,.2f})",
                suspect_unit(key, value)))
    logger.info("%s: %d numeric outlier(s)", key, len(found))
    return found


def categorical_outliers(features: Iterable[tuple[int, Mapping[str, str]]],
                         key: str) -> list[Outlier]:
    """Rarity alone cannot separate a typo from a genuinely rare value.
    Rarity plus closeness to a common value usually can."""
    counts = Counter(tags[key] for _, tags in features if key in tags)
    common = [v for v, n in counts.items() if n >= MIN_SAMPLE]
    total = sum(counts.values()) or 1

    suspicious: dict[str, str] = {}
    for value, n in counts.items():
        if n >= MIN_RARE:
            continue
        near = difflib.get_close_matches(value, common, n=1, cutoff=0.86)
        if near and n / max(counts[near[0]], 1) < TYPO_RATIO:
            suspicious[value] = near[0]

    found = [Outlier(osm_id, key, tags[key], "-",
                     f"{counts[tags[key]]} use(s) against "
                     f"{counts[suspicious[tags[key]]]:,} for a near-identical value",
                     f"probably {suspicious[tags[key]]!r}")
             for osm_id, tags in features
             if tags.get(key) in suspicious]
    logger.info("%s: %d suspected typo(s) across %d distinct value(s), "
                "%.4f%% of uses", key, len(found), len(suspicious),
                100 * sum(counts[v] for v in suspicious) / total)
    return found


if __name__ == "__main__":
    logger.info("robust bands per category; rarity plus edit distance for text")