ranklens.metrics

ranklens.metrics

Ranking metrics, the registry of built-in metrics and evaluation of a run.

What has which shape, and why there is no base class to inherit:

• a metric is any callable (ranked, judgements, k) -> float (ranklens.core.Metric, a Protocol). A plain function is a metric;
• metrics with parameters are frozen dataclasses (NDCG, AP, RR, ERR, RBP): the parameters are then visible, comparable and validated once, at construction;
• dcg, gain_function and friends in base are functions: they are pieces of formulas, not metrics, and are not registered;
• rbo is a function as well — it compares two rankings, not a ranking with judgements.

ERR dataclass

ERR@k = sum over ranks r <= k of (1/r) * R_r * prod_{i<r} (1 - R_i).

R = (2rel - 1) / 2max_rel is the probability that the user is satisfied by the document and stops (Chapelle et al., 2009). max_rel is the top grade of the relevance scale, not of the query; the default 4 is that of TREC gdeval. Relevance above max_rel counts as max_rel, non-positive as 0.

Source code in src/ranklens/metrics/err_rbp.py

@dataclass(frozen=True, slots=True)
class ERR:
    """ERR@k = sum over ranks r <= k of (1/r) * R_r * prod_{i<r} (1 - R_i).

    R = (2**rel - 1) / 2**max_rel is the probability that the user is satisfied by
    the document and stops (Chapelle et al., 2009). ``max_rel`` is the top grade of the
    relevance **scale**, not of the query; the default 4 is that of TREC gdeval.
    Relevance above ``max_rel`` counts as ``max_rel``, non-positive as 0.
    """

    max_rel: float = 4

    def __post_init__(self) -> None:
        positive("max_rel", self.max_rel)

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        scale = 2.0**self.max_rel
        total, still_looking = 0.0, 1.0
        for rank, doc in enumerate(cutoff(ranked, k), start=1):
            rel = min(max(judgements.get(doc, 0.0), 0.0), self.max_rel)
            satisfied = (2.0**rel - 1.0) / scale
            total += still_looking * satisfied / rank
            still_looking *= 1.0 - satisfied
        return total

    def batch(self, batch: Batch, k: int | None) -> Vector:
        rel = np.clip(batch.relevance[:, :k], 0.0, self.max_rel)
        satisfied = (np.exp2(rel) - 1.0) / 2.0**self.max_rel
        # probability of reaching rank r: product of (1 - R_i) over the ranks before it
        reached = np.cumprod(1.0 - satisfied, axis=1)
        reached = np.hstack([np.ones((len(rel), 1)), reached[:, :-1]])
        ranks = np.arange(1, rel.shape[1] + 1)
        result: Vector = (reached * satisfied / ranks).sum(axis=1)
        return result

RBP dataclass

RBP@k = (1 - p) * sum over ranks r <= k of rel_r * p**(r - 1).

The user moves to the next document with probability p (Moffat & Zobel, 2008); p = 0.8 by default, 0.5 is impatient, 0.95 patient. Relevance is binary (rel >= 1) unless max_rel is given: then it is graded as min(rel, max_rel) / max_rel, which keeps RBP within [0, 1]. RBP@k is the lower bound: the unseen tail is ignored.

Source code in src/ranklens/metrics/err_rbp.py

@dataclass(frozen=True, slots=True)
class RBP:
    """RBP@k = (1 - p) * sum over ranks r <= k of rel_r * p**(r - 1).

    The user moves to the next document with probability ``p`` (Moffat & Zobel, 2008);
    p = 0.8 by default, 0.5 is impatient, 0.95 patient. Relevance is binary (rel >= 1)
    unless ``max_rel`` is given: then it is graded as min(rel, max_rel) / max_rel,
    which keeps RBP within [0, 1]. RBP@k is the lower bound: the unseen tail is ignored.
    """

    p: float = 0.8
    max_rel: float | None = None

    def __post_init__(self) -> None:
        probability("p", self.p)
        if self.max_rel is not None:
            positive("max_rel", self.max_rel)

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        total = 0.0
        for rank, doc in enumerate(cutoff(ranked, k), start=1):
            total += self._utility(judgements.get(doc, 0.0)) * self.p ** (rank - 1)
        return (1.0 - self.p) * total

    def batch(self, batch: Batch, k: int | None) -> Vector:
        utility = self._utilities(batch.relevance[:, :k])
        weights = self.p ** np.arange(utility.shape[1])
        result: Vector = (1.0 - self.p) * (utility @ weights)
        return result

    def _utilities(self, relevance: Matrix) -> Matrix:
        if self.max_rel is None:
            return (relevance >= 1).astype(np.float64)
        return np.clip(relevance, 0.0, self.max_rel) / self.max_rel

    def _utility(self, rel: float) -> float:
        if self.max_rel is None:
            return 1.0 if rel >= 1 else 0.0
        return min(max(rel, 0.0), self.max_rel) / self.max_rel

Judged dataclass

Share of the top k documents that have a judgement (of any grade, 0 included).

The denominator is the number of documents actually in the top k — a ranking shorter than k is not penalized for documents it did not return. An empty ranking gives 0.

Source code in src/ranklens/metrics/judged.py

@dataclass(frozen=True, slots=True)
class Judged:
    """Share of the top ``k`` documents that have a judgement (of any grade, 0 included).

    The denominator is the number of documents actually in the top ``k`` — a ranking
    shorter than ``k`` is not penalized for documents it did not return. An empty
    ranking gives 0.
    """

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        top = cutoff(ranked, k)
        if not top:
            return 0.0
        return sum(doc in judgements for doc in top) / len(top)

AP dataclass

AP@k = (sum of P@i over relevant documents at ranks i <= k) / R.

A document is relevant when its relevance is >= rel. R counts all relevant judged documents of the query, as trec_eval does, so AP@k < 1 whenever R > k. The mean over queries is MAP. No relevant documents: 0.

Source code in src/ranklens/metrics/map_mrr.py

@dataclass(frozen=True, slots=True)
class AP:
    """AP@k = (sum of P@i over relevant documents at ranks i <= k) / R.

    A document is relevant when its relevance is >= ``rel``. R counts **all** relevant
    judged documents of the query, as trec_eval does, so AP@k < 1 whenever R > k.
    The mean over queries is MAP. No relevant documents: 0.
    """

    rel: float = 1

    def __post_init__(self) -> None:
        positive("rel", self.rel)

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        n_relevant = sum(r >= self.rel for r in judgements.values())
        if n_relevant == 0:
            return 0.0
        hits, precision_sum = 0, 0.0
        for rank, doc in enumerate(cutoff(ranked, k), start=1):
            if judgements.get(doc, 0.0) >= self.rel:
                hits += 1
                precision_sum += hits / rank
        return precision_sum / n_relevant

    def batch(self, batch: Batch, k: int | None) -> Vector:
        relevant = batch.relevance[:, :k] >= self.rel
        ranks = np.arange(1, relevant.shape[1] + 1)
        precision = np.cumsum(relevant, axis=1) / ranks
        n_relevant = (batch.ideal >= self.rel).sum(axis=1)
        total = (precision * relevant).sum(axis=1)
        result: Vector = np.divide(
            total, n_relevant, out=np.zeros_like(total), where=n_relevant > 0
        )
        return result

RR dataclass

RR@k = 1 / rank of the first relevant document within k, else 0.

A document is relevant when its relevance is >= rel. The mean over queries is MRR.

Source code in src/ranklens/metrics/map_mrr.py

@dataclass(frozen=True, slots=True)
class RR:
    """RR@k = 1 / rank of the first relevant document within k, else 0.

    A document is relevant when its relevance is >= ``rel``. The mean over queries is MRR.
    """

    rel: float = 1

    def __post_init__(self) -> None:
        positive("rel", self.rel)

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        for rank, doc in enumerate(cutoff(ranked, k), start=1):
            if judgements.get(doc, 0.0) >= self.rel:
                return 1.0 / rank
        return 0.0

    def batch(self, batch: Batch, k: int | None) -> Vector:
        relevant = batch.relevance[:, :k] >= self.rel
        first = relevant.argmax(axis=1)  # 0 when nothing is relevant, masked below
        return np.where(relevant.any(axis=1), 1.0 / (first + 1), 0.0)

NDCG dataclass

NDCG@k = DCG@k / IDCG@k.

IDCG is the DCG of the ideal order of all judged documents of the query, not only the retrieved ones, cut at k. No relevant documents: 0 (as trec_eval). gain="linear" matches trec_eval; gain="exp" matches LTR libraries.

Source code in src/ranklens/metrics/ndcg.py

@dataclass(frozen=True, slots=True)
class NDCG:
    """NDCG@k = DCG@k / IDCG@k.

    IDCG is the DCG of the ideal order of **all** judged documents of the query,
    not only the retrieved ones, cut at k. No relevant documents: 0 (as trec_eval).
    ``gain="linear"`` matches trec_eval; ``gain="exp"`` matches LTR libraries.
    """

    gain: str = "linear"

    def __post_init__(self) -> None:
        gain_function(self.gain)  # fail at construction, not on the first query

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float:
        gain = gain_function(self.gain)
        ideal = sorted((gain(r) for r in judgements.values()), reverse=True)
        best = dcg(ideal[:k] if k is not None else ideal)
        if best == 0:
            return 0.0
        return dcg(gain(judgements.get(doc, 0.0)) for doc in cutoff(ranked, k)) / best

    def batch(self, batch: Batch, k: int | None) -> Vector:
        found = _gains(self.gain, batch.relevance[:, :k])
        ideal = _gains(self.gain, batch.ideal[:, :k])
        best = ideal @ discounts(ideal.shape[1])
        actual = found @ discounts(found.shape[1])
        return np.divide(actual, best, out=np.zeros_like(actual), where=best > 0)

resolve

resolve(spec: str | MetricSpec) -> BoundMetric

Resolve a spec such as ndcg@10 against the built-in metrics.

Source code in src/ranklens/metrics/builtin.py

def resolve(spec: str | MetricSpec) -> BoundMetric:
    """Resolve a spec such as ``ndcg@10`` against the built-in metrics."""
    return registry.resolve(spec)

evaluate

evaluate(
    runs: Iterable[RankedList],
    qrels: Qrels,
    metrics: Sequence[str | BoundMetric],
    *,
    registry: Registry = builtin_registry,
    strict: bool = False,
    batch_size: int = 1024,
) -> Evaluation

Compute every metric for every query of runs that has judgements.

runs is consumed once, so a streaming reader works. A query of the run without judgements is skipped and counted (strict: MissingQrelsError). Metric specs are resolved with registry (built-in metrics by default) before the first query is read, so a typo fails fast.

Queries are scored in batches of batch_size: metrics implementing BatchMetric share one relevance matrix per batch, the others are called query by query. Memory holds one batch, not the run; a batch is cut earlier when its matrices would exceed BATCH_CELLS.

Source code in src/ranklens/metrics/evaluation.py

def evaluate(
    runs: Iterable[RankedList],
    qrels: Qrels,
    metrics: Sequence[str | BoundMetric],
    *,
    registry: Registry = builtin_registry,
    strict: bool = False,
    batch_size: int = 1024,
) -> Evaluation:
    """Compute every metric for every query of ``runs`` that has judgements.

    ``runs`` is consumed once, so a streaming reader works. A query of the run without
    judgements is skipped and counted (``strict``: `MissingQrelsError`).
    Metric specs are resolved with ``registry`` (built-in metrics by default) before
    the first query is read, so a typo fails fast.

    Queries are scored in batches of ``batch_size``: metrics implementing
    `BatchMetric` share one relevance matrix per
    batch, the others are called query by query. Memory holds one batch, not the run;
    a batch is cut earlier when its matrices would exceed `BATCH_CELLS`.
    """
    if batch_size < 1:
        raise ValueError(f"batch_size must be >= 1, got {batch_size}")
    bound = [registry.resolve(m) if isinstance(m, str) else m for m in metrics]
    values: list[dict[QueryId, float]] = [{} for _ in bound]
    segments: dict[QueryId, SegmentKey] = {}
    seen: set[QueryId] = set()
    n_unjudged = n_without_relevant = 0
    pending: list[_Item] = []
    widest = 0  # the widest row of the pending batch: its matrices are len(pending) x widest

    for ranked in runs:
        seen.add(ranked.query_id)
        judgements = qrels.get(ranked.query_id)
        if judgements is None:
            if strict:
                raise MissingQrelsError(ranked.query_id)
            n_unjudged += 1
            continue
        if not any(rel > 0 for rel in judgements.values()):
            n_without_relevant += 1
        if ranked.segments:
            segments[ranked.query_id] = ranked.segments
        width = max(len(ranked.docs), len(judgements))
        if pending and (len(pending) + 1) * max(widest, width) > BATCH_CELLS:
            _score(pending, bound, values)
            pending.clear()
            widest = 0
        pending.append((ranked.query_id, ranked.docs, judgements))
        widest = max(widest, width)
        if len(pending) == batch_size:
            _score(pending, bound, values)
            pending.clear()
            widest = 0
    _score(pending, bound, values)

    return Evaluation(
        metrics=tuple(
            MetricResult(metric.label, MappingProxyType(per_query))
            for metric, per_query in zip(bound, values, strict=True)
        ),
        n_queries=len(seen) - n_unjudged,
        n_without_relevant=n_without_relevant,
        n_unjudged=n_unjudged,
        n_not_retrieved=sum(1 for query_id in qrels if query_id not in seen),
        segments=MappingProxyType(segments),
    )