"""Pure retrieval metrics with an explicit, bounded label contract. The functions score supplied judgments. They do not establish that the corpus, labels, eligibility rules, or evaluation cohort are representative. """ from math import isfinite, log2 from typing import Sequence, Union Numeric = Union[int, float] MAX_LABELS = 10_000 MAX_RELEVANCE = 64.0 def _labels(relevances: Sequence[Numeric]) -> tuple[float, ...]: if type(relevances) not in (list, tuple): raise ValueError("relevances must be a concrete list or tuple") if len(relevances) > MAX_LABELS: raise ValueError("too many relevance labels") normalized = [] for value in relevances: if type(value) not in (int, float): raise ValueError("relevance values must be exact numbers") numeric = float(value) if not isfinite(numeric): raise ValueError("relevance values must be finite") if numeric < 0 or numeric > MAX_RELEVANCE: raise ValueError("relevance values must be within the declared bounds") normalized.append(numeric) return tuple(normalized) def _cutoff(k: int) -> int: if type(k) is not int or k <= 0 or k > MAX_LABELS: raise ValueError("k must be a bounded positive integer") return k def recall_at_k( relevances: Sequence[Union[int, float]], k: int, total_relevant: int, ) -> float: labels = _labels(relevances) cutoff = _cutoff(k) if type(total_relevant) is not int or total_relevant <= 0: raise ValueError("total_relevant must be a positive integer") if total_relevant > MAX_LABELS: raise ValueError("total_relevant exceeds the declared bound") retrieved = sum(value > 0 for value in labels[:cutoff]) if retrieved > total_relevant: raise ValueError("retrieved labels exceed total_relevant") return retrieved / total_relevant def reciprocal_rank(relevances: Sequence[Union[int, float]]) -> float: for rank, value in enumerate(_labels(relevances), start=1): if value > 0: return 1 / rank return 0.0 def dcg_at_k(relevances: Sequence[Union[int, float]], k: int) -> float: labels = _labels(relevances) cutoff = _cutoff(k) return sum( (2**relevance - 1) / log2(rank + 1) for rank, relevance in enumerate(labels[:cutoff], start=1) ) def ndcg_at_k(relevances: Sequence[Union[int, float]], k: int) -> float: labels = _labels(relevances) cutoff = _cutoff(k) ideal = dcg_at_k(sorted(labels, reverse=True), cutoff) return 0.0 if ideal == 0 else dcg_at_k(labels, cutoff) / ideal def main() -> None: labels = [0, 2, 1, 0] print(f"{recall_at_k(labels, 3, 2):.2f}") print(f"{reciprocal_rank(labels):.2f}") print(f"{ndcg_at_k(labels, 3):.3f}") if __name__ == "__main__": main()