| 7 | |
| 8 | |
| 9 | class SignalWeightTracker: |
| 10 | def __init__(self, initial_weights: list[float]) -> None: |
| 11 | self._weights = list(initial_weights) |
| 12 | self._update_count = 0 |
| 13 | |
| 14 | @property |
| 15 | def weights(self) -> list[float]: |
| 16 | return list(self._weights) |
| 17 | |
| 18 | def update( |
| 19 | self, |
| 20 | predictions: list[int | None], |
| 21 | abstained: list[bool], |
| 22 | actual_tier: int, |
| 23 | ) -> None: |
| 24 | lr = max(0.01, 0.1 / (1 + self._update_count * 0.001)) |
| 25 | self._update_count += 1 |
| 26 | for i in range(len(self._weights)): |
| 27 | if abstained[i]: |
| 28 | continue |
| 29 | if predictions[i] == actual_tier: |
| 30 | self._weights[i] *= (1 + lr) |
| 31 | else: |
| 32 | self._weights[i] *= (1 - lr) |
| 33 | self._normalize() |
| 34 | |
| 35 | def _normalize(self) -> None: |
| 36 | total = sum(self._weights) |
| 37 | if total > 0: |
| 38 | self._weights = [w / total for w in self._weights] |
no outgoing calls