Merge pull request 'feat(tradein): sber-index freshness monitor (п.1 observability)' (#2503) from fix/tradein-sber-freshness-monitor into main

tradein-mvp/backend/app/services/product_handlers.py

@@ -193,6 +193,16 @@ async def _job_deals_freshness_monitor(
     await loop.run_in_executor(None, check_deals_freshness, db, run_id, params)
 
 
+# ── sber_freshness_monitor — sync DB-only freshness check в executor ──────────
+async def _job_sber_freshness_monitor(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.sber_freshness_monitor import check_sber_freshness
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, check_sber_freshness, db, run_id, params)
+
+
 # ── newbuilding_enrich — async, owns lifecycle ───────────────────────────────
 async def _job_newbuilding_enrich(
     db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@@ -394,6 +404,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
         "sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"),
         "rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"),
         "deals_freshness_monitor": Handler(_job_deals_freshness_monitor, "deals_freshness_monitor"),
+        "sber_freshness_monitor": Handler(_job_sber_freshness_monitor, "sber_freshness_monitor"),
         "newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"),
         "yandex_newbuilding_sweep": Handler(
             _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"

tradein-mvp/backend/app/tasks/sber_freshness_monitor.py

@@ -0,0 +1,196 @@
+"""Мониторинг свежести ДАННЫХ СберИндекса (не статуса джобы) — audit п.1.
+
+Проблема аудита: estimator._load_sber_index_series (#794/#audit-5a) применяет
+СберИндекс time-adjustment к ДКП-сделкам и лишь ЛОГИРУЕТ per-estimate warning,
+когда latest месяц серии старее settings.sber_index_max_age_days (35д). Джоба
+`sber_index_pull` крутится ежемесячно (enabled), а источник СберИндекса публикует
+данные с лагом ~1-2 месяца, поэтому `sber_price_index.period_month` дрейфит
+(на 2026-07-12 latest=2026-05-01, ~72д). Это НЕ silent failure, но staleness
+видна только в debug-подобном per-estimate warning'е, тонущем в логах оценок.
+
+Этот монитор смотрит на `max(period_month)` вторичного сегмента по региону и
+поднимает per-day WARNING-алерт, когда данные устарели СВЕРХ допустимого лага
+публикации — так ops видит дрейф на MONITOR-частоте, а не по крупицам в логах.
+
+Порог алерта (документирование выбора):
+  Per-estimate guard (estimator): age > settings.sber_index_max_age_days (35д).
+  Монитор:                        age > sber_index_max_age_days + lag_allowance.
+  lag_allowance (DEFAULT_LAG_ALLOWANCE_DAYS=25) — запас на ИНХЕРЕНТНЫЙ лаг
+  публикации СберИндекса: источник отстаёт на 1-2 месяца, period_month — лейбл
+  ПЕРВОГО числа месяца, а месячный pull ещё не подтянул новейший период. Итог:
+  35 + 25 = 60д. Ниже 60д latest считается «нормально отстающим» → алерта нет
+  (иначе daily-шум на штатном лаге). Выше 60д данные застряли сверх ~2 месяцев
+  → алерт. Проверено на проде 2026-07-12: max=2026-05-01, age=72д > 60 → alert=1.
+
+Задача синхронная (DB-only, один SELECT max(period_month)) — запускается
+kit-scheduler'ом через product_handlers._job_sber_freshness_monitor в
+run_in_executor, по образцу deals_freshness_monitor. Вердикт вычисляет ЧИСТАЯ
+функция evaluate_sber_freshness() (frozen-now тестируется без БД).
+
+Прогон НЕ помечается failed при алерте (это МОНИТОР, а не сбой джобы) — WARNING
+достаточен. mark_failed только если sber_price_index недоступна/пуста (нечего
+оценивать).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import UTC, date, datetime
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "DEFAULT_LAG_ALLOWANCE_DAYS",
+    "SberFreshnessVerdict",
+    "check_sber_freshness",
+    "evaluate_sber_freshness",
+]
+
+# Запас на инхерентный лаг публикации СберИндекса (дней) СВЕРХ per-estimate
+# guard'а settings.sber_index_max_age_days. Читается из default_params.lag_allowance_days.
+DEFAULT_LAG_ALLOWANCE_DAYS = 25
+
+# Регион продукта (Trade-in — Свердловская область). Совпадает с city-значениями
+# sber_price_index для областного уровня.
+SBER_MONITOR_CITY = "Свердловская область"
+
+_LATEST_SBER_PERIOD_SQL = text("""
+    SELECT max(period_month) AS latest
+    FROM sber_price_index
+    WHERE city = CAST(:city AS text)
+      -- #R2-H1: только вторичный рынок (эстиматор — вторичка); первичка
+      -- (новостройки) = направленно неверная коррекция. Зеркалит фильтр
+      -- estimator._load_sber_index_series.
+      AND (segment IS NULL OR segment ILIKE '%вторичн%')
+""")
+
+
+@dataclass(frozen=True)
+class SberFreshnessVerdict:
+    """Вердикт свежести СберИндекса по max(period_month)."""
+
+    latest_period: date
+    age_days: int
+    stale: bool
+
+
+def evaluate_sber_freshness(
+    latest_period: date,
+    now: datetime,
+    max_age_days: int,
+) -> SberFreshnessVerdict:
+    """Чистая логика: устарел ли latest период СберИндекса.
+
+    stale = age_days > max_age_days, где age_days = now.date() - latest_period.
+    `max_age_days` — ПОЛНЫЙ порог монитора (per-estimate guard + lag_allowance),
+    вычисляется вызывающим check_sber_freshness. Тестируется с frozen `now` без БД.
+    """
+    age_days = (now.date() - latest_period).days
+    stale = age_days > max_age_days
+    return SberFreshnessVerdict(
+        latest_period=latest_period,
+        age_days=age_days,
+        stale=stale,
+    )
+
+
+def check_sber_freshness(
+    db: Session,
+    run_id: int,
+    params: dict | None = None,  # type: ignore[type-arg]
+    now: datetime | None = None,
+) -> dict[str, int]:
+    """Проверить свежесть СберИндекса по max(period_month) и алертить при staleness.
+
+    Sync (вызывается scheduler-триггером в executor, как check_deals_freshness).
+    Читает один SELECT max(period_month) вторичного сегмента по региону, считает
+    вердикт чистой функцией, логирует WARNING при stale (per-day surfacing для ops)
+    и финализирует run.
+
+    Params (default_params jsonb):
+        lag_allowance_days: int — запас сверх sber_index_max_age_days (default 25).
+    `now` инъектируется в тестах (frozen); в проде — None → datetime.now(UTC).
+
+    Returns counters {latest_year, latest_month, age_days, alert}.
+    mark_failed только если sber_price_index пуста/недоступна (нечего оценивать);
+    при алерте прогон помечается done (это монитор, не сбой джобы).
+    """
+    params = params or {}
+    now = now or datetime.now(UTC)
+    counters: dict[str, int] = {
+        "latest_year": 0,
+        "latest_month": 0,
+        "age_days": 0,
+        "alert": 0,
+    }
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        row = db.execute(_LATEST_SBER_PERIOD_SQL, {"city": SBER_MONITOR_CITY}).first()
+        latest: date | None = row.latest if row is not None else None
+        if latest is None:
+            logger.warning(
+                "sber freshness: sber_price_index пуст/недоступен для region=%s "
+                "(вторичка) — оценить свежесть нельзя",
+                SBER_MONITOR_CITY,
+            )
+            runs_mod.mark_failed(db, run_id, "sber_price_index empty or unavailable", counters)
+            return counters
+
+        lag_days = int(params.get("lag_allowance_days", DEFAULT_LAG_ALLOWANCE_DAYS))
+        max_age_days = settings.sber_index_max_age_days + lag_days
+        verdict = evaluate_sber_freshness(latest, now, max_age_days)
+
+        counters = {
+            "latest_year": latest.year,
+            "latest_month": latest.month,
+            "age_days": verdict.age_days,
+            "alert": int(verdict.stale),
+        }
+
+        if verdict.stale:
+            logger.warning(
+                "sber freshness: max(period_month)=%s устарел на %d дней "
+                "(> порога %d = sber_index_max_age_days %d + lag %d); "
+                "СберИндекс time-adjustment ДКП-сделок мог отстать — "
+                "проверь sber_index_pull и доступность новых периодов источника",
+                latest,
+                verdict.age_days,
+                max_age_days,
+                settings.sber_index_max_age_days,
+                lag_days,
+            )
+        else:
+            logger.info(
+                "sber freshness: max(period_month)=%s свежий (age=%d дней ≤ порога %d) "
+                "region=%s — алерта нет",
+                latest,
+                verdict.age_days,
+                max_age_days,
+                SBER_MONITOR_CITY,
+            )
+
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "check_sber_freshness run_id=%d done: latest=%s alert=%d age_days=%d",
+            run_id,
+            latest,
+            counters["alert"],
+            counters["age_days"],
+        )
+        return counters
+    except Exception as exc:
+        logger.exception("check_sber_freshness run_id=%d failed", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/180_seed_sber_freshness_monitor.sql

@@ -0,0 +1,75 @@
+-- 180_seed_sber_freshness_monitor.sql
+-- Audit п.1 — seed scrape_schedules row for the daily СберИндекс data-freshness monitor.
+--
+-- Context (verified on prod 2026-07-12):
+--   estimator._load_sber_index_series (#794/#audit-5a) applies a СберИндекс time-adjustment
+--   to ДКП deals and only LOGS a per-estimate warning when the latest series month is older
+--   than settings.sber_index_max_age_days (35d). The `sber_index_pull` job runs monthly
+--   (enabled), but СберИндекс publishes with an inherent ~1-2 month lag, so
+--   sber_price_index.period_month drifts. Current max(period_month) for the oblast secondary
+--   segment = 2026-05-01 (~72d old). NOT a silent failure, but the staleness is buried in
+--   per-estimate warnings — invisible at monitor cadence.
+--
+--   Audit gap: ops has no monitor-frequency signal for СберИндекс staleness. This monitor
+--   watches max(period_month) directly and raises a per-day WARNING when the data is stale
+--   beyond the allowed publication lag.
+--
+--   Staleness threshold (app/tasks/sber_freshness_monitor.py::evaluate_sber_freshness):
+--     age_days      = now() - max(period_month)
+--     max_age_days  = settings.sber_index_max_age_days (35) + lag_allowance_days (25) = 60
+--     stale         = age_days > max_age_days
+--   The +25 lag_allowance covers the INHERENT СберИндекс publication lag (source is 1-2mo
+--   behind; period_month is the first-of-month label; the monthly pull has not yet caught the
+--   newest period). Below 60d the latest period is treated as normally-lagging -> NO alert
+--   (avoids daily noise on штатный lag). Above 60d the data is stuck beyond ~2 months -> alert.
+--   At 2026-07-12: max=2026-05-01, age=72d > 60 -> alert=1 (surfaces the current drift).
+--
+--   Components deployed together:
+--     1. app/tasks/sber_freshness_monitor.py — check_sber_freshness() + evaluate_sber_freshness()
+--     2. app/services/product_handlers.py — kit Handler 'sber_freshness_monitor'
+--     3. This migration — seeds the scrape_schedules row (enabled=true, daily window)
+--
+-- Schedule window 09:00-10:00 UTC:
+--   Daily cadence, placed AFTER deals_freshness_monitor (08:00-09:00 UTC) so the two
+--   data-freshness monitors do not overlap. Cheap: one SELECT max(period_month), no external
+--   HTTP — safe to run daily even though sber_index_pull itself is monthly.
+--
+-- default_params.lag_allowance_days = 25:
+--   Allowance for СберИндекс publication lag on top of the 35d per-estimate guard (-> 60d).
+--
+-- next_run_at bootstrapped to tomorrow 09:00 UTC — scheduler will not fire immediately
+-- on deploy (same pattern as 162/093/096/160).
+--
+-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
+--
+-- Dependencies:
+--   052_scrape_schedules.sql (table + UNIQUE(source))
+--   sber_price_index table (populated by app.tasks.sber_index_pull)
+--   app/services/product_handlers.py + tasks/sber_freshness_monitor.py deployed
+--
+-- Deploy order:
+--   Apply after deploying the product_handlers.py + task changes so the dispatch can resolve
+--   'sber_freshness_monitor' correctly on first fire.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'sber_freshness_monitor',
+    true,                                        -- SAFE: single SELECT max(period_month), no ext calls
+    9,                                           -- window 09:00-10:00 UTC (after deals_freshness_monitor)
+    10,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 9)) AT TIME ZONE 'UTC',
+    '{"lag_allowance_days": 25}'::jsonb          -- publication-lag allowance over the 35d per-estimate guard
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/tests/test_sber_freshness_monitor.py

@@ -0,0 +1,227 @@
+"""Freshness-монитор данных СберИндекса по max(period_month) — audit п.1.
+
+Покрывает:
+  1. Чистую логику evaluate_sber_freshness (frozen now, без БД):
+     - fresh:  age <= max_age_days (алерта нет);
+     - stale:  age > max_age_days (алерт);
+     - граница порога (== max_age_days → нет алерта; +1 день → алерт).
+  2. check_sber_freshness с FakeDB (fresh / stale / empty→mark_failed / кастомный lag).
+  3. Свойства миграции 180 (по образцу test_deals_freshness_monitor).
+  4. Регистрацию в kit product_handlers (registry остаётся зелёным).
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.core.config import settings
+from app.services.product_handlers import build_product_handlers
+from app.tasks import sber_freshness_monitor as mon
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_180 = _SQL_DIR / "180_seed_sber_freshness_monitor.sql"
+
+# max(period_month) вторичного сегмента = 2026-05-01 (проверено на проде 2026-07-12).
+_MAY_2026 = date(2026, 5, 1)
+
+
+# ── evaluate_sber_freshness (чистая логика, frozen now) ───────────────────────
+
+
+def _now(y: int, m: int, d: int) -> datetime:
+    return datetime(y, m, d, tzinfo=UTC)
+
+
+def test_fresh_within_max_age() -> None:
+    """age=30 ≤ max_age_days=60 — свежий, алерта нет."""
+    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 5, 31), max_age_days=60)
+    assert v.stale is False
+    assert v.age_days == 30
+    assert v.latest_period == _MAY_2026
+
+
+def test_stale_beyond_max_age() -> None:
+    """Прод-состояние: 2026-05-01 @ 2026-07-12 — age=72 > 60 → алерт."""
+    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 7, 12), max_age_days=60)
+    assert v.stale is True
+    assert v.age_days == 72
+
+
+def test_threshold_boundary_exact_no_alert() -> None:
+    """Ровно на пороге (age == max_age_days) алерта ещё нет (строгое >)."""
+    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 6, 30), max_age_days=60)
+    assert v.age_days == 60
+    assert v.stale is False
+
+
+def test_threshold_boundary_next_day_alert() -> None:
+    """Порог + 1 день (age=61) — первый алерт."""
+    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 7, 1), max_age_days=60)
+    assert v.age_days == 61
+    assert v.stale is True
+
+
+# ── check_sber_freshness (FakeDB) ─────────────────────────────────────────────
+
+
+class _Row:
+    def __init__(self, latest: date | None) -> None:
+        self.latest = latest
+
+
+class _FakeResult:
+    def __init__(self, latest: date | None) -> None:
+        self._latest = latest
+
+    def first(self) -> _Row:
+        return _Row(self._latest)
+
+
+class _FakeDB:
+    def __init__(self, latest: date | None) -> None:
+        self._latest = latest
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        return _FakeResult(self._latest)
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+
+def _patch_runs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
+    calls: dict[str, Any] = {"done": None, "failed": None, "heartbeat": 0}
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "update_heartbeat",
+        lambda *a, **k: calls.__setitem__("heartbeat", calls["heartbeat"] + 1),
+    )
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: calls.__setitem__("done", dict(counters)),
+    )
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: calls.__setitem__("failed", err),
+    )
+    return calls
+
+
+def test_check_fresh_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(_MAY_2026)
+    # @2026-05-31: age=30 ≤ 35+25=60 → нет алерта.
+    out = mon.check_sber_freshness(db, run_id=1, params={}, now=_now(2026, 5, 31))  # type: ignore[arg-type]
+    assert out == {"latest_year": 2026, "latest_month": 5, "age_days": 30, "alert": 0}
+    assert calls["done"] == out
+    assert calls["failed"] is None
+
+
+def test_check_stale_marks_done_with_alert(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(_MAY_2026)
+    # @2026-07-12 (прод): age=72 > 60 → алерт.
+    out = mon.check_sber_freshness(db, run_id=2, params={}, now=_now(2026, 7, 12))  # type: ignore[arg-type]
+    assert out["alert"] == 1
+    assert out["age_days"] == 72
+    assert out["latest_month"] == 5
+    # Монитор НЕ падает при алерте — прогон done, а не failed.
+    assert calls["done"] == out
+    assert calls["failed"] is None
+
+
+def test_check_empty_index_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(None)
+    out = mon.check_sber_freshness(db, run_id=3, params={}, now=_now(2026, 7, 12))  # type: ignore[arg-type]
+    assert out["alert"] == 0
+    assert calls["done"] is None
+    assert calls["failed"] is not None
+
+
+def test_check_reads_lag_from_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch_runs(monkeypatch)
+    db = _FakeDB(_MAY_2026)
+    # lag=0 → порог = sber_index_max_age_days (35) → @2026-07-12 (age=72) просрочено.
+    out = mon.check_sber_freshness(
+        db, run_id=4, params={"lag_allowance_days": 0}, now=_now(2026, 7, 12)
+    )  # type: ignore[arg-type]
+    assert out["alert"] == 1
+
+
+def test_check_default_threshold_uses_setting_plus_lag(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Дефолтный порог = sber_index_max_age_days + DEFAULT_LAG_ALLOWANCE_DAYS."""
+    _patch_runs(monkeypatch)
+    db = _FakeDB(_MAY_2026)
+    threshold = settings.sber_index_max_age_days + mon.DEFAULT_LAG_ALLOWANCE_DAYS
+    # Ровно на пороге (age == threshold) — алерта нет; +1 день — алерт.
+    exact = datetime(2026, 5, 1, tzinfo=UTC) + timedelta(days=threshold)
+    out_exact = mon.check_sber_freshness(db, run_id=5, params={}, now=exact)  # type: ignore[arg-type]
+    assert out_exact["age_days"] == threshold
+    assert out_exact["alert"] == 0
+    out_over = mon.check_sber_freshness(db, run_id=6, params={}, now=exact + timedelta(days=1))  # type: ignore[arg-type]
+    assert out_over["alert"] == 1
+
+
+# ── Миграция 180 ──────────────────────────────────────────────────────────────
+
+
+def test_migration_180_exists() -> None:
+    assert _MIGRATION_180.is_file(), f"missing migration: {_MIGRATION_180}"
+
+
+def test_migration_180_seeds_source() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert "'sber_freshness_monitor'" in sql
+
+
+def test_migration_180_enabled_true() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert "true" in sql
+
+
+def test_migration_180_is_idempotent() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+
+
+def test_migration_180_is_transactional() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_180_window_9_to_10_utc() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert re.search(r"\b9\b", sql), "window_start_hour 9 missing"
+    assert re.search(r"\b10\b", sql), "window_end_hour 10 missing"
+
+
+def test_migration_180_lag_allowance_25() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert "lag_allowance_days" in sql
+    assert "25" in sql
+
+
+def test_migration_180_no_psycopg_trap() -> None:
+    sql = _MIGRATION_180.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+# ── Регистрация в kit registry ─────────────────────────────────────────────────
+
+
+def test_kit_product_handler_registered() -> None:
+    # ctx не нужен для сборки dict ключей — build_product_handlers его не замыкает.
+    handlers = build_product_handlers(ctx=None)  # type: ignore[arg-type]
+    assert "sber_freshness_monitor" in handlers