ranklens.core

ranklens.core

Project vocabulary: types and errors. Depends on the standard library only.

Qrels module-attribute

Qrels: TypeAlias = dict[QueryId, dict[DocId, float]]

Graded relevance judgements: qrels[query_id][doc_id] -> relevance.

SegmentKey module-attribute

SegmentKey: TypeAlias = tuple[str, ...]

Segment field values in a fixed order, e.g. ("mobile", "ru").

ConfigError

Bases: RankLensError

Invalid configuration: metric names, parameters, plugins.

Source code in src/ranklens/core/exceptions.py

class ConfigError(RankLensError):
    """Invalid configuration: metric names, parameters, plugins."""

DataError

Bases: RankLensError

Input data is malformed or violates an input contract.

Source code in src/ranklens/core/exceptions.py

class DataError(RankLensError):
    """Input data is malformed or violates an input contract."""

DegenerateWeightsWarning

Bases: RankLensWarning

A few importance weights dominate an off-policy estimate.

Source code in src/ranklens/core/exceptions.py

class DegenerateWeightsWarning(RankLensWarning):
    """A few importance weights dominate an off-policy estimate."""

DuplicateDocumentError

Bases: DataError

The same document appears twice in one ranked list.

Source code in src/ranklens/core/exceptions.py

class DuplicateDocumentError(DataError):
    """The same document appears twice in one ranked list."""

    def __init__(self, query_id: str, doc_id: str) -> None:
        super().__init__(query_id, doc_id)
        self.query_id = query_id
        self.doc_id = doc_id

    def __str__(self) -> str:
        return f"query {self.query_id!r}: document {self.doc_id!r} is ranked more than once"

DuplicateMetricError

Bases: ConfigError

A metric with this name is already registered.

Source code in src/ranklens/core/exceptions.py

class DuplicateMetricError(ConfigError):
    """A metric with this name is already registered."""

    def __init__(self, name: str) -> None:
        super().__init__(name)
        self.name = name

    def __str__(self) -> str:
        return (
            f"metric {self.name!r} is already registered; "
            "choose another name or check for a plugin that registers it twice"
        )

InsufficientSampleError

Bases: StatisticalError

Too few observations (queries, impressions) for the requested procedure.

Source code in src/ranklens/core/exceptions.py

class InsufficientSampleError(StatisticalError):
    """Too few observations (queries, impressions) for the requested procedure."""

    def __init__(self, n: int, required: int, unit: str = "queries") -> None:
        super().__init__(n, required, unit)
        self.n = n
        self.required = required
        self.unit = unit

    def __str__(self) -> str:
        return (
            f"got {self.n} {self.unit}, at least {self.required} are required; "
            "use the MDE calculator to estimate the sample size you need"
        )

MalformedRowError

Bases: DataError

A row cannot be parsed according to the schema.

Source code in src/ranklens/core/exceptions.py

class MalformedRowError(DataError):
    """A row cannot be parsed according to the schema."""

    def __init__(self, line_no: int, raw: str, reason: str, path: str | None = None) -> None:
        super().__init__(line_no, raw, reason, path)
        self.line_no = line_no
        self.raw = raw
        self.reason = reason
        self.path = path

    def __str__(self) -> str:
        return f"{_location(self.path, self.line_no)}: {self.reason}: {_preview(self.raw)}"

MetricNotFoundError

Bases: ConfigError

Requested metric is not registered.

Source code in src/ranklens/core/exceptions.py

class MetricNotFoundError(ConfigError):
    """Requested metric is not registered."""

    def __init__(self, name: str, available: Iterable[str]) -> None:
        names = tuple(sorted(available))
        super().__init__(name, names)
        self.name = name
        self.available = names

    def __str__(self) -> str:
        message = f"unknown metric {self.name!r}"
        close = difflib.get_close_matches(self.name, self.available, n=1)
        if close:
            message += f"; did you mean {close[0]!r}?"
        listing = ", ".join(self.available) if self.available else "none registered"
        return f"{message}{' ' if close else '. '}Available: {listing}"

MetricSpecError

Bases: ConfigError

A metric spec cannot be parsed or has invalid parameters.

Source code in src/ranklens/core/exceptions.py

class MetricSpecError(ConfigError):
    """A metric spec cannot be parsed or has invalid parameters."""

    def __init__(self, spec: str, reason: str) -> None:
        super().__init__(spec, reason)
        self.spec = spec
        self.reason = reason

    def __str__(self) -> str:
        return f"invalid metric spec {self.spec!r}: {self.reason}"

MissingColumnError

Bases: DataError

A required column is absent from the input.

Source code in src/ranklens/core/exceptions.py

class MissingColumnError(DataError):
    """A required column is absent from the input."""

    def __init__(self, column: str, available: Iterable[str], path: str | None = None) -> None:
        names = tuple(available)
        super().__init__(column, names, path)
        self.column = column
        self.available = names
        self.path = path

    def __str__(self) -> str:
        message = f"{self.path or 'input'}: required column {self.column!r} is missing"
        close = difflib.get_close_matches(self.column, self.available, n=1)
        if close:
            message += f"; did you mean {close[0]!r}?"
        listing = ", ".join(self.available) if self.available else "no columns found"
        return f"{message}{' ' if close else '. '}Available: {listing}"

MissingDependencyError

Bases: ConfigError

An optional dependency required by a feature is not installed.

Source code in src/ranklens/core/exceptions.py

class MissingDependencyError(ConfigError):
    """An optional dependency required by a feature is not installed."""

    def __init__(self, package: str, extra: str, feature: str) -> None:
        super().__init__(package, extra, feature)
        self.package = package
        self.extra = extra
        self.feature = feature

    def __str__(self) -> str:
        return (
            f"{self.feature} requires the optional dependency {self.package!r}; "
            f"install it with: pip install 'ranklens[{self.extra}]'"
        )

MissingQrelsError

Bases: DataError

A query has no relevance judgements.

Source code in src/ranklens/core/exceptions.py

class MissingQrelsError(DataError):
    """A query has no relevance judgements."""

    def __init__(self, query_id: str) -> None:
        super().__init__(query_id)
        self.query_id = query_id

    def __str__(self) -> str:
        return f"query {self.query_id!r} has no relevance judgements in qrels"

ModelError

Bases: RankLensError

A model cannot be used for scoring or explanation.

Source code in src/ranklens/core/exceptions.py

class ModelError(RankLensError):
    """A model cannot be used for scoring or explanation."""

PluginWarning

Bases: RankLensWarning

A third-party plugin could not be loaded and was skipped.

Source code in src/ranklens/core/exceptions.py

class PluginWarning(RankLensWarning):
    """A third-party plugin could not be loaded and was skipped."""

RankLensError

Bases: Exception

Base class for all errors raised by ranklens.

Source code in src/ranklens/core/exceptions.py

class RankLensError(Exception):
    """Base class for all errors raised by ranklens."""

RankLensWarning

Bases: UserWarning

Base class for all warnings emitted by ranklens.

Source code in src/ranklens/core/exceptions.py

class RankLensWarning(UserWarning):
    """Base class for all warnings emitted by ranklens."""

SkippedRowsWarning

Bases: RankLensWarning

Invalid input rows were skipped in non-strict mode.

Source code in src/ranklens/core/exceptions.py

class SkippedRowsWarning(RankLensWarning):
    """Invalid input rows were skipped in non-strict mode."""

SmallSampleWarning

Bases: RankLensWarning

The sample is large enough to compute a number, but too small to trust it.

Source code in src/ranklens/core/exceptions.py

class SmallSampleWarning(RankLensWarning):
    """The sample is large enough to compute a number, but too small to trust it."""

StatisticalError

Bases: RankLensError

A statistical procedure cannot produce a meaningful result.

Source code in src/ranklens/core/exceptions.py

class StatisticalError(RankLensError):
    """A statistical procedure cannot produce a meaningful result."""

UngroupedInputError

Bases: DataError

Rows of one query (or one logged impression) are not contiguous.

Source code in src/ranklens/core/exceptions.py

class UngroupedInputError(DataError):
    """Rows of one query (or one logged impression) are not contiguous."""

    def __init__(
        self, line_no: int, query_id: str, path: str | None = None, unit: str = "query"
    ) -> None:
        super().__init__(line_no, query_id, path, unit)
        self.line_no = line_no
        self.query_id = query_id
        self.path = path
        self.unit = unit

    def __str__(self) -> str:
        column = "query_id" if self.unit == "query" else f"{self.unit}_id"
        return (
            f"{_location(self.path, self.line_no)}: {self.unit} {self.query_id!r} appears "
            f"again after others; rows of each {self.unit} must be contiguous. "
            f"Sort the file by {column}."
        )

UnsortedInputError

Bases: DataError

Rows are not sorted by query_id.

Source code in src/ranklens/core/exceptions.py

class UnsortedInputError(DataError):
    """Rows are not sorted by query_id."""

    def __init__(
        self, line_no: int | None, prev_qid: str, curr_qid: str, path: str | None = None
    ) -> None:
        super().__init__(line_no, prev_qid, curr_qid, path)
        self.line_no = line_no
        self.prev_qid = prev_qid
        self.curr_qid = curr_qid
        self.path = path

    def __str__(self) -> str:
        return (
            f"{_location(self.path, self.line_no)}: input is not sorted by query_id "
            f"({self.curr_qid!r} comes after {self.prev_qid!r}). "
            "Sort the file by query_id as strings, where '10' comes before '9'."
        )

UnsupportedFormatError

Bases: ConfigError

The input format cannot be determined or is not supported.

Source code in src/ranklens/core/exceptions.py

class UnsupportedFormatError(ConfigError):
    """The input format cannot be determined or is not supported."""

    def __init__(self, path: str, supported: Iterable[str]) -> None:
        names = tuple(supported)
        super().__init__(path, names)
        self.path = path
        self.supported = names

    def __str__(self) -> str:
        return (
            f"cannot determine the format of {self.path!r}; "
            f"use one of the extensions {', '.join(self.supported)} (optionally .gz) "
            "or pass the format explicitly"
        )

UnsupportedModelError

Bases: ModelError

No adapter exists for this model type.

Source code in src/ranklens/core/exceptions.py

class UnsupportedModelError(ModelError):
    """No adapter exists for this model type."""

    def __init__(self, model_type: str, supported: Iterable[str]) -> None:
        names = tuple(supported)
        super().__init__(model_type, names)
        self.model_type = model_type
        self.supported = names

    def __str__(self) -> str:
        return (
            f"unsupported model type {self.model_type!r}; supported: "
            f"{', '.join(self.supported)}. "
            "Wrap any scoring function with CallableAdapter to use it."
        )

FeedbackScale dataclass

Ordered feedback events and their rewards, weakest first.

FeedbackScale.of({"click": 1, "cart": 3, "purchase": 10}): a document shown without any event gets 0; a stronger event must not be worth less than a weaker one, otherwise the order of the scale would mean nothing.

Source code in src/ranklens/core/feedback.py

@dataclass(frozen=True, slots=True)
class FeedbackScale:
    """Ordered feedback events and their rewards, weakest first.

    ``FeedbackScale.of({"click": 1, "cart": 3, "purchase": 10})``: a document shown
    without any event gets 0; a stronger event must not be worth less than a weaker one,
    otherwise the order of the scale would mean nothing.
    """

    levels: tuple[tuple[str, float], ...]

    def __post_init__(self) -> None:
        if not isinstance(self.levels, tuple):
            raise TypeError(f"levels must be a tuple, got {type(self.levels).__name__}")
        if not self.levels:
            raise ValueError("a feedback scale needs at least one event")
        names = [name for name, _ in self.levels]
        if len(set(names)) != len(names):
            raise ValueError(f"events repeat in the scale: {names}")
        previous = 0.0
        for name, reward in self.levels:
            if not name.strip():
                raise ValueError("event names must not be empty")
            if not math.isfinite(reward) or reward <= 0:
                raise ValueError(f"reward of {name!r} must be a positive number, got {reward}")
            if reward < previous:
                raise ValueError(
                    f"{name!r} is worth {reward}, less than a weaker event ({previous}): "
                    "list events from weakest to strongest"
                )
            previous = reward

    @classmethod
    def of(cls, rewards: Mapping[str, float]) -> "FeedbackScale":
        """Scale from a mapping in the order of its keys, weakest first."""
        return cls(tuple((name, float(reward)) for name, reward in rewards.items()))

    @property
    def events(self) -> tuple[str, ...]:
        return tuple(name for name, _ in self.levels)

    def reward(self, event: str) -> float:
        """Reward of ``event``; an empty event (shown, nothing happened) is 0."""
        if not event:
            return 0.0
        for name, reward in self.levels:
            if name == event:
                return reward
        raise ValueError(f"unknown event {event!r}; the scale has {', '.join(self.events)}")

of classmethod

of(rewards: Mapping[str, float]) -> FeedbackScale

Scale from a mapping in the order of its keys, weakest first.

Source code in src/ranklens/core/feedback.py

@classmethod
def of(cls, rewards: Mapping[str, float]) -> "FeedbackScale":
    """Scale from a mapping in the order of its keys, weakest first."""
    return cls(tuple((name, float(reward)) for name, reward in rewards.items()))

reward

reward(event: str) -> float

Reward of event; an empty event (shown, nothing happened) is 0.

Source code in src/ranklens/core/feedback.py

def reward(self, event: str) -> float:
    """Reward of ``event``; an empty event (shown, nothing happened) is 0."""
    if not event:
        return 0.0
    for name, reward in self.levels:
        if name == event:
            return reward
    raise ValueError(f"unknown event {event!r}; the scale has {', '.join(self.events)}")

Impression dataclass

One showing of a ranking by the logging policy and the feedback it got.

docs[i] was shown at 1-based positions[i] and earned rewards[i]. Positions ascend but may have gaps: a log may keep only the documents with feedback, and the position is what the propensity of examination depends on.

Source code in src/ranklens/core/feedback.py

@dataclass(frozen=True, slots=True)
class Impression:
    """One showing of a ranking by the logging policy and the feedback it got.

    ``docs[i]`` was shown at 1-based ``positions[i]`` and earned ``rewards[i]``.
    Positions ascend but may have gaps: a log may keep only the documents with
    feedback, and the position is what the propensity of examination depends on.
    """

    impression_id: str
    query_id: QueryId
    docs: tuple[DocId, ...]
    positions: tuple[int, ...]
    rewards: tuple[float, ...]

    def __post_init__(self) -> None:
        for field in ("docs", "positions", "rewards"):
            value = getattr(self, field)
            if not isinstance(value, tuple):
                raise TypeError(f"{field} must be a tuple, got {type(value).__name__}")
        if not len(self.docs) == len(self.positions) == len(self.rewards):
            raise ValueError(
                f"impression {self.impression_id!r}: {len(self.docs)} docs, "
                f"{len(self.positions)} positions and {len(self.rewards)} rewards"
            )
        if any(position < 1 for position in self.positions):
            raise ValueError(f"impression {self.impression_id!r}: positions are 1-based")
        if any(b <= a for a, b in zip(self.positions, self.positions[1:], strict=False)):
            raise ValueError(f"impression {self.impression_id!r}: positions must ascend")
        if any(not math.isfinite(r) or r < 0 for r in self.rewards):
            raise ValueError(f"impression {self.impression_id!r}: rewards must be >= 0")
        seen: set[DocId] = set()
        for doc in self.docs:
            if doc in seen:
                raise DuplicateDocumentError(self.query_id, doc)
            seen.add(doc)

    def __len__(self) -> int:
        return len(self.docs)

Metric

Bases: Protocol

Value of one ranking against the judgements of its query.

Any callable with this signature is a metric, including a plain function. k is the cutoff (None means the whole ranking). The display name comes from the spec the metric was resolved from, not from the metric itself.

Source code in src/ranklens/core/protocols.py

class Metric(Protocol):
    """Value of one ranking against the judgements of its query.

    Any callable with this signature is a metric, including a plain function.
    ``k`` is the cutoff (``None`` means the whole ranking). The display name comes
    from the spec the metric was resolved from, not from the metric itself.
    """

    def __call__(
        self, ranked: Sequence[DocId], judgements: Mapping[DocId, float], k: int | None
    ) -> float: ...

BootstrapInterval dataclass

Percentile bootstrap interval for the mean difference between two runs.

delta is the observed mean of per-query differences (b - a); low and high bound it with confidence 1 - alpha. seed and n_resamples are kept so that the number in a report can be reproduced exactly.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class BootstrapInterval:
    """Percentile bootstrap interval for the mean difference between two runs.

    ``delta`` is the observed mean of per-query differences (b - a); ``low`` and ``high``
    bound it with confidence ``1 - alpha``. ``seed`` and ``n_resamples`` are kept so that
    the number in a report can be reproduced exactly.
    """

    delta: float
    low: float
    high: float
    alpha: float
    n_queries: int
    n_resamples: int
    seed: int

    @property
    def excludes_zero(self) -> bool:
        """Whether the interval lies entirely on one side of zero."""
        return self.low > 0.0 or self.high < 0.0

excludes_zero property

excludes_zero: bool

Whether the interval lies entirely on one side of zero.

ComparisonResult dataclass

Comparison of one metric between two runs, A (baseline) and B.

Scope: one metric at one cutoff, averaged over the queries both runs measured. It does not say whether the runs order documents differently (same metric value, different order is entirely possible), only whether this metric differs.

delta is the mean of per-query differences B - A; ci_low/ci_high are its bootstrap interval and p_value comes from the permutation test — the bootstrap gives an interval, not a p-value, so the two are computed separately and both are reported. q_value is filled in after correcting for multiple comparisons.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class ComparisonResult:
    """Comparison of one metric between two runs, A (baseline) and B.

    Scope: one metric at one cutoff, averaged over the queries both runs measured.
    It does not say whether the runs order documents differently (same metric value,
    different order is entirely possible), only whether this metric differs.

    ``delta`` is the mean of per-query differences B - A; ``ci_low``/``ci_high`` are its
    bootstrap interval and ``p_value`` comes from the permutation test — the bootstrap
    gives an interval, not a p-value, so the two are computed separately and both are
    reported. ``q_value`` is filled in after correcting for multiple comparisons.
    """

    metric: str
    mean_a: float
    mean_b: float
    delta: float
    ci_low: float
    ci_high: float
    alpha: float
    p_value: float
    n_queries: int
    n_only_a: int
    n_only_b: int
    n_resamples: int
    n_permutations: int
    seed: int
    q_value: float | None = None

    @property
    def significant(self) -> bool:
        """Whether the difference passes ``alpha``; after correction, by ``q_value``."""
        return (self.p_value if self.q_value is None else self.q_value) < self.alpha

significant property

significant: bool

Whether the difference passes alpha; after correction, by q_value.

ErrorSummary dataclass

Snapshot of data errors skipped while reading in non-strict mode.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class ErrorSummary:
    """Snapshot of data errors skipped while reading in non-strict mode."""

    total: int = 0
    counts: tuple[tuple[str, int], ...] = ()
    """``(error type, count)``, most frequent first."""
    examples: tuple[tuple[str, tuple[str, ...]], ...] = ()
    """``(error type, first messages)``, in the order types were first seen."""

    def __str__(self) -> str:
        if not self.total:
            return "no data errors"
        counts = ", ".join(f"{kind}: {n}" for kind, n in self.counts)
        lines = [f"skipped {self.total} invalid rows ({counts})"]
        for kind, messages in self.examples:
            lines.append(f"  {kind}, first {len(messages)}:")
            lines.extend(f"    - {message}" for message in messages)
        return "\n".join(lines)

counts class-attribute instance-attribute

counts: tuple[tuple[str, int], ...] = ()

(error type, count), most frequent first.

examples class-attribute instance-attribute

examples: tuple[tuple[str, tuple[str, ...]], ...] = ()

(error type, first messages), in the order types were first seen.

Evaluation dataclass

Metrics of one run and what happened to its queries.

• n_queries: evaluated queries (in the run and in the qrels);
• n_without_relevant: evaluated queries without a positive grade — every metric scores 0 for them (map and mrr also score 0 when no grade reaches rel);
• n_unjudged: queries of the run absent from the qrels — skipped, as in trec_eval;
• n_not_retrieved: queries of the qrels absent from the run — ignored, as in trec_eval.

segments carries the segment values of the evaluated queries, in the order of the schema fields, so that the same evaluation can later be sliced without reading the run again. It is empty when the run has no segment columns.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class Evaluation:
    """Metrics of one run and what happened to its queries.

    - ``n_queries``: evaluated queries (in the run and in the qrels);
    - ``n_without_relevant``: evaluated queries without a positive grade — every metric
      scores 0 for them (``map`` and ``mrr`` also score 0 when no grade reaches ``rel``);
    - ``n_unjudged``: queries of the run absent from the qrels — skipped, as in trec_eval;
    - ``n_not_retrieved``: queries of the qrels absent from the run — ignored, as in trec_eval.

    ``segments`` carries the segment values of the evaluated queries, in the order of the
    schema fields, so that the same evaluation can later be sliced without reading the run
    again. It is empty when the run has no segment columns.
    """

    metrics: tuple[MetricResult, ...]
    n_queries: int
    n_without_relevant: int = 0
    n_unjudged: int = 0
    n_not_retrieved: int = 0
    segments: Mapping[QueryId, SegmentKey] = field(default_factory=lambda: MappingProxyType({}))

    def __getitem__(self, metric: str) -> MetricResult:
        """The result of ``metric``, written as it was resolved or in any spelling of it."""
        labels = {metric}
        with contextlib.suppress(MetricSpecError):
            labels.add(str(parse_spec(metric)))  # "NDCG@010" is "ndcg@10"
        for result in self.metrics:
            if result.metric in labels:
                return result
        raise KeyError(metric)

ListwiseAttribution dataclass

How much a metric drops when a feature (or a group of them) stops carrying signal.

For each feature the values of the documents of a query are shuffled among them, the documents are ranked again, and the drop of the metric is recorded per query: drops[i] is a MetricResult of the drop for features[i], so the usual statistics (compare, bootstrap) apply to it. baseline is the metric itself.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class ListwiseAttribution:
    """How much a metric drops when a feature (or a group of them) stops carrying signal.

    For each feature the values of the documents of a query are shuffled among them,
    the documents are ranked again, and the drop of the metric is recorded per query:
    ``drops[i]`` is a `MetricResult` of the drop for ``features[i]``, so the usual
    statistics (``compare``, bootstrap) apply to it. ``baseline`` is the metric itself.
    """

    metric: str
    features: tuple[str, ...]
    drops: tuple[MetricResult, ...]
    baseline: MetricResult
    n_repeats: int
    seed: int

    def ranked(self) -> tuple[tuple[str, float], ...]:
        """(feature, mean drop over queries), the largest drop first."""
        means = zip(self.features, (drop.mean for drop in self.drops), strict=True)
        return tuple(sorted(means, key=lambda pair: -pair[1]))

ranked

ranked() -> tuple[tuple[str, float], ...]

(feature, mean drop over queries), the largest drop first.

Source code in src/ranklens/core/result.py

def ranked(self) -> tuple[tuple[str, float], ...]:
    """(feature, mean drop over queries), the largest drop first."""
    means = zip(self.features, (drop.mean for drop in self.drops), strict=True)
    return tuple(sorted(means, key=lambda pair: -pair[1]))

MetricResult dataclass

Values of one metric: per query and their mean.

per_query is a read-only mapping (types.MappingProxyType): results are shared between statistics and reports and must not be changed by either. Any other mapping is copied on construction, so changing it later does not change the result.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class MetricResult:
    """Values of one metric: per query and their mean.

    ``per_query`` is a read-only mapping (``types.MappingProxyType``): results are
    shared between statistics and reports and must not be changed by either. Any other
    mapping is copied on construction, so changing it later does not change the result.
    """

    metric: str
    per_query: Mapping[QueryId, float]

    def __post_init__(self) -> None:
        if not isinstance(self.per_query, MappingProxyType):
            object.__setattr__(self, "per_query", MappingProxyType(dict(self.per_query)))

    @property
    def n_queries(self) -> int:
        return len(self.per_query)

    @property
    def mean(self) -> float:
        """Mean over queries; NaN when there are none."""
        if not self.per_query:
            return math.nan
        return math.fsum(self.per_query.values()) / len(self.per_query)

mean property

mean: float

Mean over queries; NaN when there are none.

OffPolicyEstimate dataclass

Value of a new policy estimated from a log collected under an old one.

estimator is "ips" (expected discounted reward, unbiased under PBM) or "snips" (its ratio to the reward of everything shown — a different quantity with lower variance, see docs/math/offpolicy). The interval is a normal approximation over impressions. n_clipped counts rewarded documents whose weight hit clip; n_skipped counts impressions of queries the policy lacks. ess is the effective sample size of the weights 1/p of the rewarded documents, (Σw)² / Σw²: equal to n_rewarded when all weights are equal, far below it when a few weights dominate.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class OffPolicyEstimate:
    """Value of a new policy estimated from a log collected under an old one.

    ``estimator`` is ``"ips"`` (expected discounted reward, unbiased under PBM) or
    ``"snips"`` (its ratio to the reward of everything shown — a different quantity
    with lower variance, see docs/math/offpolicy). The interval is a normal
    approximation over impressions. ``n_clipped`` counts rewarded documents whose
    weight hit ``clip``; ``n_skipped`` counts impressions of queries the policy lacks.
    ``ess`` is the effective sample size of the weights 1/p of the rewarded documents,
    (Σw)² / Σw²: equal to ``n_rewarded`` when all weights are equal, far below it when
    a few weights dominate.
    """

    estimator: str
    value: float
    ci_low: float
    ci_high: float
    alpha: float
    n_impressions: int
    n_skipped: int
    n_rewarded: int
    n_clipped: int
    clip: float | None = None
    ess: float = 0.0

PairAttribution dataclass

Why a model scores document B above (or below) document A, feature by feature.

contributions[i] is the Shapley value of feature feature_names[i] in going from A's features to B's: they add up to score_b - score_a exactly. method is "exact" (all subsets) or "sampled" (n_permutations random orders).

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class PairAttribution:
    """Why a model scores document B above (or below) document A, feature by feature.

    ``contributions[i]`` is the Shapley value of feature ``feature_names[i]`` in going
    from A's features to B's: they add up to ``score_b - score_a`` exactly. ``method``
    is ``"exact"`` (all subsets) or ``"sampled"`` (``n_permutations`` random orders).
    """

    feature_names: tuple[str, ...]
    contributions: tuple[float, ...]
    score_a: float
    score_b: float
    method: str
    n_permutations: int = 0

    @property
    def delta(self) -> float:
        return self.score_b - self.score_a

    def ranked(self) -> tuple[tuple[str, float], ...]:
        """(feature, contribution) by absolute contribution, largest first."""
        pairs = zip(self.feature_names, self.contributions, strict=True)
        return tuple(sorted(pairs, key=lambda pair: -abs(pair[1])))

ranked

ranked() -> tuple[tuple[str, float], ...]

(feature, contribution) by absolute contribution, largest first.

Source code in src/ranklens/core/result.py

def ranked(self) -> tuple[tuple[str, float], ...]:
    """(feature, contribution) by absolute contribution, largest first."""
    pairs = zip(self.feature_names, self.contributions, strict=True)
    return tuple(sorted(pairs, key=lambda pair: -abs(pair[1])))

PPIComparison dataclass

Comparison of two runs on cheap (proxy) judgements corrected by a gold sample.

Prediction-powered inference: the difference measured with proxy judgements on every query, plus a correction measured on a random sample of queries that also have gold judgements. lam weighs the proxy part (0 — gold only, 1 — classic PPI); by default it is chosen from the data to minimize the variance (PPI++). The interval and the p-value are normal approximations; q_value is filled in after correcting for multiple comparisons, as in ComparisonResult.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class PPIComparison:
    """Comparison of two runs on cheap (proxy) judgements corrected by a gold sample.

    Prediction-powered inference: the difference measured with proxy judgements on
    every query, plus a correction measured on a random sample of queries that also
    have gold judgements. ``lam`` weighs the proxy part (0 — gold only, 1 — classic
    PPI); by default it is chosen from the data to minimize the variance (PPI++).
    The interval and the p-value are normal approximations; ``q_value`` is filled in
    after correcting for multiple comparisons, as in `ComparisonResult`.
    """

    metric: str
    mean_a: float
    mean_b: float
    delta: float
    ci_low: float
    ci_high: float
    alpha: float
    p_value: float
    lam: float
    n_proxy: int
    n_gold: int
    q_value: float | None = None

    @property
    def significant(self) -> bool:
        """Whether the difference passes ``alpha``; after correction, by ``q_value``."""
        return (self.p_value if self.q_value is None else self.q_value) < self.alpha

significant property

significant: bool

Whether the difference passes alpha; after correction, by q_value.

QueryDiagnosis dataclass

A query where the model does badly, and which features move its metric.

causes are (feature, drop of the metric when the feature is shuffled), largest absolute drop first. A positive drop: the feature helps here. A negative one: shuffling it improves the ranking — the model trusts a feature that misleads it on this query, which is usually the finding worth acting on.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class QueryDiagnosis:
    """A query where the model does badly, and which features move its metric.

    ``causes`` are (feature, drop of the metric when the feature is shuffled), largest
    absolute drop first. A positive drop: the feature helps here. A negative one:
    shuffling it **improves** the ranking — the model trusts a feature that misleads it
    on this query, which is usually the finding worth acting on.
    """

    query_id: QueryId
    metric: str
    value: float
    causes: tuple[tuple[str, float], ...]

WeightDiagnostics dataclass

How the weights 1/p of the rewarded documents of a log are distributed.

ess = (Σw)² / Σw²; ess_share = ess / n_weights — near 1 when the weights are alike, near 0 when a few dominate. top1_mass is the share of the total weight held by the largest 1% of weights. weights are the weights themselves, one per rewarded document, for plotting their distribution.

Source code in src/ranklens/core/result.py

@dataclass(frozen=True, slots=True)
class WeightDiagnostics:
    """How the weights 1/p of the rewarded documents of a log are distributed.

    ``ess`` = (Σw)² / Σw²; ``ess_share`` = ess / n_weights — near 1 when the weights
    are alike, near 0 when a few dominate. ``top1_mass`` is the share of the total
    weight held by the largest 1% of weights. ``weights`` are the weights themselves,
    one per rewarded document, for plotting their distribution.
    """

    n_weights: int
    ess: float
    ess_share: float
    max_weight: float
    p99_weight: float
    top1_mass: float
    n_clipped: int
    clip: float | None = None
    weights: tuple[float, ...] = ()

RankedList dataclass

Ranking for one query: docs[position] is shown at 0-based position.

Source code in src/ranklens/core/types.py

@dataclass(frozen=True, slots=True)
class RankedList:
    """Ranking for one query: ``docs[position]`` is shown at 0-based ``position``."""

    query_id: QueryId
    docs: tuple[DocId, ...]
    scores: tuple[float, ...] | None = None
    segments: SegmentKey = ()

    def __post_init__(self) -> None:
        # frozen protects attributes, not a mutable container inside them
        if not isinstance(self.docs, tuple):
            raise TypeError(f"docs must be a tuple, got {type(self.docs).__name__}")
        if not isinstance(self.segments, tuple):
            raise TypeError(f"segments must be a tuple, got {type(self.segments).__name__}")
        if self.scores is not None:
            if not isinstance(self.scores, tuple):
                raise TypeError(f"scores must be a tuple, got {type(self.scores).__name__}")
            if len(self.scores) != len(self.docs):
                raise ValueError(
                    f"query {self.query_id!r}: {len(self.scores)} scores for {len(self.docs)} docs"
                )
        seen: set[DocId] = set()
        for doc in self.docs:
            if doc in seen:
                raise DuplicateDocumentError(self.query_id, doc)
            seen.add(doc)

    def __len__(self) -> int:
        return len(self.docs)

    def top(self, k: int) -> tuple[DocId, ...]:
        """First ``k`` documents (fewer if the list is shorter)."""
        if k < 1:
            raise ValueError(f"k must be >= 1, got {k}")
        return self.docs[:k]

top

top(k: int) -> tuple[DocId, ...]

First k documents (fewer if the list is shorter).

Source code in src/ranklens/core/types.py

def top(self, k: int) -> tuple[DocId, ...]:
    """First ``k`` documents (fewer if the list is shorter)."""
    if k < 1:
        raise ValueError(f"k must be >= 1, got {k}")
    return self.docs[:k]