Перейти к содержанию

ranklens.offpolicy

ranklens.offpolicy

Off-policy evaluation: estimate a new ranking from feedback logged under an old one.

ESS_WARNING_SHARE module-attribute 

ESS_WARNING_SHARE = 0.1

ESS below this share of the weights means a few weights dominate the estimate.

Discount module-attribute 

Discount = Callable[[Positions], npt.NDArray[np.float64]]

Weight λ(r) of a 1-based position r in the metric being estimated.

EventColumns dataclass

Column names of the show and feedback tables.

Shows hold either an items array per request (position = index + 1) or one row per shown document with doc_id and position. Feedback holds request_id, doc_id and event; ts is needed in both only when a window is set.

Source code in src/ranklens/offpolicy/events.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@dataclass(frozen=True, slots=True)
class EventColumns:
    """Column names of the show and feedback tables.

    Shows hold either an ``items`` array per request (position = index + 1) or one row
    per shown document with ``doc_id`` and ``position``. Feedback holds ``request_id``,
    ``doc_id`` and ``event``; ``ts`` is needed in both only when a window is set.
    """

    request_id: str = "request_id"
    query_id: str = "query_id"
    items: str = "items"
    doc_id: str = "doc_id"
    position: str = "position"
    event: str = "event"
    ts: str = "ts"

EventLog dataclass

Impressions built from events and what did not make it into them.

Source code in src/ranklens/offpolicy/events.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass(frozen=True, slots=True)
class EventLog:
    """Impressions built from events and what did not make it into them."""

    impressions: tuple[Impression, ...]
    n_duplicate_shows: int = 0
    """Repeated show events (array form) or repeated shown rows (row form)."""
    n_matched: int = 0
    n_not_shown: int = 0
    n_outside_window: int = 0
    ignored_events: Mapping[str, int] = field(default_factory=lambda: MappingProxyType({}))
    n_invalid_shows: int = 0
    """Shown rows without a request, query or document id."""

    def __str__(self) -> str:
        ignored = ", ".join(f"{event}: {n}" for event, n in self.ignored_events.items())
        return (
            f"{len(self.impressions)} impressions; feedback events: {self.n_matched} matched, "
            f"{self.n_not_shown} on documents not shown, {self.n_outside_window} outside "
            f"the window; {self.n_duplicate_shows} repeated show events dropped"
            + (f"; {self.n_invalid_shows} shown rows without an id" if self.n_invalid_shows else "")
            + (f"; ignored event types: {ignored}" if ignored else "")
        )

n_duplicate_shows class-attribute instance-attribute 

n_duplicate_shows: int = 0

Repeated show events (array form) or repeated shown rows (row form).

n_invalid_shows class-attribute instance-attribute 

n_invalid_shows: int = 0

Shown rows without a request, query or document id.

Propensity dataclass

Examination probability of each position: examination[r - 1] is p_r.

Source code in src/ranklens/offpolicy/propensity.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@dataclass(frozen=True, slots=True)
class Propensity:
    """Examination probability of each position: ``examination[r - 1]`` is p_r."""

    examination: tuple[float, ...]

    def __post_init__(self) -> None:
        if not isinstance(self.examination, tuple):
            raise TypeError(f"examination must be a tuple, got {type(self.examination).__name__}")
        if not self.examination:
            raise ValueError("at least one position is needed")
        if not all(0.0 < p <= 1.0 for p in self.examination):
            # p = 0 would make the position invisible to any estimator, not merely rare
            raise ValueError("examination probabilities must lie in (0, 1]")

    @classmethod
    def power(cls, depth: int, eta: float = 1.0) -> "Propensity":
        """p_r = (1 / r)^eta for positions 1..depth; eta = 0 means everything is examined."""
        if depth < 1:
            raise ValueError(f"depth must be >= 1, got {depth}")
        if eta < 0:
            raise ValueError(f"eta must be >= 0, got {eta}")
        return cls(tuple(float(r**-eta) for r in range(1, depth + 1)))

    @classmethod
    def of(cls, values: Sequence[float]) -> "Propensity":
        return cls(tuple(float(value) for value in values))

    @property
    def depth(self) -> int:
        return len(self.examination)

    def at(self, positions: Sequence[int] | Positions) -> npt.NDArray[np.float64]:
        """p_r for 1-based positions; a position deeper than known is an error, not 0."""
        index = np.asarray(positions, dtype=np.int64) - 1
        if index.size and (index.min() < 0 or index.max() >= self.depth):
            raise ValueError(
                f"positions must lie in 1..{self.depth}, got {int(index.min()) + 1}.."
                f"{int(index.max()) + 1}"
            )
        return np.asarray(self.examination, dtype=np.float64)[index]

power classmethod 

power(depth: int, eta: float = 1.0) -> Propensity

p_r = (1 / r)^eta for positions 1..depth; eta = 0 means everything is examined.

Source code in src/ranklens/offpolicy/propensity.py 
37
38
39
40
41
42
43
44
@classmethod
def power(cls, depth: int, eta: float = 1.0) -> "Propensity":
    """p_r = (1 / r)^eta for positions 1..depth; eta = 0 means everything is examined."""
    if depth < 1:
        raise ValueError(f"depth must be >= 1, got {depth}")
    if eta < 0:
        raise ValueError(f"eta must be >= 0, got {eta}")
    return cls(tuple(float(r**-eta) for r in range(1, depth + 1)))

at 

at(
    positions: Sequence[int] | Positions,
) -> npt.NDArray[np.float64]

p_r for 1-based positions; a position deeper than known is an error, not 0.

Source code in src/ranklens/offpolicy/propensity.py 
54
55
56
57
58
59
60
61
62
def at(self, positions: Sequence[int] | Positions) -> npt.NDArray[np.float64]:
    """p_r for 1-based positions; a position deeper than known is an error, not 0."""
    index = np.asarray(positions, dtype=np.int64) - 1
    if index.size and (index.min() < 0 or index.max() >= self.depth):
        raise ValueError(
            f"positions must lie in 1..{self.depth}, got {int(index.min()) + 1}.."
            f"{int(index.max()) + 1}"
        )
    return np.asarray(self.examination, dtype=np.float64)[index]

ClickWorld dataclass

Queries with known relevance; builds policies, logs and true policy values.

The relevance is copied into read-only mappings: the truth a log was simulated from must not change under it. Worlds compare by value and are not hashable.

Source code in src/ranklens/offpolicy/simulator.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@dataclass(frozen=True, slots=True)
class ClickWorld:
    """Queries with known relevance; builds policies, logs and true policy values.

    The relevance is copied into read-only mappings: the truth a log was simulated from
    must not change under it. Worlds compare by value and are not hashable.
    """

    relevance: Mapping[QueryId, Mapping[DocId, float]] = field(repr=False)
    __hash__ = None  # type: ignore[assignment]

    def __post_init__(self) -> None:
        frozen = {query: MappingProxyType(dict(judged)) for query, judged in self.relevance.items()}
        object.__setattr__(self, "relevance", MappingProxyType(frozen))

    def __repr__(self) -> str:
        n_docs = sum(len(judged) for judged in self.relevance.values())
        return f"ClickWorld({len(self.relevance)} queries, {n_docs} documents)"

    @classmethod
    def random(cls, n_queries: int = 200, docs_per_query: int = 20, seed: int = 0) -> "ClickWorld":
        if n_queries < 1 or docs_per_query < 1:
            raise ValueError("a world needs at least one query and one document per query")
        rng = np.random.default_rng(seed)
        grades = rng.choice(MAX_GRADE + 1, size=(n_queries, docs_per_query), p=_GRADE_SHARES)
        return cls(
            {
                QueryId(f"q{q:04d}"): {
                    DocId(f"d{d:03d}"): float(grades[q, d]) for d in range(docs_per_query)
                }
                for q in range(n_queries)
            }
        )

    def attractiveness(self, query: QueryId, doc: DocId) -> float:
        """Probability of a click on ``doc`` once it is examined."""
        return float((2.0 ** self.relevance[query][doc] - 1.0) / (2.0**MAX_GRADE - 1.0))

    def policy(self, noise: float, seed: int = 0) -> dict[QueryId, RankedList]:
        """Rank every query by relevance plus N(0, noise²): 0 is the ideal ranking."""
        if noise < 0:
            raise ValueError(f"noise must be >= 0, got {noise}")
        rng = np.random.default_rng(seed)
        rankings: dict[QueryId, RankedList] = {}
        for query, judgements in self.relevance.items():
            docs = list(judgements)
            scores = np.array([judgements[d] for d in docs]) + rng.normal(0.0, noise, len(docs))
            order = np.lexsort((np.arange(len(docs)), -scores))  # ties: first listed wins
            rankings[query] = RankedList(query, tuple(docs[i] for i in order))
        return rankings

    def simulate(
        self,
        logging: Mapping[QueryId, RankedList],
        propensity: Propensity,
        n_impressions: int,
        seed: int = 0,
    ) -> tuple[Impression, ...]:
        """A log of ``logging``: queries drawn uniformly, top ``propensity.depth`` shown."""
        if n_impressions < 1:
            raise ValueError(f"n_impressions must be >= 1, got {n_impressions}")
        rng = np.random.default_rng(seed)
        queries = list(logging)
        depth = min(propensity.depth, min(len(logging[q]) for q in queries))
        shown = [logging[q].docs[:depth] for q in queries]
        attraction = np.array(
            [
                [self.attractiveness(q, d) for d in docs]
                for q, docs in zip(queries, shown, strict=True)
            ]
        )
        examination = propensity.at(np.arange(1, depth + 1))

        drawn = rng.integers(len(queries), size=n_impressions)
        examined = rng.random((n_impressions, depth)) < examination
        attracted = rng.random((n_impressions, depth)) < attraction[drawn]
        clicks = (examined & attracted).astype(np.float64)
        positions = tuple(range(1, depth + 1))
        return tuple(
            Impression(
                impression_id=f"i{i:07d}",
                query_id=queries[q],
                docs=shown[q],
                positions=positions,
                rewards=tuple(clicks[i].tolist()),
            )
            for i, q in enumerate(drawn.tolist())
        )

    def value(self, policy: Mapping[QueryId, RankedList], discount: Discount) -> float:
        """True value of ``policy``: mean over queries of Σ_d λ(rank(d)) · a(d).

        This is the expected discounted number of clicks if every position were
        examined — what an unbiased estimator recovers from a log of another policy.
        """
        totals = []
        for query, ranking in policy.items():
            weights = discount(np.arange(1, len(ranking) + 1))
            gains = np.array([self.attractiveness(query, doc) for doc in ranking.docs])
            totals.append(float(weights @ gains))
        return float(np.mean(totals))

attractiveness 

attractiveness(query: QueryId, doc: DocId) -> float

Probability of a click on doc once it is examined.

Source code in src/ranklens/offpolicy/simulator.py 
64
65
66
def attractiveness(self, query: QueryId, doc: DocId) -> float:
    """Probability of a click on ``doc`` once it is examined."""
    return float((2.0 ** self.relevance[query][doc] - 1.0) / (2.0**MAX_GRADE - 1.0))

policy 

policy(
    noise: float, seed: int = 0
) -> dict[QueryId, RankedList]

Rank every query by relevance plus N(0, noise²): 0 is the ideal ranking.

Source code in src/ranklens/offpolicy/simulator.py 
68
69
70
71
72
73
74
75
76
77
78
79
def policy(self, noise: float, seed: int = 0) -> dict[QueryId, RankedList]:
    """Rank every query by relevance plus N(0, noise²): 0 is the ideal ranking."""
    if noise < 0:
        raise ValueError(f"noise must be >= 0, got {noise}")
    rng = np.random.default_rng(seed)
    rankings: dict[QueryId, RankedList] = {}
    for query, judgements in self.relevance.items():
        docs = list(judgements)
        scores = np.array([judgements[d] for d in docs]) + rng.normal(0.0, noise, len(docs))
        order = np.lexsort((np.arange(len(docs)), -scores))  # ties: first listed wins
        rankings[query] = RankedList(query, tuple(docs[i] for i in order))
    return rankings

simulate 

simulate(
    logging: Mapping[QueryId, RankedList],
    propensity: Propensity,
    n_impressions: int,
    seed: int = 0,
) -> tuple[Impression, ...]

A log of logging: queries drawn uniformly, top propensity.depth shown.

Source code in src/ranklens/offpolicy/simulator.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def simulate(
    self,
    logging: Mapping[QueryId, RankedList],
    propensity: Propensity,
    n_impressions: int,
    seed: int = 0,
) -> tuple[Impression, ...]:
    """A log of ``logging``: queries drawn uniformly, top ``propensity.depth`` shown."""
    if n_impressions < 1:
        raise ValueError(f"n_impressions must be >= 1, got {n_impressions}")
    rng = np.random.default_rng(seed)
    queries = list(logging)
    depth = min(propensity.depth, min(len(logging[q]) for q in queries))
    shown = [logging[q].docs[:depth] for q in queries]
    attraction = np.array(
        [
            [self.attractiveness(q, d) for d in docs]
            for q, docs in zip(queries, shown, strict=True)
        ]
    )
    examination = propensity.at(np.arange(1, depth + 1))

    drawn = rng.integers(len(queries), size=n_impressions)
    examined = rng.random((n_impressions, depth)) < examination
    attracted = rng.random((n_impressions, depth)) < attraction[drawn]
    clicks = (examined & attracted).astype(np.float64)
    positions = tuple(range(1, depth + 1))
    return tuple(
        Impression(
            impression_id=f"i{i:07d}",
            query_id=queries[q],
            docs=shown[q],
            positions=positions,
            rewards=tuple(clicks[i].tolist()),
        )
        for i, q in enumerate(drawn.tolist())
    )

value 

value(
    policy: Mapping[QueryId, RankedList], discount: Discount
) -> float

True value of policy: mean over queries of Σ_d λ(rank(d)) · a(d).

This is the expected discounted number of clicks if every position were examined — what an unbiased estimator recovers from a log of another policy.

Source code in src/ranklens/offpolicy/simulator.py 
119
120
121
122
123
124
125
126
127
128
129
130
def value(self, policy: Mapping[QueryId, RankedList], discount: Discount) -> float:
    """True value of ``policy``: mean over queries of Σ_d λ(rank(d)) · a(d).

    This is the expected discounted number of clicks if every position were
    examined — what an unbiased estimator recovers from a log of another policy.
    """
    totals = []
    for query, ranking in policy.items():
        weights = discount(np.arange(1, len(ranking) + 1))
        gains = np.array([self.attractiveness(query, doc) for doc in ranking.docs])
        totals.append(float(weights @ gains))
    return float(np.mean(totals))

weight_diagnostics 

weight_diagnostics(
    log: Iterable[Impression],
    propensity: Propensity,
    *,
    clip: float | None = None,
) -> WeightDiagnostics

Distribution of the weights 1/p of the rewarded documents, after clip.

Keeps one number per rewarded document: rewards are rare, so this is far less than the log itself.

Source code in src/ranklens/offpolicy/diagnostics.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def weight_diagnostics(
    log: Iterable[Impression], propensity: Propensity, *, clip: float | None = None
) -> WeightDiagnostics:
    """Distribution of the weights 1/p of the rewarded documents, after ``clip``.

    Keeps one number per rewarded document: rewards are rare, so this is far less
    than the log itself.
    """
    if clip is not None and not clip >= 1.0:
        raise ValueError(f"clip must be >= 1 (a weight 1/p is never below 1), got {clip}")
    chunks = [
        1.0 / propensity.at([p for p, r in zip(i.positions, i.rewards, strict=True) if r > 0])
        for i in log
    ]
    raw = np.concatenate(chunks) if chunks else np.empty(0)
    if raw.size == 0:
        return WeightDiagnostics(0, 0.0, 0.0, 0.0, 0.0, 0.0, 0, clip)
    weights = raw if clip is None else np.minimum(raw, clip)
    ess = float(weights.sum() ** 2 / (weights**2).sum())
    top = np.sort(weights)[::-1][: max(1, int(np.ceil(weights.size / 100)))]
    return WeightDiagnostics(
        n_weights=int(weights.size),
        ess=ess,
        ess_share=ess / weights.size,
        max_weight=float(weights.max()),
        p99_weight=float(np.quantile(weights, 0.99)),
        top1_mass=float(top.sum() / weights.sum()),
        n_clipped=int((raw > clip).sum()) if clip is not None else 0,
        clip=clip,
        weights=tuple(weights.tolist()),
    )

estimate 

estimate(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    estimators: Sequence[Estimator] = ("ips", "snips"),
    clip: float | None = None,
    alpha: float = 0.05,
) -> tuple[OffPolicyEstimate, ...]

Several estimators from one pass over the log, in the order of estimators.

Warns with DegenerateWeightsWarning when the effective sample size of the weights falls below 10% of their number: then a handful of rewards at deep positions decide the estimate, and its interval is not to be trusted.

Source code in src/ranklens/offpolicy/estimators.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def estimate(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    estimators: Sequence[Estimator] = ("ips", "snips"),
    clip: float | None = None,
    alpha: float = 0.05,
) -> tuple[OffPolicyEstimate, ...]:
    """Several estimators from one pass over the log, in the order of ``estimators``.

    Warns with `DegenerateWeightsWarning` when the effective sample size of the weights
    falls below 10% of their number: then a handful of rewards at deep positions
    decide the estimate, and its interval is not to be trusted.
    """
    return _run(log, policy, propensity, discount, estimators, clip, alpha)

ips 

ips(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    clip: float | None = None,
    alpha: float = 0.05,
) -> OffPolicyEstimate

Expected discounted reward of policy: mean over impressions of Σ c·λ/p.

Source code in src/ranklens/offpolicy/estimators.py 
58
59
60
61
62
63
64
65
66
67
68
69
def ips(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    clip: float | None = None,
    alpha: float = 0.05,
) -> OffPolicyEstimate:
    """Expected discounted reward of ``policy``: mean over impressions of Σ c·λ/p."""
    (result,) = _run(log, policy, propensity, discount, ("ips",), clip, alpha)
    return result

snips 

snips(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    clip: float | None = None,
    alpha: float = 0.05,
) -> OffPolicyEstimate

Share of the reweighted reward of the shown documents that policy collects.

Source code in src/ranklens/offpolicy/estimators.py 
72
73
74
75
76
77
78
79
80
81
82
83
def snips(
    log: Iterable[Impression],
    policy: Mapping[QueryId, RankedList],
    propensity: Propensity,
    discount: Discount,
    *,
    clip: float | None = None,
    alpha: float = 0.05,
) -> OffPolicyEstimate:
    """Share of the reweighted reward of the shown documents that ``policy`` collects."""
    (result,) = _run(log, policy, propensity, discount, ("snips",), clip, alpha)
    return result

impressions_from_events 

impressions_from_events(
    shows: DataFrame,
    feedback: DataFrame,
    scale: FeedbackScale,
    *,
    window: Timedelta | str | None = None,
    columns: EventColumns | None = None,
) -> EventLog

Join show and feedback events into one Impression per request.

window (e.g. "1h") counts from the show: a purchase three days later is not credited to this ranking. Without it, time is not looked at. Impressions come in the order of the first show of each request.

Source code in src/ranklens/offpolicy/events.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def impressions_from_events(
    shows: pd.DataFrame,
    feedback: pd.DataFrame,
    scale: FeedbackScale,
    *,
    window: pd.Timedelta | str | None = None,
    columns: EventColumns | None = None,
) -> EventLog:
    """Join show and feedback events into one `Impression` per request.

    ``window`` (e.g. ``"1h"``) counts from the show: a purchase three days later is
    not credited to this ranking. Without it, time is not looked at.
    Impressions come in the order of the first show of each request.
    """
    c = columns or EventColumns()
    limit = None if window is None else pd.Timedelta(window)
    if limit is not None and limit < pd.Timedelta(0):
        raise ValueError(f"window must not be negative, got {window}")
    timed = limit is not None
    rows_form = c.items not in shows.columns
    _require(
        shows,
        [
            c.request_id,
            c.query_id,
            *([c.doc_id, c.position] if rows_form else []),
            *([c.ts] if timed else []),
        ],
        "shows",
    )
    _require(feedback, [c.request_id, c.doc_id, c.event, *([c.ts] if timed else [])], "feedback")
    shows = shows.assign(
        **{c.request_id: _ids(shows[c.request_id]), c.query_id: _ids(shows[c.query_id])}
    )
    shown, n_duplicate, n_invalid = _long_shows(shows, c, need_ts=timed)
    feedback = feedback.assign(
        **{c.request_id: _ids(feedback[c.request_id]), c.doc_id: _ids(feedback[c.doc_id])}
    )
    # an event without an id matches nothing; NA keys would match each other in a merge
    feedback = feedback.fillna({c.request_id: "\x00", c.doc_id: "\x00"})

    known = feedback[c.event].isin(scale.events)
    ignored = feedback.loc[~known, c.event].value_counts(sort=False)
    events = feedback.loc[known, [c.request_id, c.doc_id, c.event, *([c.ts] if timed else [])]]
    events = events.assign(_reward=events[c.event].map(scale.reward))

    joined = events.merge(
        shown[[c.request_id, c.doc_id, "_shown_at"]],
        on=[c.request_id, c.doc_id],
        how="left",
        indicator=True,
    )
    on_shown = joined["_merge"] == "both"
    n_not_shown = int((~on_shown).sum())
    joined = joined[on_shown]
    n_outside = 0
    if limit is not None:
        delay = joined[c.ts] - joined["_shown_at"]
        inside = (delay >= pd.Timedelta(0)) & (delay <= limit)
        n_outside = int((~inside).sum())
        joined = joined[inside]

    strongest = joined.groupby([c.request_id, c.doc_id], sort=False)["_reward"].max()
    rewards = shown.join(strongest, on=[c.request_id, c.doc_id])["_reward"]
    rewards = rewards.astype("float64").fillna(0.0)  # object dtype when nothing matched
    shown = shown.assign(_reward=rewards.to_numpy())

    impressions = tuple(
        Impression(
            impression_id=str(request),
            query_id=QueryId(str(group[c.query_id].iloc[0])),
            docs=tuple(DocId(str(doc)) for doc in group[c.doc_id]),
            positions=tuple(int(p) for p in group[c.position]),
            rewards=tuple(float(r) for r in group["_reward"]),
        )
        for request, group in shown.groupby(c.request_id, sort=False)
    )
    return EventLog(
        impressions=impressions,
        n_duplicate_shows=n_duplicate,
        n_matched=len(joined),
        n_not_shown=n_not_shown,
        n_outside_window=n_outside,
        ignored_events=MappingProxyType({str(e): int(n) for e, n in ignored.items()}),
        n_invalid_shows=n_invalid,
    )

dcg_discount 

dcg_discount(k: int) -> Discount

λ(r) = 1[r ≤ k] / log2(r + 1): the estimated metric is DCG@k with the reward as gain.

Source code in src/ranklens/offpolicy/propensity.py 
65
66
67
68
69
70
71
72
73
def dcg_discount(k: int) -> Discount:
    """λ(r) = 1[r ≤ k] / log2(r + 1): the estimated metric is DCG@k with the reward as gain."""
    _check_k(k)

    def discount(positions: Positions) -> npt.NDArray[np.float64]:
        r = np.asarray(positions, dtype=np.float64)
        return np.where(r <= k, 1.0 / np.log2(r + 1.0), 0.0)

    return discount

topk_discount 

topk_discount(k: int) -> Discount

λ(r) = 1[r ≤ k]: the estimated metric is the reward collected in the top k.

Source code in src/ranklens/offpolicy/propensity.py 
76
77
78
79
80
81
82
83
def topk_discount(k: int) -> Discount:
    """λ(r) = 1[r ≤ k]: the estimated metric is the reward collected in the top k."""
    _check_k(k)

    def discount(positions: Positions) -> npt.NDArray[np.float64]:
        return np.where(np.asarray(positions) <= k, 1.0, 0.0)

    return discount