feat(tradein/estimator): premium_building flag from premium_houses MV (#2002)

Read-only _is_premium_building(db, house_id) looks up target_house_id in the
premium_houses MV (data/sql/139) and surfaces premium_building +
premium_building_median_ppm2 on AggregatedEstimate. Metadata only for
manual-review routing (#4) — does NOT touch _price_from_inputs / expected_sold /
median / ranges, so the backtest regression gate is unaffected.

MV-missing / SQL errors degrade to (False, None) with a rollback so an estimate
never breaks where the MV does not yet exist.

tradein-mvp/backend/app/schemas/trade_in.py

@@ -219,6 +219,15 @@ class AggregatedEstimate(BaseModel):
     #   null             — нет данных / оценка не построена
     # НЕ удаляет/заменяет confidence_explanation (фронт fallback'ает на него).
     analog_tier: Literal["same_building", "micro_radius", "district", "city"] | None = None
+    # ── #2002: премиальный дом (флаг, НЕ ценовой сигнал) ──
+    # premium_building — целевой дом найден в materialized view `premium_houses`
+    #   (data/sql/139): ~298 EKB-домов, отобранных по концентрации дорогих листингов.
+    #   Это МЕТАДАННЫЕ для manual-review routing (#4) — НЕ влияет на median/expected_sold/
+    #   ranges. Дефолт False (поведение/сериализация без изменений, когда не премиум).
+    # premium_building_median_ppm2 — med_ppm2 из MV (₽/м² медиана дорогих листингов дома),
+    #   контекст для ревьюера. None если дом не премиальный / MV недоступна.
+    premium_building: bool = False
+    premium_building_median_ppm2: int | None = None
     # ── Параметры оценённой квартиры — нужны, чтобы восстановить карточку
     #    при открытии оценки по ссылке (?id=), когда формы-инпута уже нет ──
     area_m2: float | None = None

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

@@ -3188,6 +3188,11 @@ async def estimate_quality(
         else None
     )
 
+    # #2002: премиальный-дом флаг из MV `premium_houses` для target_house_id.
+    # МЕТАДАННЫЕ (manual-review routing #4), НЕ ценовой сигнал — не трогает
+    # median/expected_sold/ranges. Best-effort: degrade в (False, None) при отсутствии MV.
+    premium_building, premium_building_median_ppm2 = _is_premium_building(db, target_house_id)
+
     # #audit-2: структурный analog_tier — стабильный enum для фронта.
     # anchor-путь: anchor_tier "A" → "same_building", "C" → "micro_radius".
     # radius-путь: analog_tier "W" → "city", остальные → "district".
@@ -3283,6 +3288,8 @@ async def estimate_quality(
         metro_nearest=(dadata.metro if dadata and dadata.metro else []),
         address_precision=_qc_geo_to_precision(dadata.qc_geo if dadata else None),
         analog_tier=api_analog_tier,  # type: ignore[arg-type]
+        premium_building=premium_building,
+        premium_building_median_ppm2=premium_building_median_ppm2,
     )
 
 
@@ -3297,6 +3304,36 @@ def _qc_geo_to_precision(qc_geo: int | None) -> str | None:
     return "approximate"
 
 
+def _is_premium_building(db: Session, house_id: int | None) -> tuple[bool, int | None]:
+    """#2002: флаг «целевой дом — премиальный» из MV `premium_houses` (data/sql/139).
+
+    `premium_houses` — ~298 EKB-домов, отобранных по концентрации дорогих листингов.
+    Это МЕТАДАННЫЕ для manual-review routing (#4), НЕ ценовой сигнал: не трогает
+    median/expected_sold/ranges. Возвращает (is_premium, med_ppm2) где med_ppm2 —
+    медиана ₽/м² дорогих листингов дома (контекст для ревьюера) или None.
+
+    Best-effort: MV может ещё не существовать в части окружений — любая ошибка
+    (нет MV / SQL) деградирует в (False, None) и НИКОГДА не валит оценку.
+    """
+    if house_id is None:
+        return (False, None)
+    try:
+        row = db.execute(
+            text("SELECT med_ppm2 FROM premium_houses WHERE house_id = CAST(:hid AS bigint)"),
+            {"hid": house_id},
+        ).first()
+    except Exception as exc:  # pragma: no cover — defensive (MV может ещё не существовать)
+        logger.warning("premium_houses lookup failed (graceful): %s", exc)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        return (False, None)
+    if row is None:
+        return (False, None)
+    return (True, int(row[0]) if row[0] is not None else None)
+
+
 def _estimate_days_on_market(
     listings: list[dict[str, Any]], deals: list[dict[str, Any]]
 ) -> int | None:

tradein-mvp/backend/tests/test_estimator_premium_building.py

@@ -0,0 +1,113 @@
+"""#2002: premium_building флаг из MV `premium_houses`.
+
+Покрывает read-only helper `_is_premium_building`:
+  - target резолвится в премиальный дом → (True, med_ppm2);
+  - target не в MV → (False, None);
+  - house_id is None → (False, None) без обращения к БД;
+  - MV отсутствует / SQL-ошибка → graceful (False, None), оценка не падает.
+
+Плюс smoke-проверка дефолтов в response-модели AggregatedEstimate (флаг — метаданные,
+сериализация без изменений, когда дом не премиальный).
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import MagicMock
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+import pytest
+
+from app.services.estimator import _is_premium_building
+
+
+def _make_mock_db(*, first_row: Any, raises: bool = False) -> MagicMock:
+    """mock db.execute(...).first() → first_row; raises=True имитирует отсутствие MV."""
+    mock_db = MagicMock()
+    if raises:
+        mock_db.execute.side_effect = RuntimeError('relation "premium_houses" does not exist')
+    else:
+        mock_db.execute.return_value.first.return_value = first_row
+    return mock_db
+
+
+def test_premium_house_yields_true_with_median() -> None:
+    """house_id в premium_houses → (True, med_ppm2)."""
+    db = _make_mock_db(first_row=(255_000,))
+
+    is_premium, median = _is_premium_building(db, 4242)
+
+    assert is_premium is True
+    assert median == 255_000
+    # CAST(:hid AS bigint) — psycopg v3, не :hid::bigint
+    sql = str(db.execute.call_args.args[0].text)
+    assert "CAST(:hid AS bigint)" in sql
+    assert "::bigint" not in sql
+
+
+def test_non_premium_house_yields_false_none() -> None:
+    """house_id отсутствует в MV (first() → None) → (False, None)."""
+    db = _make_mock_db(first_row=None)
+
+    assert _is_premium_building(db, 9999) == (False, None)
+
+
+def test_house_id_none_short_circuits_without_query() -> None:
+    """house_id is None → (False, None), к БД не обращаемся."""
+    db = _make_mock_db(first_row=(255_000,))
+
+    assert _is_premium_building(db, None) == (False, None)
+    db.execute.assert_not_called()
+
+
+def test_missing_mv_degrades_gracefully() -> None:
+    """MV не существует / SQL-ошибка → (False, None), без проброса исключения."""
+    db = _make_mock_db(first_row=None, raises=True)
+
+    assert _is_premium_building(db, 4242) == (False, None)
+    # после ошибки откатываем poisoned-транзакцию
+    db.rollback.assert_called_once()
+
+
+def test_premium_median_null_in_mv_is_handled() -> None:
+    """Дом в MV, но med_ppm2 NULL → (True, None) (премиум-флаг есть, медианы нет)."""
+    db = _make_mock_db(first_row=(None,))
+
+    assert _is_premium_building(db, 4242) == (True, None)
+
+
+def test_response_model_defaults_non_premium() -> None:
+    """AggregatedEstimate по умолчанию: premium_building=False, median=None.
+
+    Гарантирует, что добавление флага не меняет сериализацию для не-премиум оценок.
+    """
+    from datetime import UTC, datetime
+    from uuid import uuid4
+
+    from app.schemas.trade_in import AggregatedEstimate
+
+    est = AggregatedEstimate(
+        estimate_id=uuid4(),
+        median_price_rub=10_000_000,
+        range_low_rub=9_000_000,
+        range_high_rub=11_000_000,
+        median_price_per_m2=200_000,
+        confidence="medium",
+        n_analogs=12,
+        period_months=24,
+        analogs=[],
+        actual_deals=[],
+        expires_at=datetime.now(tz=UTC),
+    )
+
+    assert est.premium_building is False
+    assert est.premium_building_median_ppm2 is None
+    dumped = est.model_dump()
+    assert dumped["premium_building"] is False
+    assert dumped["premium_building_median_ppm2"] is None
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))