Coverage for cafl4ds/eval.py: 90%
56 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-17 14:39 +0000
1"""Downstream evaluation — probe-on-past, forgetting, and the B5 adaptation floor.
3The *validating* axis of the study (the payoff, `[STD]`), kept separate from the *diagnostic*
4health signals in :mod:`cafl4ds.monitor`. Three pieces, all label-in-the-probe-only:
6* **Probe-on-past** — the current encoder, evaluated on every past era's held-out set, builds
7 an accuracy matrix ``R[i][j]`` = accuracy on era ``j`` after training through era ``i``
8 (:class:`PerEraProbe`, one row recorded per era as the correlated stream advances).
9* **Forgetting metrics** — from that matrix, :func:`backward_transfer` (Lopez-Paz & Ranzato
10 2017) and :func:`forgetting_measure` (Chaudhry et al. 2018): how much learning later eras
11 eroded accuracy on earlier ones. The direct readout of *degradation present* (question **b**).
12* **Adaptation vs. B5** — :func:`adaptation_report` compares the adapted encoder's downstream
13 probe accuracy against the init-matched **frozen** backbone (baseline B5,
14 :func:`cafl4ds.measurements.frozen_baseline`): does adapting beat doing nothing (question
15 **a**)?
17Forgetting is a *correlated-stream* phenomenon: with a single era (IID) there is no past to
18forget, so the matrix has one row and the forgetting metrics are ``None`` (undefined).
19"""
21from __future__ import annotations
23from typing import TypedDict
25from cafl4ds import measurements
26from cafl4ds.data.streams import EvalSet, EvalSets
27from cafl4ds.measurements import Encoder
29# An accuracy matrix: ``matrix[i][j]`` = accuracy on era ``j`` after training through era ``i``.
30AccuracyMatrix = dict[int, dict[int, float]]
33class EraSummary(TypedDict):
34 """The reduced probe-on-past report (final per-era accuracies + forgetting metrics)."""
36 per_era_final: dict[int, float]
37 backward_transfer: float | None
38 forgetting_measure: float | None
39 num_eras: int
42def _probe(encode: Encoder, support: EvalSet, query: EvalSet, probe: str, knn_k: int) -> float:
43 """Run one downstream probe (kNN or linear) on frozen features.
45 Args:
46 encode: Frozen encoder callable ``inputs -> embeddings``.
47 support: The probe's support/database split.
48 query: The split to score.
49 probe: ``"knn"`` or ``"linear"``.
50 knn_k: Neighbour count for the kNN probe (ignored by the linear probe).
52 Returns:
53 Top-1 query accuracy in ``[0, 1]``.
55 Raises:
56 ValueError: If ``probe`` is not ``"knn"`` or ``"linear"``.
57 """
58 if probe == "knn": 58 ↛ 60line 58 didn't jump to line 60 because the condition on line 58 was always true
59 return measurements.knn_probe(encode, (support.images, support.labels), (query.images, query.labels), k=knn_k)
60 if probe == "linear":
61 return measurements.linear_probe(encode, (support.images, support.labels), (query.images, query.labels))
62 raise ValueError(f"probe must be 'knn' or 'linear'; got {probe!r}.")
65class PerEraProbe:
66 """Accumulates the per-era accuracy matrix by probing the current encoder over past eras.
68 Uses the stream's fixed :attr:`~cafl4ds.data.streams.EvalSets.probe_support` as the probe
69 database and each era's held-out :attr:`~cafl4ds.data.streams.EvalSets.per_era` set as the
70 query. The loop calls :meth:`record` once per era (as it completes) with the live encoder,
71 which appends one matrix row; :meth:`summary` reduces the matrix to the reportable metrics.
72 """
74 def __init__(self, eval_sets: EvalSets, probe: str = "knn", knn_k: int = 20) -> None:
75 """Configure the per-era probe.
77 Args:
78 eval_sets: The stream's held-out eval sets (fixed support + per-era queries).
79 probe: Downstream probe to use, ``"knn"`` or ``"linear"`` (labels used HERE ONLY).
80 knn_k: Neighbour count for the kNN probe.
81 """
82 self.eval_sets = eval_sets
83 self.probe = probe
84 self.knn_k = knn_k
85 self.matrix: AccuracyMatrix = {}
87 def record(self, encode: Encoder, era_completed: int) -> dict[int, float]:
88 """Probe the current encoder on every era seen so far and store the row.
90 Args:
91 encode: The live (frozen for the probe) encoder callable.
92 era_completed: Index of the era just finished; the encoder is scored on eras
93 ``0 .. era_completed`` (past eras + the one just learned).
95 Returns:
96 The recorded row ``{era: accuracy}`` (also stored in :attr:`matrix`).
97 """
98 row = {
99 era: _probe(encode, self.eval_sets.probe_support, eval_set, self.probe, self.knn_k)
100 for era, eval_set in sorted(self.eval_sets.per_era.items())
101 if era <= era_completed
102 }
103 self.matrix[era_completed] = row
104 return row
106 def summary(self) -> EraSummary:
107 """Reduce the accumulated matrix to the reportable per-era / forgetting metrics.
109 Returns:
110 An :class:`EraSummary`. The forgetting metrics are ``None`` when fewer than two eras
111 were recorded (no past to forget — e.g. an IID stream).
112 """
113 final_era = max(self.matrix) if self.matrix else None
114 return {
115 "per_era_final": self.matrix.get(final_era, {}) if final_era is not None else {},
116 "backward_transfer": backward_transfer(self.matrix),
117 "forgetting_measure": forgetting_measure(self.matrix),
118 "num_eras": len(self.matrix),
119 }
122def backward_transfer(matrix: AccuracyMatrix) -> float | None:
123 """Backward Transfer: mean change in past-era accuracy from first-learned to final.
125 ``BWT = mean_{j < T-1} (R[T-1][j] - R[j][j])`` (Lopez-Paz & Ranzato 2017), where ``R[j][j]``
126 is accuracy on era ``j`` right after learning it and ``R[T-1][j]`` is its accuracy at the
127 end. **Negative** BWT means later eras eroded earlier ones (forgetting); positive means
128 later learning helped past eras.
130 Args:
131 matrix: The per-era accuracy matrix (one row per completed era).
133 Returns:
134 The BWT, or ``None`` if fewer than two eras were recorded.
135 """
136 eras = sorted(matrix)
137 if len(eras) < 2:
138 return None
139 final = eras[-1]
140 row_final = matrix[final]
141 diffs = [row_final[j] - matrix[j][j] for j in eras[:-1] if j in row_final and j in matrix[j]]
142 return sum(diffs) / len(diffs) if diffs else None
145def forgetting_measure(matrix: AccuracyMatrix) -> float | None:
146 """Forgetting Measure: mean drop from an era's *best-ever* accuracy to its final accuracy.
148 ``FM = mean_{j < T-1} ( max_{j <= l < T-1} R[l][j] - R[T-1][j] )`` (Chaudhry et al. 2018) —
149 for each past era, how far its accuracy fell from the highest it reached at any earlier
150 checkpoint to its value at the end. Higher = more forgetting.
152 Args:
153 matrix: The per-era accuracy matrix (one row per completed era).
155 Returns:
156 The forgetting measure, or ``None`` if fewer than two eras were recorded.
157 """
158 eras = sorted(matrix)
159 if len(eras) < 2:
160 return None
161 final = eras[-1]
162 row_final = matrix[final]
163 forgets = []
164 for j in eras[:-1]:
165 prior = [matrix[k][j] for k in eras if j <= k < final and j in matrix[k]]
166 if prior and j in row_final: 166 ↛ 164line 166 didn't jump to line 164 because the condition on line 166 was always true
167 forgets.append(max(prior) - row_final[j])
168 return sum(forgets) / len(forgets) if forgets else None
171def adaptation_report(
172 adapted_encode: Encoder,
173 frozen_encode: Encoder,
174 eval_sets: EvalSets,
175 probe: str = "knn",
176 knn_k: int = 20,
177) -> dict[str, float]:
178 """Compare the adapted encoder against the init-matched frozen backbone (B5).
180 Answers the existential question (a): does adapting the backbone beat the frozen model?
181 Both are scored with the same downstream probe on the stream's fixed support/query split.
183 Args:
184 adapted_encode: The encoder after streaming adaptation.
185 frozen_encode: The init-matched never-updated encoder (B5: frozen-pretrained in the
186 pretrained regime, frozen-random from scratch).
187 eval_sets: The stream's held-out eval sets (fixed support + query).
188 probe: Downstream probe, ``"knn"`` or ``"linear"``.
189 knn_k: Neighbour count for the kNN probe.
191 Returns:
192 ``{"adapted_acc", "b5_acc", "gain"}`` — the two accuracies and ``adapted - b5``.
193 """
194 support, query = eval_sets.probe_support, eval_sets.probe_query
195 adapted_acc = _probe(adapted_encode, support, query, probe, knn_k)
196 kwargs = {"k": knn_k} if probe == "knn" else {}
197 b5_acc = measurements.frozen_baseline(
198 frozen_encode, (support.images, support.labels), (query.images, query.labels), probe=probe, **kwargs
199 )
200 return {"adapted_acc": adapted_acc, "b5_acc": b5_acc, "gain": adapted_acc - b5_acc}