feat(inference): backfill

README.md

@@ -761,7 +761,7 @@ inference/
 
 Healthcheck `python -m inference.health`: флаг `ready` после прогрева и свежий `heartbeat`.
 
-Env (default): `INFERENCE_RUNTIME_DIR` (`runtime`, volume), `POLL_INTERVAL_SECONDS` (2), `EVENTS_BATCH_SIZE` (5000), `EVENTS_REORDER_LAG_SECONDS` (300), `TTL_INTERVAL_SECONDS` (60), `METADATA_REFRESH_SECONDS` (3600), `METADATA_RETRY_SECONDS` (600), `HEALTH_MAX_AGE_SECONDS` (120).
+Env (default): `INFERENCE_RUNTIME_DIR` (`runtime`, volume), `POLL_INTERVAL_SECONDS` (2), `EVENTS_BATCH_SIZE` (5000), `EVENTS_REORDER_LAG_SECONDS` (300), `TTL_INTERVAL_SECONDS` (60), `METADATA_REFRESH_SECONDS` (3600), `METADATA_RETRY_SECONDS` (600), `HEALTH_MAX_AGE_SECONDS` (120), `BACKFILL_HOURS` (0: выкл.; >0 — один раз на том runtime, маркер `backfill.done`, курсор отматывается на события за N часов и поток прогнозирует их заново).
 
 ### Docker и деплой
 

inference/runner.py

@@ -42,8 +42,11 @@
 
 CURSOR_FILE = "events_cursor.json"
 TARGETS_FILE = "incident_targets.json"
+# Бэкфилл выполняется один раз на том runtime: маркер переживает рестарты.
+BACKFILL_FILE = "backfill.done"
 
 SELECT_MAX_ID = "SELECT coalesce(max(id), 0) FROM events"
+SELECT_BACKFILL_LOW = "SELECT min(id) - 1 FROM events WHERE ts >= %s"
 
 # Прогрев — только то, что уже учтено курсором: остальное придёт
 # в цикле, и второй update задвоил бы окна.
@@ -80,6 +83,7 @@ class RunnerSettings:
     backoff_max_seconds: float = 60.0
     shap_max_per_tick: int = 8
     shap_budget_ms: float = 500.0
+    backfill_hours: float = 0.0
 
     @classmethod
     def from_env(
@@ -102,6 +106,7 @@ def from_env(
             ),
             shap_max_per_tick=int(env.get("SHAP_MAX_PER_TICK") or 8),
             shap_budget_ms=_env_float(env, "SHAP_BUDGET_MS", 500.0),
+            backfill_hours=_env_float(env, "BACKFILL_HOURS", 0.0),
         )
 
 
@@ -216,6 +221,9 @@ def start(self, conn: psycopg.Connection) -> None:
                 cursor = EventCursor(
                     low, lag_seconds=self.options.reorder_lag_seconds
                 )
+            backfill = self._backfill_cursor(conn)
+            if backfill is not None:
+                cursor, fresh = backfill, True
             rows = self._warmup(conn, service, cursor)
             self.classed = classed_channels(conn)
             if self.settings.prediction_kind == "INCIDENT":
@@ -239,6 +247,8 @@ def start(self, conn: psycopg.Connection) -> None:
         cursor.save(self.path(CURSOR_FILE))
         if self.targets is not None:
             self.targets.save(self.path(TARGETS_FILE), cursor.low, cursor.seen)
+        if self.options.backfill_hours > 0:
+            self.path(BACKFILL_FILE).touch()
         self.path(READY_FILE).touch()
         log.info(
             "started fresh=%s low=%d seen=%d warmup_rows=%d channels=%d "
@@ -248,14 +258,32 @@ def start(self, conn: psycopg.Connection) -> None:
             self.window.total_seconds() / 3600,
         )
 
+    def _backfill_span(self) -> timedelta:
+        if self.options.backfill_hours <= 0 or self.path(BACKFILL_FILE).exists():
+            return timedelta(0)
+        return timedelta(hours=self.options.backfill_hours)
+
+    def _backfill_cursor(self, conn: psycopg.Connection) -> EventCursor | None:
+        """Курсор на начало окна бэкфилла: поток заново прогнозирует события
+        за backfill_hours, чтобы у тихих с момента старта каналов был прогноз."""
+        span = self._backfill_span()
+        if not span:
+            return None
+        low = conn.execute(SELECT_BACKFILL_LOW, (self.clock() - span,)).fetchone()[0]
+        if low is None:
+            return None
+        log.info("backfill hours=%.0f low=%d", span.total_seconds() / 3600, low)
+        return EventCursor(low, lag_seconds=self.options.reorder_lag_seconds)
+
     def _warmup(
         self,
         conn: psycopg.Connection,
         service: PredictionService,
         cursor: EventCursor,
     ) -> int:
         params = {
-            "since": self.clock() - self.window,
+            # При бэкфилле окна прогреваются и до его начала.
+            "since": self.clock() - self.window - self._backfill_span(),
             "low": cursor.low,
             "seen": list(cursor.seen),
         }

tests/test_runner.py

@@ -109,6 +109,26 @@ def started(make_runner, db, **kwargs) -> tuple[Runner, psycopg.Connection]:
     return runner, conn
 
 
+def test_backfill_predicts_recent_events_once(db, make_runner) -> None:
+    ts = datetime.now(timezone.utc) - timedelta(hours=2)
+    with db.admin() as admin:
+        add_event(admin, CHANNEL, ts, "Неисправен")
+
+    # Без бэкфилла первый старт не прогнозирует уже лежащие события.
+    runner, conn = started(make_runner, db, threshold=0.0)
+    runner.tick(conn)
+    assert predictions(db) == []
+
+    runner, conn = started(make_runner, db, threshold=0.0, backfill_hours=3)
+    runner.tick(conn)
+    assert [row[1] for row in predictions(db)] == [str(CHANNEL)]
+
+    # Маркер в runtime: повторный старт окно заново не прогоняет.
+    runner, conn = started(make_runner, db, threshold=0.0, backfill_hours=3)
+    runner.tick(conn)
+    assert len(predictions(db)) == 1
+
+
 def test_tick_writes_prediction_rows(db, make_runner) -> None:
     runner, conn = started(make_runner, db, threshold=0.0)
     ts = datetime.now(timezone.utc) - timedelta(minutes=5)