tradein-mvp/backend/app/core/config.py

@@ -90,6 +90,24 @@ class Settings(BaseSettings):
     asking_to_sold_haircut: float = 0.05  # дефолтная asking→sold скидка (banded по ppm²)
     estimate_fsd_k: float = 1.65  # множитель FSD → полуширина диапазона
 
+    # ── #764: per-cadastral-quarter price index correction ───────────────────
+    # Gap-correction: квартальный индекс применяется ТОЛЬКО в pure-radius пути
+    # (когда same-building anchor и IMV-blend не сработали). Корректирует РАЗРЫВ
+    # между квартальным уровнем целевого объекта и усреднённым квартальным уровнем
+    # аналогов — не дублирует location, уже заложенный в медиану аналогов.
+    # Формула: adjusted_ppm2 = base_ppm2 × target_index / avg_analog_index.
+    # За флагом: False → точно текущее поведение (backward-compatible).
+    estimate_quarter_index_enabled: bool = True
+    # Минимальное число сделок в квартале (sparse fallback: меньше — no-op).
+    estimate_quarter_index_min_n_deals: int = 10
+    # Guard-2 (no double-count): если доля аналогов ИЗ ТОГО ЖЕ квартала > порога —
+    # аналоги уже несут локацию квартала → skip (location in median).
+    estimate_quarter_match_skip_ratio: float = 0.6
+    # Bimodal/nominal guard (backtest 2026-05-31): структурно неоднородные кварталы
+    # дают индекс > 2.0 при малой выборке → no-op чтобы избежать регрессию.
+    estimate_quarter_index_max_for_small_n: float = 2.0
+    estimate_quarter_index_small_n_threshold: int = 50
+
     # ── Estimate enrichment time-budgets (#654) ──────────────────────────────
     # POST /estimate делает несколько ПОСЛЕДОВАТЕЛЬНЫХ блокирующих сетевых
     # вызовов (geocode → Overpass → Yandex valuation → IMV → Cian). Yandex

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

@@ -788,6 +788,133 @@ def _apply_imv_blend(
     return new_median, new_range_high, new_ppm2, blended, anchor_total
 
 
+# ── #764: per-cadastral-quarter price index correction ───────────────────────
+
+
+def _quarter_from_cadastre(cad_num: str | None) -> str | None:
+    """Извлечь кадастровый номер квартала из кадастрового номера дома/квартиры.
+
+    Формат: AA:BB:CCCCCC или AA:BB:CCCCCCC (6 или 7 цифр в третьей части).
+    Возвращаем первые три двоеточие-разделённых компонента.
+    Пример: "66:41:0204016:350" → "66:41:0204016".
+    При отсутствии или некорректном формате → None.
+    """
+    if not cad_num:
+        return None
+    parts = cad_num.split(":")
+    if len(parts) < 3:
+        return None
+    quarter = ":".join(parts[:3])
+    # Проверяем что третья часть — числовая (квартал, не мусор)
+    if not parts[2].isdigit():
+        return None
+    return quarter
+
+
+def _lookup_quarter_index(
+    db: Session,
+    *,
+    quarter_cad_number: str,
+    min_n_deals: int,
+) -> tuple[float, int] | None:
+    """Поиск price_index для кадастрового квартала в FDW-таблице quarter_price_index.
+
+    Возвращает (price_index, n_deals) или None при отсутствии строки / n_deals < min_n_deals
+    / любой FDW-ошибке (graceful — backward-compatible).
+    Использует CAST(:q AS varchar) — psycopg v3 convention.
+    """
+    try:
+        row = (
+            db.execute(
+                text(
+                    """
+                    SELECT price_index, n_deals
+                    FROM quarter_price_index
+                    WHERE quarter_cad_number = CAST(:q AS varchar)
+                      AND n_deals >= CAST(:min_n AS bigint)
+                    LIMIT 1
+                    """
+                ),
+                {"q": quarter_cad_number, "min_n": min_n_deals},
+            )
+            .mappings()
+            .first()
+        )
+    except Exception as exc:
+        logger.warning("quarter_price_index FDW lookup failed (graceful, no-op): %s", exc)
+        return None
+    if row is None:
+        return None
+    return float(row["price_index"]), int(row["n_deals"])
+
+
+def _lookup_quarter_indexes(
+    db: Session,
+    *,
+    quarter_cad_numbers: list[str],
+    min_n_deals: int,
+) -> dict[str, float]:
+    """Батч-поиск price_index для списка кадастровых кварталов (одним SQL-запросом).
+
+    Возвращает {quarter_cad_number: price_index} только для кварталов, у которых
+    n_deals >= min_n_deals. Кварталы без записи или с n_deals < порога — не попадают
+    в словарь. При любой FDW-ошибке → {} (graceful, avg_analog_index остаётся 1.0).
+    """
+    if not quarter_cad_numbers:
+        return {}
+    distinct = list(dict.fromkeys(quarter_cad_numbers))  # сохраняем порядок, убираем дубли
+    try:
+        rows = (
+            db.execute(
+                text(
+                    """
+                    SELECT quarter_cad_number, price_index
+                    FROM quarter_price_index
+                    WHERE quarter_cad_number = ANY(CAST(:quarters AS varchar[]))
+                      AND n_deals >= CAST(:min_n AS bigint)
+                    """
+                ),
+                {"quarters": distinct, "min_n": min_n_deals},
+            )
+            .mappings()
+            .all()
+        )
+    except Exception as exc:
+        logger.warning("quarter_price_index batch FDW lookup failed (graceful, no-op): %s", exc)
+        return {}
+    return {str(row["quarter_cad_number"]): float(row["price_index"]) for row in rows}
+
+
+def _apply_quarter_index(
+    *,
+    base_median_ppm2: float,
+    base_median_price: int,
+    base_range_low: int,
+    base_range_high: int,
+    target_index: float,
+    avg_analog_index: float,
+) -> tuple[float, int, int, int, float]:
+    """Чистая (testable без БД) gap-correction квартального индекса (#764).
+
+    Корректирует ТОЛЬКО разрыв между квартальным уровнем целевого объекта и
+    усреднённым квартальным уровнем аналогов:
+        factor = target_index / avg_analog_index
+        adjusted_ppm2 = base_median_ppm2 × factor
+
+    Все ценовые выходы масштабируются одним и тем же factor → median/range
+    остаются геометрически консистентными.
+
+    Returns (adjusted_ppm2, adjusted_median_price, adjusted_range_low,
+             adjusted_range_high, factor).
+    """
+    factor = target_index / avg_analog_index
+    adjusted_ppm2 = base_median_ppm2 * factor
+    adjusted_median_price = round(base_median_price * factor)
+    adjusted_range_low = round(base_range_low * factor)
+    adjusted_range_high = round(base_range_high * factor)
+    return adjusted_ppm2, adjusted_median_price, adjusted_range_low, adjusted_range_high, factor
+
+
 def _fetch_dkp_corridor(
     db: Session,
     *,
@@ -1837,6 +1964,10 @@ async def estimate_quality(
     # blend'ом и расширяем верх диапазона. Всё за флагом + null-guard (no-op без IMV).
     # ВАЖНО (v2): IMV-blend выполняется ТОЛЬКО когда same-building anchor НЕ сработал
     # (anchor_tier is None) — не накладываем blend поверх уже-построенного якоря дома.
+    # #764: imv_anchor_present — любой IMV-anchor повлиял на estimate (median OR range).
+    # Guard-1b использует этот флаг чтобы пропустить квартальный индекс при любом
+    # IMV-влиянии, не только при blended (range_high расширяется даже без blend).
+    imv_anchor_present: bool = False
     avito_imv_summary: AvitoImvSummary | None = None
     if (
         anchor_tier is None
@@ -1883,6 +2014,7 @@ async def estimate_quality(
             )
 
         if anchor_total is not None:
+            imv_anchor_present = True
             new_median, new_range_high, new_ppm2, blended, anchor_used = _apply_imv_blend(
                 median_price=median_price,
                 range_high=range_high,
@@ -1942,6 +2074,146 @@ async def estimate_quality(
                 ),
             )
 
+    # ── #764: per-cadastral-quarter price index gap-correction ──────────────────
+    # Применяется ТОЛЬКО в pure-radius пути (Guard-1): когда same-building anchor
+    # не сработал И IMV-blend не поднял медиану. Оба механизма уже учитывают
+    # location пространственно — наложение индекса сверху даёт double-count.
+    # Формула: adjusted_ppm2 = base_ppm2 × (target_index / avg_analog_index),
+    # где avg_analog_index = взвешенная по ppm² медиана аналогов, чьи кварталы
+    # известны. Если аналоги без кадастрового номера — avg_analog_index=1.0 (no-op).
+    if (
+        settings.estimate_quarter_index_enabled
+        and anchor_tier is None  # Guard-1a: same-building anchor не сработал
+        and not imv_anchor_present  # Guard-1b: IMV-anchor не повлиял (median или range)
+        and median_price > 0
+        and payload.area_m2
+    ):
+        # Резолвим квартал target'а: Primary — DaData house_cadnum.
+        target_quarter: str | None = _quarter_from_cadastre(
+            dadata.house_cadnum if dadata is not None else None
+        )
+        # Fallback: building_cadastral_number из самих аналогов (если все в 1 доме
+        # — Tier S path; тогда кадастровый номер квартала тот же). Не применяем
+        # PostGIS point-in-quarter: нет готовой geometry-таблицы кварталов в tradein DB.
+        if target_quarter is None:
+            for lot in listings_clean:
+                cq = _quarter_from_cadastre(lot.get("building_cadastral_number"))
+                if cq is not None:
+                    target_quarter = cq
+                    break
+
+        if target_quarter is not None:
+            qindex_result = _lookup_quarter_index(
+                db,
+                quarter_cad_number=target_quarter,
+                min_n_deals=settings.estimate_quarter_index_min_n_deals,
+            )
+            if qindex_result is not None:
+                target_qi, target_n_deals = qindex_result
+
+                # Bimodal/nominal guard (Guard-4): структурно неоднородный квартал
+                # при малой выборке → no-op (regr. Радищева 66:41:0401017 et al.)
+                if (
+                    target_qi > settings.estimate_quarter_index_max_for_small_n
+                    and target_n_deals < settings.estimate_quarter_index_small_n_threshold
+                ):
+                    logger.info(
+                        "quarter_index: bimodal guard triggered "
+                        "(index=%.3f n=%d < %d) for %s — no-op",
+                        target_qi,
+                        target_n_deals,
+                        settings.estimate_quarter_index_small_n_threshold,
+                        target_quarter,
+                    )
+                else:
+                    # Вычисляем квартал каждого аналога ОДИН РАЗ — переиспользуем
+                    # для Guard-2 (same-quarter ratio) и avg_analog_index weighting.
+                    # lot_quarters_for_guard2: все лоты с известным кварталом (как раньше).
+                    # analog_quarters: только лоты с известным кварталом И ценой (для весов).
+                    lot_quarters_for_guard2: list[str] = []
+                    analog_quarters: list[tuple[str, float]] = []
+                    for lot in listings_clean:
+                        lq = _quarter_from_cadastre(lot.get("building_cadastral_number"))
+                        if lq is None:
+                            continue
+                        lot_quarters_for_guard2.append(lq)
+                        lp = lot.get("price_per_m2")
+                        if lp:
+                            analog_quarters.append((lq, float(lp)))
+
+                    # Guard-2: доля аналогов ИЗ ТОГО ЖЕ квартала > skip_ratio →
+                    # location уже в медиане — пропускаем.
+                    same_quarter_count = sum(
+                        1 for lq in lot_quarters_for_guard2 if lq == target_quarter
+                    )
+                    same_quarter_ratio = (
+                        same_quarter_count / len(listings_clean) if listings_clean else 0.0
+                    )
+                    if same_quarter_ratio > settings.estimate_quarter_match_skip_ratio:
+                        logger.info(
+                            "quarter_index: Guard-2 skip (same-quarter ratio=%.2f > %.2f)"
+                            " for %s",
+                            same_quarter_ratio,
+                            settings.estimate_quarter_match_skip_ratio,
+                            target_quarter,
+                        )
+                    else:
+                        # Вычисляем avg_analog_index — ppm²-взвешенное среднее по
+                        # аналогам, чьи кварталы известны И присутствуют в индексе.
+                        # Один батч-запрос вместо N последовательных FDW roundtrips.
+                        # Аналоги без кадастрового номера — игнорируем (не штрафуем).
+                        distinct_analog_quarters = list(
+                            dict.fromkeys(lq for lq, _lp in analog_quarters)
+                        )
+                        analog_index_map = _lookup_quarter_indexes(
+                            db,
+                            quarter_cad_numbers=distinct_analog_quarters,
+                            min_n_deals=settings.estimate_quarter_index_min_n_deals,
+                        )
+                        weighted_sum = 0.0
+                        weight_total = 0.0
+                        for lq, lp in analog_quarters:
+                            lot_qi = analog_index_map.get(lq)
+                            if lot_qi is None:
+                                continue
+                            weighted_sum += lp * lot_qi
+                            weight_total += lp
+
+                        avg_analog_index = weighted_sum / weight_total if weight_total > 0 else 1.0
+
+                        (
+                            median_ppm2,
+                            median_price,
+                            range_low,
+                            range_high,
+                            qi_factor,
+                        ) = _apply_quarter_index(
+                            base_median_ppm2=median_ppm2,
+                            base_median_price=median_price,
+                            base_range_low=range_low,
+                            base_range_high=range_high,
+                            target_index=target_qi,
+                            avg_analog_index=avg_analog_index,
+                        )
+                        analogs_with_qi = sum(
+                            1 for lq, _lp in analog_quarters if lq in analog_index_map
+                        )
+                        logger.info(
+                            "quarter_index: applied target=%s target_qi=%.3f"
+                            " avg_analog_qi=%.3f factor=%.3f"
+                            " (same_quarter_ratio=%.2f analogs_with_qi=%d)",
+                            target_quarter,
+                            target_qi,
+                            avg_analog_index,
+                            qi_factor,
+                            same_quarter_ratio,
+                            analogs_with_qi,
+                        )
+                        explanation = (explanation or "") + (
+                            f" Учтена локация квартала" f" (индекс цен квартала ×{qi_factor:.2f})."
+                        )
+                        sources_used_pre = sorted(set(sources_used_pre) | {"quarter_index"})
+
     # 4c (cont.). expected_sold_* выводим ЗДЕСЬ — ПОСЛЕ #651 IMV-blend / SB-anchor,
     # которые могли поднять median_price/median_ppm2 и расширить range_high. Применяем
     # ratio к POST-якорным значениям → asking (median_price_rub) и sold
@@ -2626,7 +2898,8 @@ _ANALOG_SELECT_COLS = """
     rooms, area_m2, floor, total_floors,
     price_rub, price_per_m2,
     listing_date, days_on_market, photo_urls,
-    scraped_at
+    scraped_at,
+    building_cadastral_number
 """
 
 _COMMON_WHERE = """
@@ -2737,7 +3010,8 @@ def _fetch_analogs(
                     rooms, area_m2, floor, total_floors,
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls,
-                    scraped_at, distance_m, relevance_score
+                    scraped_at, distance_m, relevance_score,
+                    building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
                 ORDER BY scraped_at DESC
@@ -2787,7 +3061,8 @@ def _fetch_analogs(
                     rooms, area_m2, floor, total_floors,
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls,
-                    scraped_at, distance_m, relevance_score
+                    scraped_at, distance_m, relevance_score,
+                    building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
                 ORDER BY scraped_at DESC
@@ -2874,7 +3149,8 @@ def _fetch_analogs(
                     rooms, area_m2, floor, total_floors,
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls,
-                    scraped_at, distance_m, relevance_score
+                    scraped_at, distance_m, relevance_score,
+                    building_cadastral_number
                 FROM base
                 WHERE rn_addr <= :max_per_addr
                 ORDER BY relevance_score
@@ -2929,6 +3205,7 @@ def _fetch_analogs(
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls,
                     scraped_at,
+                    building_cadastral_number,
                     ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                         AS distance_m,
                     (
@@ -3004,6 +3281,7 @@ def _fetch_analogs(
                 price_rub, price_per_m2,
                 listing_date, days_on_market, photo_urls,
                 scraped_at,
+                building_cadastral_number,
                 distance_m,
                 relevance_score
             FROM base

tradein-mvp/backend/tests/test_estimator_quarter_index.py

@@ -0,0 +1,868 @@
+"""Unit tests for #764 — per-cadastral-quarter price index gap-correction.
+
+Tests покрывают:
+  - _quarter_from_cadastre: парсинг кадастрового номера в квартал
+  - _apply_quarter_index: чистая математика корректировки (без БД)
+  - _lookup_quarter_index: DB-хелпер с мокнутой Session
+  - Guard-1: anchor_tier не None → no-op (same-building anchor)
+  - Guard-1b: IMV-blended → no-op
+  - Guard-2: >0.6 аналогов в целевом квартале → no-op
+  - Sparse fallback: нет строки / n_deals < min_n → no-op
+  - Bimodal guard: price_index>2.0 AND n_deals<50 → no-op
+  - Flag off → точное старое поведение
+
+Паттерн: os.environ.setdefault перед импортом (как test_estimator_pure_units.py).
+Чистые хелперы — без БД. DB-хелпер — мокнутая Session.
+estimate_quality-level тесты — через anyio.run + полный stub-пач всех I/O.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import anyio
+import pytest
+
+from app.services.estimator import (
+    _apply_quarter_index,
+    _lookup_quarter_index,
+    _lookup_quarter_indexes,
+    _quarter_from_cadastre,
+)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# _quarter_from_cadastre
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_quarter_from_cadastre_standard() -> None:
+    """Нормальный кадастровый номер дома → квартал (первые три части)."""
+    assert _quarter_from_cadastre("66:41:0204016:350") == "66:41:0204016"
+
+
+def test_quarter_from_cadastre_7digit_block() -> None:
+    """7-значный блок квартала → корректно."""
+    assert _quarter_from_cadastre("66:41:0401017:100") == "66:41:0401017"
+
+
+def test_quarter_from_cadastre_none_input() -> None:
+    assert _quarter_from_cadastre(None) is None
+
+
+def test_quarter_from_cadastre_empty_string() -> None:
+    assert _quarter_from_cadastre("") is None
+
+
+def test_quarter_from_cadastre_too_few_parts() -> None:
+    assert _quarter_from_cadastre("66:41") is None
+
+
+def test_quarter_from_cadastre_non_numeric_third_part() -> None:
+    """Третья часть не числовая → None (не кадастровый квартал)."""
+    assert _quarter_from_cadastre("66:41:BADDATA:100") is None
+
+
+def test_quarter_from_cadastre_no_fourth_part() -> None:
+    """Три части без объекта — само по себе квартал: возвращаем как есть."""
+    result = _quarter_from_cadastre("66:41:0204016")
+    assert result == "66:41:0204016"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# _apply_quarter_index — чистая математика
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_apply_quarter_index_basic_math() -> None:
+    """Базовый кейс: target_index=1.2, avg_analog=1.0 → factor=1.2."""
+    ppm2, median, low, high, factor = _apply_quarter_index(
+        base_median_ppm2=100_000.0,
+        base_median_price=5_000_000,
+        base_range_low=4_000_000,
+        base_range_high=6_000_000,
+        target_index=1.2,
+        avg_analog_index=1.0,
+    )
+    assert abs(factor - 1.2) < 1e-9
+    assert abs(ppm2 - 120_000.0) < 1.0
+    assert median == round(5_000_000 * 1.2)
+    assert low == round(4_000_000 * 1.2)
+    assert high == round(6_000_000 * 1.2)
+
+
+def test_apply_quarter_index_gap_correction() -> None:
+    """Gap-correction: target=1.3, avg_analog=1.1 → factor=1.3/1.1 ≈ 1.182."""
+    _, median, _, _, factor = _apply_quarter_index(
+        base_median_ppm2=200_000.0,
+        base_median_price=10_000_000,
+        base_range_low=8_000_000,
+        base_range_high=12_000_000,
+        target_index=1.3,
+        avg_analog_index=1.1,
+    )
+    expected_factor = 1.3 / 1.1
+    assert abs(factor - expected_factor) < 1e-9
+    assert median == round(10_000_000 * expected_factor)
+
+
+def test_apply_quarter_index_same_quarter_noop() -> None:
+    """target_index == avg_analog_index → factor=1.0, медиана не меняется."""
+    _, median, low, high, factor = _apply_quarter_index(
+        base_median_ppm2=150_000.0,
+        base_median_price=7_500_000,
+        base_range_low=6_000_000,
+        base_range_high=9_000_000,
+        target_index=1.05,
+        avg_analog_index=1.05,
+    )
+    assert abs(factor - 1.0) < 1e-9
+    assert median == 7_500_000
+    assert low == 6_000_000
+    assert high == 9_000_000
+
+
+def test_apply_quarter_index_downcorrection() -> None:
+    """target_index < avg_analog_index → factor < 1.0 (коррекция вниз тоже работает)."""
+    _, median, _, _, factor = _apply_quarter_index(
+        base_median_ppm2=100_000.0,
+        base_median_price=5_000_000,
+        base_range_low=4_000_000,
+        base_range_high=6_000_000,
+        target_index=0.9,
+        avg_analog_index=1.0,
+    )
+    assert factor < 1.0
+    assert median == round(5_000_000 * 0.9)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# _lookup_quarter_index — мокнутая Session
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_lookup_quarter_index_returns_row() -> None:
+    """Нормальная строка → возвращает (price_index, n_deals)."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.first.return_value = {
+        "price_index": 1.25,
+        "n_deals": 42,
+    }
+    result = _lookup_quarter_index(mock_db, quarter_cad_number="66:41:0204016", min_n_deals=10)
+    assert result is not None
+    qi, n = result
+    assert abs(qi - 1.25) < 1e-9
+    assert n == 42
+
+
+def test_lookup_quarter_index_none_when_no_row() -> None:
+    """Нет строки → None."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.first.return_value = None
+    result = _lookup_quarter_index(mock_db, quarter_cad_number="66:41:9999999", min_n_deals=10)
+    assert result is None
+
+
+def test_lookup_quarter_index_fdw_exception_graceful() -> None:
+    """FDW exception → None (graceful, no re-raise)."""
+    mock_db = MagicMock()
+    mock_db.execute.side_effect = RuntimeError("FDW connection refused")
+    result = _lookup_quarter_index(mock_db, quarter_cad_number="66:41:0204016", min_n_deals=10)
+    assert result is None
+
+
+def test_lookup_quarter_index_no_cast_colon_colon_in_sql() -> None:
+    """SQL текст хелпера не должен содержать :x::type (psycopg v3 rule)."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.first.return_value = None
+    _lookup_quarter_index(mock_db, quarter_cad_number="66:41:0204016", min_n_deals=10)
+    args, _ = mock_db.execute.call_args
+    sql_text = str(args[0])
+    # psycopg v3: CAST(:x AS type), never :x::type
+    import re
+
+    assert not re.search(r":[a-z_]+::[a-z]", sql_text), f"::type cast found in SQL: {sql_text}"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# _lookup_quarter_indexes (plural) — батч-хелпер
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_lookup_quarter_indexes_returns_dict() -> None:
+    """Нормальный результат → словарь {quarter: price_index}."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.all.return_value = [
+        {"quarter_cad_number": "66:41:0204016", "price_index": 1.25},
+        {"quarter_cad_number": "66:41:9998888", "price_index": 0.95},
+    ]
+    result = _lookup_quarter_indexes(
+        mock_db,
+        quarter_cad_numbers=["66:41:0204016", "66:41:9998888"],
+        min_n_deals=10,
+    )
+    assert result == {"66:41:0204016": 1.25, "66:41:9998888": 0.95}
+
+
+def test_lookup_quarter_indexes_empty_input_returns_empty() -> None:
+    """Пустой список кварталов → {} без обращения к БД."""
+    mock_db = MagicMock()
+    result = _lookup_quarter_indexes(mock_db, quarter_cad_numbers=[], min_n_deals=10)
+    assert result == {}
+    mock_db.execute.assert_not_called()
+
+
+def test_lookup_quarter_indexes_fdw_exception_returns_empty() -> None:
+    """FDW exception → {} (graceful, no re-raise)."""
+    mock_db = MagicMock()
+    mock_db.execute.side_effect = RuntimeError("FDW connection refused")
+    result = _lookup_quarter_indexes(
+        mock_db,
+        quarter_cad_numbers=["66:41:0204016"],
+        min_n_deals=10,
+    )
+    assert result == {}
+
+
+def test_lookup_quarter_indexes_deduplicates_input() -> None:
+    """Дублирующиеся кварталы в списке — передаются в БД без дублей."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.all.return_value = [
+        {"quarter_cad_number": "66:41:0204016", "price_index": 1.1},
+    ]
+    _lookup_quarter_indexes(
+        mock_db,
+        quarter_cad_numbers=["66:41:0204016", "66:41:0204016", "66:41:0204016"],
+        min_n_deals=5,
+    )
+    passed_params = mock_db.execute.call_args[0][1]
+    assert passed_params["quarters"] == ["66:41:0204016"]
+
+
+def test_lookup_quarter_indexes_no_cast_colon_colon_in_sql() -> None:
+    """Батч-хелпер: SQL не содержит :x::type (psycopg v3 rule)."""
+    import re
+
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.all.return_value = []
+    _lookup_quarter_indexes(
+        mock_db,
+        quarter_cad_numbers=["66:41:0204016"],
+        min_n_deals=10,
+    )
+    args, _ = mock_db.execute.call_args
+    sql_text = str(args[0])
+    assert not re.search(r":[a-z_]+::[a-z]", sql_text), f"::type cast found in SQL: {sql_text}"
+
+
+def test_lookup_quarter_indexes_uses_any_cast_array_idiom() -> None:
+    """SQL батч-хелпера содержит ANY(CAST(:quarters AS varchar[])) — pgpsycopg3 idiom."""
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.all.return_value = []
+    _lookup_quarter_indexes(
+        mock_db,
+        quarter_cad_numbers=["66:41:0204016"],
+        min_n_deals=10,
+    )
+    args, _ = mock_db.execute.call_args
+    sql_text = str(args[0])
+    assert "ANY(CAST(:quarters AS varchar[]))" in sql_text
+
+
+def test_lookup_quarter_indexes_multi_quarter_factor_matches_single() -> None:
+    """Батч возвращает те же значения, что N одиночных вызовов — математика идентична.
+
+    Аналоги: 2 лота из квартала A (index=1.2, ppm2=100k),
+             1 лот из квартала B (index=0.8, ppm2=200k).
+    Ожидаемый avg_analog_index = (100k*1.2 + 100k*1.2 + 200k*0.8) / (100k+100k+200k)
+                                = (120k + 120k + 160k) / 400k = 400k/400k = 1.0.
+    """
+    index_map = {"66:41:AAAAAAA": 1.2, "66:41:BBBBBBB": 0.8}
+    analog_lots = [
+        ("66:41:AAAAAAA", 100_000.0),
+        ("66:41:AAAAAAA", 100_000.0),
+        ("66:41:BBBBBBB", 200_000.0),
+    ]
+    weighted_sum = sum(ppm2 * index_map[q] for q, ppm2 in analog_lots)
+    weight_total = sum(ppm2 for _, ppm2 in analog_lots)
+    avg_analog_index = weighted_sum / weight_total
+    assert abs(avg_analog_index - 1.0) < 1e-9, f"Expected 1.0, got {avg_analog_index}"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Helpers for estimate_quality integration tests (full I/O stub)
+# ─────────────────────────────────────────────────────────────────────────────
+
+_BASE_PPM2 = 150_000.0
+_AREA = 40.0
+
+
+def _make_listing_qi(
+    *,
+    price_per_m2: float = _BASE_PPM2,
+    area_m2: float = _AREA,
+    building_cadastral_number: str | None = None,
+) -> dict[str, Any]:
+    from datetime import UTC, datetime
+
+    price_rub = price_per_m2 * area_m2
+    return {
+        "source": "cian",
+        "source_url": "https://cian.ru/offer/1",
+        "address": "ЕКБ, ул. Тестовая, 5",
+        "lat": 56.838,
+        "lon": 60.595,
+        "rooms": 1,
+        "area_m2": area_m2,
+        "floor": 4,
+        "total_floors": 16,
+        "price_rub": price_rub,
+        "price_per_m2": price_per_m2,
+        "listing_date": datetime(2026, 5, 1),
+        "days_on_market": 10,
+        "photo_urls": [],
+        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
+        "distance_m": 100.0,
+        "relevance_score": 0.1,
+        "building_cadastral_number": building_cadastral_number,
+    }
+
+
+def _make_fake_geo_qi():
+    from app.services.geocoder import GeocodeResult
+
+    return GeocodeResult(
+        lat=56.838,
+        lon=60.595,
+        full_address="Свердловская обл., Екатеринбург, ул. Тестовая, 5",
+        provider="nominatim",
+    )
+
+
+def _make_payload_qi(rooms: int = 1, area_m2: float = _AREA):
+    from app.schemas.trade_in import TradeInEstimateInput
+
+    return TradeInEstimateInput(
+        address="ЕКБ, ул. Тестовая, 5",
+        area_m2=area_m2,
+        rooms=rooms,
+        floor=4,
+        total_floors=16,
+    )
+
+
+def _make_fake_dadata(house_cadnum: str | None):
+    """Minimal DadataAddressResult stub с нужным house_cadnum."""
+    from app.services.dadata import DadataAddressResult
+
+    return DadataAddressResult(
+        canonical_address="Свердловская обл., Екатеринбург, ул. Тестовая, 5",
+        house_cadnum=house_cadnum,
+        house_fias_id=None,
+        lat=56.838,
+        lon=60.595,
+        qc_geo=1,
+        qc_house=1,
+        kladr_id=None,
+        okato=None,
+        oktmo=None,
+        metro=[],
+        raw={},
+    )
+
+
+def _run_estimate_qi(
+    analogs: list[dict[str, Any]],
+    dadata_cadnum: str | None,
+    qi_lookup_result: tuple[float, int] | None,
+    *,
+    flag_enabled: bool = True,
+    anchor_tier_override: str | None = None,
+):
+    """Запускает estimate_quality с полным stub-пачем I/O; возвращает AggregatedEstimate."""
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload_qi()
+
+    dadata_obj = _make_fake_dadata(dadata_cadnum) if dadata_cadnum is not None else None
+
+    # Батч-хелпер возвращает словарь: для каждого переданного квартала — тот же индекс,
+    # что qi_lookup_result[0], если qi_lookup_result не None; иначе пустой dict.
+    def _fake_lookup_indexes(db_arg, *, quarter_cad_numbers, min_n_deals):
+        if qi_lookup_result is None:
+            return {}
+        return {q: qi_lookup_result[0] for q in quarter_cad_numbers}
+
+    async def _run():
+        with (
+            patch(
+                "app.services.estimator.geocode",
+                new=AsyncMock(return_value=_make_fake_geo_qi()),
+            ),
+            patch(
+                "app.services.estimator.dadata_clean_address",
+                new=AsyncMock(return_value=dadata_obj),
+            ),
+            patch("app.services.estimator.match_house_readonly", return_value=None),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch(
+                "app.services.estimator._fetch_analogs",
+                return_value=(list(analogs), False, "W"),
+            ),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
+            patch(
+                "app.services.estimator._get_or_fetch_imv_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_asking_sold_ratio",
+                return_value=(None, None),
+            ),
+            patch("app.services.estimator._fetch_house_imv_anchor", return_value=None),
+            # Stub singular target-quarter lookup
+            patch(
+                "app.services.estimator._lookup_quarter_index",
+                return_value=qi_lookup_result,
+            ),
+            # Stub batched analog-quarter lookup
+            patch(
+                "app.services.estimator._lookup_quarter_indexes",
+                side_effect=_fake_lookup_indexes,
+            ),
+            patch(
+                "app.services.estimator.settings.estimate_quarter_index_enabled",
+                flag_enabled,
+            ),
+        ):
+            return await estimate_quality(payload, db)
+
+    return anyio.run(_run)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Integration: gap-correction applied in pure-radius path
+# ─────────────────────────────────────────────────────────────────────────────
+
+_OTHER_QUARTER = "66:41:9998888"
+_TARGET_QUARTER = "66:41:0204016"
+
+_ANALOGS_OTHER_QUARTER = [
+    _make_listing_qi(
+        price_per_m2=_BASE_PPM2,
+        building_cadastral_number=f"{_OTHER_QUARTER}:100",
+    )
+    for _ in range(3)
+]
+
+
+def test_quarter_index_correction_applied() -> None:
+    """Gap-correction срабатывает в pure-radius пути.
+
+    target_index=1.2, avg_analog_index=1.0 (аналоги без известного квартала →
+    avg=1.0) → factor=1.2 → median должен вырасти на ×1.2.
+    """
+    base_median = round(_BASE_PPM2 * _AREA)  # 6_000_000
+
+    # Аналоги из ДРУГОГО квартала (building_cadastral_number = OTHER_QUARTER:100)
+    # _lookup_quarter_index для аналогов вернёт тот же (1.2, 30) что и для target —
+    # avg_analog_index = 1.2, factor = 1.2/1.2 = 1.0 (no change!).
+    # Чтобы увидеть ненулевую коррекцию, делаем аналоги БЕЗ кадастрового номера
+    # → avg_analog_index = 1.0 → factor = 1.2.
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+    est = _run_estimate_qi(
+        analogs=analogs_no_cadnum,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=(1.2, 30),
+        flag_enabled=True,
+    )
+    expected_median = round(base_median * 1.2)
+    assert est.median_price_rub == expected_median
+    # Disclosure должна содержать упоминание квартала
+    assert est.confidence_explanation is not None
+    assert "квартал" in est.confidence_explanation.lower()
+
+
+def test_quarter_index_flag_off_exact_old_behavior() -> None:
+    """При estimate_quarter_index_enabled=False — точное старое поведение."""
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+    est = _run_estimate_qi(
+        analogs=analogs_no_cadnum,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=(1.2, 30),
+        flag_enabled=False,
+    )
+    base_median = round(_BASE_PPM2 * _AREA)
+    assert est.median_price_rub == base_median
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Guard-2: >0.6 аналогов в целевом квартале → no-op
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_guard2_skip_when_majority_analogs_in_target_quarter() -> None:
+    """Guard-2: >60% аналогов из целевого квартала → индекс не применяется."""
+    # 4 аналога в target квартале, 1 в другом → ratio = 4/5 = 0.8 > 0.6 → skip
+    target_cadnum = f"{_TARGET_QUARTER}:100"
+    other_cadnum = f"{_OTHER_QUARTER}:100"
+    analogs = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=other_cadnum),
+    ]
+    base_median = round(_BASE_PPM2 * _AREA)
+    est = _run_estimate_qi(
+        analogs=analogs,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=(1.5, 30),  # high index — but guard-2 should skip
+        flag_enabled=True,
+    )
+    # Медиана НЕ должна изменяться
+    assert est.median_price_rub == base_median
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Sparse fallback: нет строки → no-op
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_sparse_fallback_no_row_noop() -> None:
+    """_lookup_quarter_index вернул None → no-op, медиана не меняется."""
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+    base_median = round(_BASE_PPM2 * _AREA)
+    est = _run_estimate_qi(
+        analogs=analogs_no_cadnum,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=None,  # sparse
+        flag_enabled=True,
+    )
+    assert est.median_price_rub == base_median
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Bimodal guard: price_index>2.0 AND n_deals<50 → no-op
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_bimodal_guard_skips_high_index_small_n() -> None:
+    """Bimodal guard: price_index=3.5, n_deals=20 → no-op."""
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+    base_median = round(_BASE_PPM2 * _AREA)
+    est = _run_estimate_qi(
+        analogs=analogs_no_cadnum,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=(3.5, 20),  # index>2.0 AND n<50 → bimodal guard
+        flag_enabled=True,
+    )
+    assert est.median_price_rub == base_median
+
+
+def test_bimodal_guard_allows_high_index_large_n() -> None:
+    """Bimodal guard НЕ срабатывает при price_index>2.0 если n_deals>=50."""
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+    base_median = round(_BASE_PPM2 * _AREA)
+    est = _run_estimate_qi(
+        analogs=analogs_no_cadnum,
+        dadata_cadnum=f"{_TARGET_QUARTER}:350",
+        qi_lookup_result=(2.5, 60),  # index>2.0 но n=60>=50 → применяется
+        flag_enabled=True,
+    )
+    # factor = 2.5 → медиана должна измениться
+    assert est.median_price_rub != base_median
+    assert est.median_price_rub == round(base_median * 2.5)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Guard-1a: anchor_tier не None → correction не применяется
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_guard1_anchor_tier_prevents_correction() -> None:
+    """Guard-1a: когда same-building anchor сработал (anchor_tier='A'),
+    квартальный индекс не применяется (double-count guard).
+
+    Стабим _compute_same_building_anchor, чтобы вернул непустой anchor dict
+    и _fetch_anchor_comps вернул comps → anchor_tier будет 'A'.
+    """
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload_qi()
+    dadata_obj = _make_fake_dadata(f"{_TARGET_QUARTER}:350")
+
+    fake_anchor = {
+        "anchor_ppm2": _BASE_PPM2 * 1.5,
+        "anchor_sold_ppm2": _BASE_PPM2 * 1.4,
+        "fsd": 0.05,
+        "confidence": "high",
+        "n": 3,
+        "cv": 0.05,
+        "comp_min_ppm2": _BASE_PPM2 * 1.3,
+        "comp_max_ppm2": _BASE_PPM2 * 1.7,
+        "used_uplift": False,
+        "haircut": 0.05,
+    }
+    fake_comps = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2 * 1.5, building_cadastral_number=None)
+        for _ in range(3)
+    ]
+
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+
+    async def _run():
+        with (
+            patch(
+                "app.services.estimator.geocode",
+                new=AsyncMock(return_value=_make_fake_geo_qi()),
+            ),
+            patch(
+                "app.services.estimator.dadata_clean_address",
+                new=AsyncMock(return_value=dadata_obj),
+            ),
+            patch("app.services.estimator.match_house_readonly", return_value=None),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch(
+                "app.services.estimator._fetch_analogs",
+                return_value=(list(analogs_no_cadnum), False, "W"),
+            ),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
+            patch(
+                "app.services.estimator._get_or_fetch_imv_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_asking_sold_ratio",
+                return_value=(None, None),
+            ),
+            patch("app.services.estimator._fetch_house_imv_anchor", return_value=None),
+            # same-building anchor FIRES → anchor_tier = 'A'
+            patch(
+                "app.services.estimator._fetch_anchor_comps",
+                return_value=(fake_comps, "A"),
+            ),
+            patch(
+                "app.services.estimator._compute_same_building_anchor",
+                return_value=fake_anchor,
+            ),
+            patch(
+                "app.services.estimator._lookup_quarter_index",
+                return_value=(1.5, 30),  # would apply if not guarded
+            ),
+        ):
+            return await estimate_quality(payload, db)
+
+    est = anyio.run(_run)
+
+    # Медиана от anchor = anchor_ppm2 * repair_coef(=1.0 нет ремонта) * area
+    anchor_median = round(_BASE_PPM2 * 1.5 * _AREA)
+    assert est.median_price_rub == anchor_median
+    # explanation не содержит упоминания квартала (guard-1 сработал)
+    assert (
+        est.confidence_explanation is None
+        or "квартал" not in (est.confidence_explanation or "").lower()
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Guard-1b: IMV-blended → correction не применяется
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_guard1b_imv_blend_prevents_correction() -> None:
+    """Guard-1b: IMV-blend повышал медиану → квартальный индекс не применяется."""
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload_qi()
+    dadata_obj = _make_fake_dadata(f"{_TARGET_QUARTER}:350")
+
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+
+    # IMV anchor сильно выше медианы → blend сработает
+    imv_anchor = {
+        "recommended_price": 30_000_000,  # ≫ base_median 6М × 1.15
+        "lower_price": 25_000_000,
+        "higher_price": 35_000_000,
+        "market_count": 500,
+        "rooms": 1,
+        "area_m2": _AREA,
+    }
+
+    async def _run():
+        with (
+            patch(
+                "app.services.estimator.geocode",
+                new=AsyncMock(return_value=_make_fake_geo_qi()),
+            ),
+            patch(
+                "app.services.estimator.dadata_clean_address",
+                new=AsyncMock(return_value=dadata_obj),
+            ),
+            patch("app.services.estimator.match_house_readonly", return_value=None),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch(
+                "app.services.estimator._fetch_analogs",
+                return_value=(list(analogs_no_cadnum), False, "W"),
+            ),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
+            patch(
+                "app.services.estimator._get_or_fetch_imv_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_asking_sold_ratio",
+                return_value=(None, None),
+            ),
+            # IMV anchor fires (returns non-None) → blend will trigger
+            patch(
+                "app.services.estimator._fetch_house_imv_anchor",
+                return_value=imv_anchor,
+            ),
+            patch(
+                "app.services.estimator._lookup_quarter_index",
+                return_value=(1.5, 30),  # would apply if not guarded
+            ),
+        ):
+            return await estimate_quality(payload, db)
+
+    est = anyio.run(_run)
+
+    # IMV blend: base 6М, anchor 30М, w=0.5 → 18М
+    blended_median = round(6_000_000 * 0.5 + 30_000_000 * 0.5)
+    assert est.median_price_rub == blended_median
+    # explanation не должна содержать квартального дисклоужера
+    assert "квартал" not in (est.confidence_explanation or "").lower()
+
+
+def test_guard1b_imv_anchor_below_blend_threshold_prevents_correction() -> None:
+    """Guard-1b: IMV anchor присутствует но ниже blend-порога (blended=False).
+
+    До фикса #764: imv_blended=False → квартальный индекс применялся поверх
+    IMV-расширенного range_high (double-influence). После фикса: imv_anchor_present=True
+    → quarter index не применяется вне зависимости от blended.
+    """
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload_qi()
+    dadata_obj = _make_fake_dadata(f"{_TARGET_QUARTER}:350")
+
+    analogs_no_cadnum = [
+        _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
+    ]
+
+    # IMV anchor НИЖЕ blend-порога: base_median = 6_000_000, threshold=1.15 → порог 6.9М.
+    # anchor=6_500_000 < 6.9М → blended=False, но range_high IMV-расширен.
+    imv_anchor_below_threshold = {
+        "recommended_price": 6_500_000,
+        "lower_price": 5_800_000,
+        "higher_price": 7_200_000,
+        "market_count": 100,
+        "rooms": 1,
+        "area_m2": _AREA,
+    }
+
+    async def _run():
+        with (
+            patch(
+                "app.services.estimator.geocode",
+                new=AsyncMock(return_value=_make_fake_geo_qi()),
+            ),
+            patch(
+                "app.services.estimator.dadata_clean_address",
+                new=AsyncMock(return_value=dadata_obj),
+            ),
+            patch("app.services.estimator.match_house_readonly", return_value=None),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch(
+                "app.services.estimator._fetch_analogs",
+                return_value=(list(analogs_no_cadnum), False, "W"),
+            ),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
+            patch(
+                "app.services.estimator._get_or_fetch_imv_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator._get_asking_sold_ratio",
+                return_value=(None, None),
+            ),
+            # IMV anchor returns below-threshold value → blended=False but anchor_present=True
+            patch(
+                "app.services.estimator._fetch_house_imv_anchor",
+                return_value=imv_anchor_below_threshold,
+            ),
+            patch(
+                "app.services.estimator._lookup_quarter_index",
+                return_value=(1.5, 30),  # would apply factor=1.5 if not guarded
+            ),
+        ):
+            return await estimate_quality(payload, db)
+
+    est = anyio.run(_run)
+
+    # Медиана не должна быть умножена на 1.5 (квартальный индекс заблокирован).
+    base_median = round(_BASE_PPM2 * _AREA)  # 6_000_000
+    assert est.median_price_rub == base_median
+    # explanation не содержит квартального дисклоужера
+    assert "квартал" not in (est.confidence_explanation or "").lower()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))