feat(tradein): premium_houses MV + premium_building flag (#2002)

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/app/tasks/refresh_search_matview.py

@@ -8,6 +8,7 @@ in tradein-mvp/backend/app/). This module provides both:
 NEEDS COORDINATION: main session must decide whether to bootstrap Celery in tradein-mvp
 or run via external scheduler (systemd timer / OS cron). Until then, no Beat schedule.
 """
+
 from __future__ import annotations
 
 import logging
@@ -21,20 +22,35 @@ logger = logging.getLogger(__name__)
 
 
 def refresh_search_matview() -> None:
-    """REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3, sync).
+    """REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv + premium_houses.
 
-    Logs duration. Idempotent. Safe to run during read traffic (CONCURRENTLY).
+    psycopg v3, sync. Logs duration. Idempotent. Safe to run during read traffic
+    (CONCURRENTLY). premium_houses (#2002) derives from `listings`, so it is refreshed
+    right after the search MV in the same session. Each MV is refreshed independently —
+    a failure on one is logged and does not abort the other (autocommit, so a failed
+    REFRESH does not leave the connection in an aborted transaction).
     """
     start = time.monotonic()
     # DATABASE_URL is SQLAlchemy dialect form (postgresql+psycopg://...) in tradein-mvp
     # (see tradein-mvp/docker-compose.prod.yml). libpq / psycopg.connect() accepts only
     # postgresql:// or postgres:// — strip the +psycopg dialect prefix.
     dsn = settings.database_url.replace("postgresql+psycopg://", "postgresql://", 1)
+    # CONCURRENTLY requires a UNIQUE index on each MV (listings_search_mv,
+    # premium_houses_house_id_uidx). Order matters: premium_houses depends on the same
+    # `listings` rows, so refresh it after the search MV is fresh.
+    matviews = ("listings_search_mv", "premium_houses")
     with psycopg.connect(dsn, autocommit=True) as conn:
-        with conn.cursor() as cur:
-            cur.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv")
-    elapsed = time.monotonic() - start
-    logger.info("listings_search_mv refresh completed in %.2fs", elapsed)
+        for mv in matviews:
+            mv_start = time.monotonic()
+            try:
+                with conn.cursor() as cur:
+                    cur.execute(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {mv}")
+            except Exception:
+                # Don't let one MV's failure skip the others; surface for alerting.
+                logger.exception("%s refresh failed", mv)
+            else:
+                logger.info("%s refresh completed in %.2fs", mv, time.monotonic() - mv_start)
+    logger.info("matview refresh batch completed in %.2fs", time.monotonic() - start)
 
 
 # Celery-task wrapper — only registers if celery_app exists.

tradein-mvp/backend/data/sql/139_premium_houses.sql

@@ -0,0 +1,56 @@
+-- 139_premium_houses.sql
+-- Purpose: premium_houses materialized view — premium / elite EKB buildings
+--          identified by listing concentration (#2002, premium-building identification).
+--
+-- Rationale (#2002):
+--   We flag a house as "premium" by the concentration of expensive listings on it.
+--   Validated query returns ~297 houses that together cover ~65% of all >=20M RUB
+--   elite listings. A house qualifies when it has a meaningful sample (n >= 4 active
+--   listings with sane price_per_m2) AND either:
+--     - a high median price_per_m2 (>= 220000 RUB/m2), OR
+--     - at least 3 listings priced >= 20M RUB (n_elite >= 3).
+--   price_per_m2 is clamped to [100000, 800000] to drop scrape garbage / mis-parsed
+--   areas before aggregating, so the median is robust.
+--
+-- house_id dupes: house_dedup is imperfect, so the same physical building can appear
+--   under several house_id values. That is FINE here — we deliberately keep every
+--   variant in the set; dropping dupes would only shrink coverage, never improve it.
+--   The unique index is on house_id (each MV row is one house_id), not on a physical
+--   building, so distinct house_id rows never collide.
+--
+-- Refresh: refreshed by the existing MV-refresh task (app/tasks/refresh_search_matview.py,
+--   scheduled as `refresh_search_matview` in scrape_schedules) right AFTER
+--   listings_search_mv — premium_houses derives from `listings`, so refreshing it once
+--   the search MV is fresh keeps both in step. REFRESH ... CONCURRENTLY requires the
+--   UNIQUE index below.
+--
+-- Dependencies: 002_core_tables.sql (listings: house_id_fk, price_rub, price_per_m2,
+--               area_m2, is_active).
+-- Deploy order: after 002_core_tables.sql. No backend schema coupling.
+-- Re-run safe: CREATE MATERIALIZED VIEW IF NOT EXISTS + CREATE UNIQUE INDEX IF NOT EXISTS.
+--   No DROP — dropping would break in-flight REFRESH ... CONCURRENTLY and momentarily
+--   delete the object during deploy.
+
+BEGIN;
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS premium_houses AS
+WITH lh AS (
+    SELECT l.house_id_fk AS house_id,
+        count(*) AS n,
+        count(*) FILTER (WHERE l.price_rub >= 20000000) AS n_elite,
+        percentile_cont(0.5) WITHIN GROUP (ORDER BY l.price_per_m2)
+            FILTER (WHERE l.price_per_m2 BETWEEN 100000 AND 800000) AS med_ppm2
+    FROM listings l
+    WHERE l.is_active AND l.house_id_fk IS NOT NULL AND l.area_m2 > 0
+          AND l.price_per_m2 BETWEEN 100000 AND 800000
+    GROUP BY l.house_id_fk
+)
+SELECT house_id, n, n_elite, round(med_ppm2)::int AS med_ppm2
+FROM lh
+WHERE n >= 4 AND (med_ppm2 >= 220000 OR n_elite >= 3);
+
+-- UNIQUE index is mandatory for REFRESH MATERIALIZED VIEW CONCURRENTLY.
+CREATE UNIQUE INDEX IF NOT EXISTS premium_houses_house_id_uidx
+    ON premium_houses (house_id);
+
+COMMIT;

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"]))