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

"""Мониторинг свежести ДАННЫХ СберИндекса (не статуса джобы) — 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 ERROR-алерт, когда данные устарели СВЕРХ допустимого лага
публикации — так ops видит дрейф на MONITOR-частоте, а не по крупицам в логах.

#2674 — почему ERROR, а не WARNING. В контейнере скрапера GlitchTip поднят с
LoggingIntegration(event_level=ERROR) (scheduler_main.py), поэтому WARNING
событием НЕ становится вообще. Бенчмарк цен участвует в сверке наших медиан, его
застой — сбой, а не наблюдение. Сосед по конструкции (deals_freshness_monitor)
писал ERROR с самого начала — расходилась только эта джоба.

ВАЖНО про «9 срабатываний» из #2674 (ревью PR #2681, прод-разбор всех 24 прогонов
монитора 2026-08-06). Эти девять НЕ были застоем бенчмарка — это была ПИЛА нашего
собственного такта загрузки:
  13-16.07  alert=1  age 73..76  latest=май    01-05.08  alert=1  age 61..65
  17.07     alert=0  age 46      latest=июнь   (день загрузки)
Загрузка ходила раз в 28 дней и приносила период на месяц новее, возраст же
считается от ПЕРВОГО числа покрытого месяца → пол ~46 в момент загрузки, потолок
46+28=74, порог 60 ВНУТРИ диапазона, тревога 14 суток из 28 каждый цикл. Поднимать
такое до ERROR без починки такта значило бы завести ежедневное ложное событие на
две недели в месяц. Поэтому миграция 212 перевела sber_index_pull на НЕДЕЛЬНЫЙ
такт: потолок возраста ≈ пол+7 ≈ 53 при пороге 60, тревога снова означает
«источник/загрузка встали», а не «мы давно не ходили».

Порог алерта (документирование выбора):
  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 — лейбл
  ПЕРВОГО числа месяца, поэтому даже свежайшая загрузка даёт возраст ~46 суток.
  Итог: 35 + 25 = 60д. При недельном такте (миграция 212) рабочий диапазон возраста
  ~46..53 — до порога остаётся ~7 суток запаса: один пропущенный недельный цикл
  поглощается, два подряд дают тревогу. Порог НЕ должен снова оказаться внутри
  рабочего диапазона — если такт загрузки будут менять, пересчитай потолок
  (пол + interval_days) и сверь с 60.

Задача синхронная (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 при алерте (это МОНИТОР, а не сбой джобы) — ERROR-записи
достаточно. 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:
            # ERROR (#2674): монитор не может выполнить свою работу вовсе — это сбой,
            # а не наблюдение. mark_failed ниже виден только стрик-алерту (3 подряд),
            # а монитор ходит раз в сутки — три дня молчания на пустом бенчмарке.
            logger.error(
                "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:
            # ERROR (#2674): WARNING не долетает до GlitchTip (event_level=ERROR) —
            # 9 срабатываний на проде дали ноль событий. См. докстринг модуля.
            logger.error(
                "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