feat(tradein): scheduled monthly Rosreestr new-quarter poll → ingest alert (#888)

Ops automation — today the ДКП ingest is manual (JOBS hardcoded to 2026Q1);
Q2'26 lands ~Aug 2026. Adds a monthly scheduler job that detects a new
Rosreestr quarter and logs an actionable ingest alert. Detect+alert only —
does NOT download the multi-GB zip or shell out from the tick.

- rosreestr_poll.py: latest_loaded_quarter (MAX deal_date) + next-quarter calc
  + HEAD-probe dataset availability; defensive (network err → False, never raises)
- tasks/rosreestr_quarter_poll.py: task wrapper (mirrors sber_index_pull)
- scheduler.py: trigger_rosreestr_quarter_poll_run + dispatch branch
- migration 096: scrape_schedules monthly seed, ON CONFLICT DO NOTHING

Refs #888, #727, #724

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

@@ -0,0 +1,253 @@
+"""Scheduled poll: detect when a new Rosreestr quarter becomes available (#888).
+
+Граница ответственности модуля:
+  - DETECT: определить, появился ли новый квартал (next after latest loaded) в
+    открытых данных Росреестра.
+  - ALERT: залогировать actionable-сообщение, если квартал доступен.
+  - НЕ скачивает много-гигабайтный ZIP и НЕ вызывает shell-loaders — это ручной
+    ops-шаг. После алерта оператор запускает:
+      data/sql/02_load_all_quarters.sh
+      tradein-mvp/deploy/import-rosreestr.sh
+
+Архив открытых данных Росреестра (сделки ДКП/ДДУ) публикуется на портале:
+  https://rosreestr.gov.ru/opendata/
+
+Файлы именуются по шаблону:
+  dataset_СДЕЛКИ_r-r_01-92_y_{YYYY}_q_{N}.csv.zip
+
+Индексный JSON живёт по адресу:
+  https://rosreestr.gov.ru/opendata/f.json (список всех датасетов)
+
+Проверяем HEAD-запросом (без скачивания) прямой ссылки на следующий квартал.
+Формат ссылки:
+  https://rosreestr.gov.ru/opendata/dataset_СДЕЛКИ_r-r_01-92_y_{YYYY}_q_{N}.csv.zip
+
+При сетевой ошибке / HTTP 5xx — логируем warning, возвращаем available=False.
+Статус 404 → квартал не опубликован → available=False (штатный случай до начала августа).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Базовый URL открытых данных Росреестра (прямая ссылка на ZIP-файл квартала).
+# Шаблон подтверждён по документации портала opendata.rosreestr.gov.ru.
+_ROSREESTR_DATASET_URL_TEMPLATE = (
+    "https://rosreestr.gov.ru/opendata/"
+    "dataset_%D0%A1%D0%94%D0%95%D0%9B%D0%9A%D0%98_r-r_01-92_y_{year}_q_{quarter}.csv.zip"
+)
+
+# Таймаут HEAD-запроса. Росреестр может быть медленным — 15s достаточно.
+_HTTP_TIMEOUT = 15.0
+
+
+def _next_quarter(year: int, quarter: int) -> tuple[int, int]:
+    """Вернуть (year, quarter) для следующего квартала.
+
+    >>> _next_quarter(2026, 1)
+    (2026, 2)
+    >>> _next_quarter(2026, 4)
+    (2027, 1)
+    """
+    if quarter < 4:
+        return year, quarter + 1
+    return year + 1, 1
+
+
+def _period_start_to_quarter(period_start_date: Any) -> tuple[int, int]:
+    """Конвертировать дату начала периода в (year, quarter).
+
+    period_start_date — дата первого числа квартала (e.g. 2026-01-01 → Q1 2026).
+    Принимает datetime.date, datetime.datetime или ISO-строку.
+    """
+    if hasattr(period_start_date, "year"):
+        year = period_start_date.year
+        month = period_start_date.month
+    else:
+        # ISO string fallback
+        import datetime
+
+        d = datetime.date.fromisoformat(str(period_start_date)[:10])
+        year = d.year
+        month = d.month
+    quarter = (month - 1) // 3 + 1
+    return year, quarter
+
+
+def latest_loaded_quarter(db: Session) -> tuple[int, int] | None:
+    """Вернуть (year, quarter) последнего загруженного квартала Росреестра.
+
+    Запрашивает MAX(period_start_date) FROM deals WHERE source='rosreestr'.
+    Возвращает None если таблица пуста или данных от rosreestr нет.
+
+    Использует синхронный Session (SQLAlchemy, psycopg v3 через SQLAlchemy dialect).
+    """
+    row = db.execute(
+        text(
+            """
+            SELECT MAX(deal_date) AS max_date
+            FROM deals
+            WHERE source = 'rosreestr'
+              AND deal_date IS NOT NULL
+            """
+        )
+    ).fetchone()
+
+    if row is None or row[0] is None:
+        logger.info(
+            "rosreestr_poll: no rosreestr deals found in DB — cannot determine latest quarter"
+        )
+        return None
+
+    year, quarter = _period_start_to_quarter(row[0])
+    logger.info(
+        "rosreestr_poll: latest loaded quarter = Q%d %d (derived from max deal_date=%s)",
+        quarter,
+        year,
+        row[0],
+    )
+    return year, quarter
+
+
+def rosreestr_dataset_url(year: int, quarter: int) -> str:
+    """Сформировать URL датасета Росреестра для заданного квартала.
+
+    Возвращает прямую ссылку на ZIP-архив (percent-encoded шаблон).
+    """
+    return _ROSREESTR_DATASET_URL_TEMPLATE.format(year=year, quarter=quarter)
+
+
+async def check_new_quarter_available(
+    client: httpx.AsyncClient,
+    year: int,
+    quarter: int,
+) -> bool:
+    """Проверить, опубликован ли датасет Росреестра для (year, quarter).
+
+    HEAD-запрос к прямой ссылке на ZIP — без скачивания файла.
+
+    Возвращает True если сервер ответил 200 (или 2xx/3xx с redirect).
+    Возвращает False при 404 (квартал ещё не опубликован) или сетевой ошибке.
+    Никогда не поднимает исключения в вызывающий код (scheduler-safe).
+    """
+    url = rosreestr_dataset_url(year, quarter)
+    try:
+        resp = await client.head(url, follow_redirects=True)
+        if resp.status_code == 200:
+            logger.info(
+                "rosreestr_poll: Q%d %d available at %s (HTTP %d)",
+                quarter,
+                year,
+                url,
+                resp.status_code,
+            )
+            return True
+        if resp.status_code == 404:
+            logger.info(
+                "rosreestr_poll: Q%d %d not yet published (HTTP 404)",
+                quarter,
+                year,
+            )
+            return False
+        # Другие коды (403, 5xx, etc.) — логируем как warning, не кидаем
+        logger.warning(
+            "rosreestr_poll: unexpected HTTP %d for Q%d %d url=%s — treating as unavailable",
+            resp.status_code,
+            quarter,
+            year,
+            url,
+        )
+        return False
+    except httpx.TimeoutException:
+        logger.warning(
+            "rosreestr_poll: timeout checking Q%d %d url=%s — treating as unavailable",
+            quarter,
+            year,
+            url,
+        )
+        return False
+    except httpx.RequestError as exc:
+        logger.warning(
+            "rosreestr_poll: network error checking Q%d %d: %s — treating as unavailable",
+            quarter,
+            year,
+            exc,
+        )
+        return False
+    except Exception as exc:
+        logger.warning(
+            "rosreestr_poll: unexpected error checking Q%d %d: %s — treating as unavailable",
+            quarter,
+            year,
+            exc,
+        )
+        return False
+
+
+async def poll_rosreestr_new_quarter(db: Session) -> dict[str, Any]:
+    """Orchestrator: определить следующий квартал после загруженного, проверить наличие.
+
+    Шаги:
+      1. Найти MAX(deal_date) WHERE source='rosreestr' → (loaded_year, loaded_quarter).
+      2. Вычислить next = _next_quarter(loaded_year, loaded_quarter).
+      3. HEAD-запросом проверить наличие датасета на rosreestr.gov.ru.
+      4. Если доступен — логировать actionable INFO с инструкцией по запуску ingest.
+      5. Вернуть dict с результатом.
+
+    Если в БД нет данных rosreestr → предполагаем 2025Q4, проверяем 2026Q1.
+    Это fallback — в штатном режиме данные всегда есть после initial load.
+
+    Возвращает:
+        {"available": bool, "year": int, "quarter": int,
+         "latest_loaded_year": int | None, "latest_loaded_quarter": int | None}
+    """
+    # 1. Текущий максимальный загруженный квартал
+    latest = latest_loaded_quarter(db)
+    if latest is None:
+        # Fallback: нет данных — проверяем 2026Q1 как первую точку
+        loaded_year, loaded_quarter = 2025, 4
+        logger.warning(
+            "rosreestr_poll: no deals in DB — using fallback baseline 2025Q4, "
+            "checking 2026Q1 availability"
+        )
+    else:
+        loaded_year, loaded_quarter = latest
+
+    # 2. Следующий квартал
+    next_year, next_quarter = _next_quarter(loaded_year, loaded_quarter)
+    logger.info(
+        "rosreestr_poll: checking Q%d %d (next after loaded Q%d %d)",
+        next_quarter,
+        next_year,
+        loaded_quarter,
+        loaded_year,
+    )
+
+    # 3. Проверка наличия
+    async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT) as client:
+        available = await check_new_quarter_available(client, next_year, next_quarter)
+
+    # 4. Алерт если доступен
+    if available:
+        logger.info(
+            "rosreestr_poll: NEW QUARTER AVAILABLE — Q%d %d. "
+            "Run ingest to load: "
+            "data/sql/02_load_all_quarters.sh + tradein-mvp/deploy/import-rosreestr.sh",
+            next_quarter,
+            next_year,
+        )
+
+    return {
+        "available": available,
+        "year": next_year,
+        "quarter": next_quarter,
+        "latest_loaded_year": loaded_year if latest is not None else None,
+        "latest_loaded_quarter": loaded_quarter if latest is not None else None,
+    }

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

@@ -32,6 +32,11 @@ Sources:
                               (tasks/sber_index_pull.py, #887; monthly pull of СберИндекс
                               city-level secondary-housing price index — pure HTTP+DB, no
                               anti-bot/proxy; window 05:00-06:00 UTC, ~28-day cadence)
+  - rosreestr_quarter_poll  → run_rosreestr_quarter_poll
+                              (tasks/rosreestr_quarter_poll.py, #888; monthly HEAD-check of
+                              Rosreestr open-data portal for new quarter availability — pure
+                              HTTP detect+alert, NO download/shell-out; window 07:00-08:00 UTC,
+                              ~28-day cadence)
 """
 
 from __future__ import annotations
@@ -636,6 +641,44 @@ async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any])
     return run_id
 
 
+async def trigger_rosreestr_quarter_poll_run(
+    db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+    """Создать scrape_runs + launch run_rosreestr_quarter_poll в asyncio.create_task.
+
+    Monthly HEAD-check (#888): polls rosreestr.gov.ru for the next quarter's open-data
+    ZIP to detect when a new quarter is published. Pure HTTP detect+alert — does NOT
+    download the multi-GB dataset or shell out to the loaders. When available=True, a
+    clear actionable INFO log is emitted for the operator.
+
+    Mirrors trigger_sber_index_pull_run: claim run → create_task → mark_done/failed
+    delegated to tasks/rosreestr_quarter_poll.py.
+
+    Returns run_id или None (skip — already running).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.rosreestr_quarter_poll import run_rosreestr_quarter_poll
+
+            await run_rosreestr_quarter_poll(run_db, run_id=run_id, params=params)
+        except Exception:
+            logger.exception("scheduler: run_rosreestr_quarter_poll crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered rosreestr_quarter_poll run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
 
@@ -958,6 +1001,8 @@ async def scheduler_loop() -> None:
                         await trigger_deactivate_stale_avito_run(db, sch)
                     elif source == "sber_index_pull":
                         await trigger_sber_index_pull_run(db, sch)
+                    elif source == "rosreestr_quarter_poll":
+                        await trigger_rosreestr_quarter_poll_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

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

@@ -0,0 +1,102 @@
+"""Scheduled task: monthly poll of Rosreestr open-data portal for a new quarter (#888).
+
+Delegates detect+alert logic to app.services.rosreestr_poll.poll_rosreestr_new_quarter().
+Wired into the in-app scheduler as source='rosreestr_quarter_poll'. Triggered monthly via
+the scrape_schedules seed (migration 096).
+
+SCOPE: detect availability + log actionable alert ONLY.
+Does NOT download the multi-GB ZIP or shell out to the loaders.
+When available=True, the operator runs manually:
+  data/sql/02_load_all_quarters.sh
+  tradein-mvp/deploy/import-rosreestr.sh
+
+Mirrors tasks/sber_index_pull.py: thin async wrapper around the service,
+heartbeat → delegate → mark_done/mark_failed.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+from app.services.rosreestr_poll import poll_rosreestr_new_quarter
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "RosreestrQuarterPollResult",
+    "run_rosreestr_quarter_poll",
+]
+
+
+@dataclass
+class RosreestrQuarterPollResult:
+    """Result from a scheduled Rosreestr quarter-poll run."""
+
+    available: bool = False
+    year: int = 0
+    quarter: int = 0
+    duration_sec: float = field(default=0.0)
+
+
+async def run_rosreestr_quarter_poll(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,  # type: ignore[type-arg]
+) -> RosreestrQuarterPollResult:
+    """Execute Rosreestr quarter poll with run lifecycle management.
+
+    Wraps poll_rosreestr_new_quarter(), emitting heartbeat before the check
+    and marking the scrape_run as done/failed on completion.
+
+    Params (from default_params jsonb in scrape_schedules):
+        interval_days: int — informational only (scheduler uses next_run_at arithmetic).
+    """
+    counters: dict[str, int] = {
+        "available": 0,
+        "year": 0,
+        "quarter": 0,
+    }
+    started_at = time.monotonic()
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        result_dict = await poll_rosreestr_new_quarter(db)
+
+        available = bool(result_dict.get("available", False))
+        year = int(result_dict.get("year", 0))
+        quarter = int(result_dict.get("quarter", 0))
+        duration_sec = time.monotonic() - started_at
+
+        counters = {
+            "available": int(available),
+            "year": year,
+            "quarter": quarter,
+            "duration_sec": int(duration_sec),
+        }
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "rosreestr_quarter_poll run_id=%d done — available=%s Q%d %d %.1fs",
+            run_id,
+            available,
+            quarter,
+            year,
+            duration_sec,
+        )
+        return RosreestrQuarterPollResult(
+            available=available,
+            year=year,
+            quarter=quarter,
+            duration_sec=duration_sec,
+        )
+
+    except Exception as exc:
+        logger.exception("rosreestr_quarter_poll run_id=%d failed", run_id)
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/096_scrape_schedules_seed_rosreestr_quarter_poll.sql

@@ -0,0 +1,77 @@
+-- 096_scrape_schedules_seed_rosreestr_quarter_poll.sql
+-- Issue #888 — seed scrape_schedules row for monthly Rosreestr new-quarter poll.
+--
+-- Context:
+--   app/services/rosreestr_poll.py checks the Rosreestr open-data portal for the
+--   next unpublished quarter via an HTTP HEAD request (no download). When the dataset
+--   ZIP becomes available, a clear actionable INFO log is emitted for the operator.
+--
+--   Dataset filename pattern:
+--     dataset_СДЕЛКИ_r-r_01-92_y_{YYYY}_q_{N}.csv.zip
+--   Published at: https://rosreestr.gov.ru/opendata/
+--
+--   Ingest (operator-run after detection, NOT automated):
+--     data/sql/02_load_all_quarters.sh
+--     tradein-mvp/deploy/import-rosreestr.sh
+--
+--   Components deployed together:
+--     1. services/rosreestr_poll.py — latest_loaded_quarter() + poll_rosreestr_new_quarter()
+--     2. tasks/rosreestr_quarter_poll.py  — run_rosreestr_quarter_poll() task wrapper
+--     3. scheduler.py — trigger_rosreestr_quarter_poll_run() + dispatch branch
+--        `elif source == "rosreestr_quarter_poll"`
+--     4. This migration — seeds the scrape_schedules row (enabled=true, monthly window)
+--
+-- Schedule window 07:00-08:00 UTC:
+--   Monthly cadence, placed after sber_index_pull (05:00-06:00 UTC) and
+--   rosreestr_dkp_import (04:00-06:00 UTC) — morning batch slot.
+--   The Rosreestr open-data portal publishes new quarters roughly once per quarter
+--   (Q2 data lands ~August). Monthly polling is sufficient to catch publication within
+--   a few weeks of release without wasting resources on daily checks.
+--   default_params.interval_days=28 follows the same monthly-cadence pattern as
+--   sber_index_pull (093) and avito/yandex monthly seeds.
+--
+-- next_run_at bootstrapped to tomorrow 07:00 UTC — scheduler will not fire immediately
+-- on deploy (same pattern as 078/079/082/088/091/093).
+--
+-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
+--
+-- Dependencies:
+--   052_scrape_schedules.sql (table + UNIQUE(source))
+--   scheduler.py change (trigger_rosreestr_quarter_poll_run must be deployed)
+--   services/rosreestr_poll.py + tasks/rosreestr_quarter_poll.py must be deployed
+--
+-- Deploy order:
+--   Apply after deploying the scheduler.py + task + service changes so the scheduler
+--   can dispatch 'rosreestr_quarter_poll' correctly on first fire.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'rosreestr_quarter_poll',
+    true,                                        -- SAFE: pure HTTP HEAD + DB read, no anti-bot
+    7,                                           -- window 07:00-08:00 UTC
+    8,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
+    '{"interval_days": 28}'::jsonb               -- monthly cadence (~28 days between runs)
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). '
+    'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
+    'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), '
+    'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), '
+    'yandex_address_backfill (#855, EKB pilot), '
+    'sber_index_pull (#887, monthly СберИндекс city-level price index), '
+    'rosreestr_quarter_poll (#888, monthly Rosreestr new-quarter availability check).';
+
+COMMIT;

tradein-mvp/backend/tests/test_rosreestr_poll.py

@@ -0,0 +1,243 @@
+"""Unit tests for rosreestr_poll service (#888).
+
+Coverage:
+  (a) _next_quarter: boundary cases (Q4 → Q1 of next year, mid-year).
+  (b) latest_loaded_quarter: derives correct (year, quarter) from a known deal_date.
+  (c) poll_rosreestr_new_quarter available=True path (mocked httpx + fake DB).
+  (d) poll_rosreestr_new_quarter available=False path.
+  (e) Network error → available=False (never raises into caller).
+  (f) No DB data → fallback baseline 2025Q4, checks 2026Q1.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from datetime import date
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+# WeasyPrint stub — not installed in CI without GTK
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest  # noqa: E402
+
+from app.services.rosreestr_poll import (  # noqa: E402
+    _next_quarter,
+    _period_start_to_quarter,
+    check_new_quarter_available,
+    latest_loaded_quarter,
+    poll_rosreestr_new_quarter,
+    rosreestr_dataset_url,
+)
+
+# ---------------------------------------------------------------------------
+# (a) _next_quarter
+# ---------------------------------------------------------------------------
+
+
+def test_next_quarter_mid_year() -> None:
+    assert _next_quarter(2026, 1) == (2026, 2)
+    assert _next_quarter(2026, 2) == (2026, 3)
+    assert _next_quarter(2026, 3) == (2026, 4)
+
+
+def test_next_quarter_year_rollover() -> None:
+    assert _next_quarter(2026, 4) == (2027, 1)
+    assert _next_quarter(2024, 4) == (2025, 1)
+
+
+def test_period_start_to_quarter_from_date() -> None:
+    assert _period_start_to_quarter(date(2026, 1, 1)) == (2026, 1)
+    assert _period_start_to_quarter(date(2026, 4, 1)) == (2026, 2)
+    assert _period_start_to_quarter(date(2026, 7, 15)) == (2026, 3)
+    assert _period_start_to_quarter(date(2026, 10, 31)) == (2026, 4)
+    assert _period_start_to_quarter(date(2025, 12, 31)) == (2025, 4)
+
+
+def test_period_start_to_quarter_from_string() -> None:
+    assert _period_start_to_quarter("2026-01-15") == (2026, 1)
+    assert _period_start_to_quarter("2026-07-01") == (2026, 3)
+
+
+def test_rosreestr_dataset_url_contains_year_quarter() -> None:
+    url = rosreestr_dataset_url(2026, 2)
+    assert "2026" in url
+    assert "_2" in url or "q_2" in url  # quarter embedded
+
+
+# ---------------------------------------------------------------------------
+# Fake DB helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeRow:
+    """Mimics a DB row with __getitem__."""
+
+    def __init__(self, val: object) -> None:
+        self._val = val
+
+    def __getitem__(self, idx: int) -> object:
+        return self._val
+
+
+class _FakeDB:
+    """Minimal fake Session returning a configurable deal_date via execute().fetchone()."""
+
+    def __init__(self, max_deal_date: date | None) -> None:
+        self._row = _FakeRow(max_deal_date) if max_deal_date is not None else None
+
+    def execute(self, stmt: object, params: object = None) -> MagicMock:
+        mock = MagicMock()
+        mock.fetchone.return_value = self._row
+        return mock
+
+    def commit(self) -> None:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# (b) latest_loaded_quarter
+# ---------------------------------------------------------------------------
+
+
+def test_latest_loaded_quarter_returns_correct_year_quarter() -> None:
+    """Q1 2026 deal_date → (2026, 1)."""
+    db = _FakeDB(date(2026, 1, 31))
+    result = latest_loaded_quarter(db)  # type: ignore[arg-type]
+    assert result == (2026, 1)
+
+
+def test_latest_loaded_quarter_q4() -> None:
+    """Dec deal_date → Q4."""
+    db = _FakeDB(date(2025, 12, 15))
+    result = latest_loaded_quarter(db)  # type: ignore[arg-type]
+    assert result == (2025, 4)
+
+
+def test_latest_loaded_quarter_returns_none_when_empty() -> None:
+    """No rows → None."""
+    db = _FakeDB(None)
+    result = latest_loaded_quarter(db)  # type: ignore[arg-type]
+    assert result is None
+
+
+# ---------------------------------------------------------------------------
+# (c) poll_rosreestr_new_quarter — available=True
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_poll_available_true_returns_correct_next_quarter() -> None:
+    """If latest loaded is Q1 2026 and head returns 200, returns available=True for Q2 2026."""
+    db = _FakeDB(date(2026, 3, 31))  # latest = Q1 2026
+
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+
+    mock_client = AsyncMock()
+    mock_client.head.return_value = mock_response
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("app.services.rosreestr_poll.httpx.AsyncClient", return_value=mock_client):
+        result = await poll_rosreestr_new_quarter(db)  # type: ignore[arg-type]
+
+    assert result["available"] is True
+    assert result["year"] == 2026
+    assert result["quarter"] == 2
+    assert result["latest_loaded_year"] == 2026
+    assert result["latest_loaded_quarter"] == 1
+
+
+# ---------------------------------------------------------------------------
+# (d) poll_rosreestr_new_quarter — available=False (404)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_poll_available_false_on_404() -> None:
+    """404 from server → available=False."""
+    db = _FakeDB(date(2026, 3, 15))  # Q1 2026
+
+    mock_response = MagicMock()
+    mock_response.status_code = 404
+
+    mock_client = AsyncMock()
+    mock_client.head.return_value = mock_response
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("app.services.rosreestr_poll.httpx.AsyncClient", return_value=mock_client):
+        result = await poll_rosreestr_new_quarter(db)  # type: ignore[arg-type]
+
+    assert result["available"] is False
+    assert result["year"] == 2026
+    assert result["quarter"] == 2
+
+
+# ---------------------------------------------------------------------------
+# (e) Network error → available=False (never raises into caller)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_poll_network_error_returns_false_no_raise() -> None:
+    """Network error → available=False, no exception propagated."""
+    import httpx
+
+    db = _FakeDB(date(2026, 1, 20))  # Q1 2026
+
+    mock_client = AsyncMock()
+    mock_client.head.side_effect = httpx.ConnectError("simulated network error")
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("app.services.rosreestr_poll.httpx.AsyncClient", return_value=mock_client):
+        # Must not raise
+        result = await poll_rosreestr_new_quarter(db)  # type: ignore[arg-type]
+
+    assert result["available"] is False
+
+
+@pytest.mark.asyncio
+async def test_check_new_quarter_available_timeout_returns_false() -> None:
+    """Timeout → returns False, no exception."""
+    import httpx
+
+    mock_client = AsyncMock()
+    mock_client.head.side_effect = httpx.TimeoutException("timeout")
+
+    result = await check_new_quarter_available(mock_client, 2026, 2)  # type: ignore[arg-type]
+    assert result is False
+
+
+# ---------------------------------------------------------------------------
+# (f) No DB data → fallback baseline checks 2026Q1
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_poll_fallback_when_no_db_data() -> None:
+    """No rosreestr deals in DB → fallback 2025Q4 baseline, checks 2026Q1."""
+    db = _FakeDB(None)
+
+    mock_response = MagicMock()
+    mock_response.status_code = 404  # not yet available
+
+    mock_client = AsyncMock()
+    mock_client.head.return_value = mock_response
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("app.services.rosreestr_poll.httpx.AsyncClient", return_value=mock_client):
+        result = await poll_rosreestr_new_quarter(db)  # type: ignore[arg-type]
+
+    # Fallback: 2025Q4 → check 2026Q1
+    assert result["year"] == 2026
+    assert result["quarter"] == 1
+    # latest_loaded_year/quarter should be None (no DB data)
+    assert result["latest_loaded_year"] is None
+    assert result["latest_loaded_quarter"] is None