(
self,
max_updates_per_hour: int = 100,
save_every: int = 10,
model_experience: Any = None,
now_fn: Any = None,
buffer_path: Path | None = None,
)
| 60 | """ |
| 61 | |
| 62 | def __init__( |
| 63 | self, |
| 64 | max_updates_per_hour: int = 100, |
| 65 | save_every: int = 10, |
| 66 | model_experience: Any = None, |
| 67 | now_fn: Any = None, |
| 68 | buffer_path: Path | None = None, |
| 69 | ) -> None: |
| 70 | self._buffer: dict[str, RequestContext] = {} |
| 71 | self._max_hourly = max_updates_per_hour |
| 72 | self._save_every = save_every |
| 73 | self._model_experience = model_experience |
| 74 | self._now = now_fn or time.time |
| 75 | self._update_ts: list[float] = [] |
| 76 | self._total_updates: int = 0 |
| 77 | self._since_save: int = 0 |
| 78 | self._buffer_path = buffer_path |
| 79 | if self._buffer_path: |
| 80 | self._load_buffer() |
| 81 | |
| 82 | # ─── Public API ─── |
| 83 |
nothing calls this directly
no test coverage detected