feat(mera/estimate): характеристики дома из справочника, а не только из формы

Tier H в `_fetch_analogs` («тот же класс дома»: год ±15 + этажность ±30%)
существует давно, но его докстринг честно предупреждал: «пропускается если
year_built или total_floors неизвестны». Он и пропускался — потому что кормили
его так:

- `total_floors` брался ИСКЛЮЧИТЕЛЬНО из `payload`, фолбэка не было вообще;
- фолбэк года шёл в Overpass/OSM (`get_house_metadata`), а в кэше
  `house_metadata` лежит 68 строк на всю базу.

При этом `target_house_id` резолвится строкой выше — дом известен, просто его
характеристики никто не читал. После наполнения справочника из ГИС ЖКХ и DOM.РФ
в `houses` год есть у 86.8% домов, этажность — у 98.3%.

Теперь приоритет: payload → houses → OSM. Пользователь по-прежнему главнее
справочника; Overpass дёргается только если после первых двух что-то осталось
неизвестным, то есть в подавляющем большинстве случаев внешний вызов с
таймаутом больше не тратится. Год из справочника проходит тот же
`_sanitize_build_year`, что и раньше.

Замер на 1123 прошлых оценках: 450 шли без года (для 301 справочник его знает),
240 без этажности (знаем для 108). Цена смешения классов домов в пуле аналогов —
на 466 оценках с восстановимым пулом медианный сдвиг медианы 2.28%, p90 = 16.3%,
у 18% оценок больше 10%. Систематического смещения нет: смешение даёт шум.

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

@@ -2405,6 +2405,112 @@ def _fetch_anchor_comps(
     return [], None
 
 
+@dataclass(frozen=True)
+class _HouseFacts:
+    """Характеристики дома из справочника `houses` (#3234).
+
+    total_floors / year_built / house_type — fallback для пропущенных
+    payload-полей ПЕРЕД OSM Overpass (houses на два порядка полнее и не тратит
+    внешний вызов). material_walls читается на будущее — потребителя ниже по
+    estimator пока нет, но данные есть: 7838 домов из 10221 (77%) на проде после
+    наполнения из ГИС ЖКХ + DOM.РФ 2026-08-29.
+    """
+
+    house_id: int
+    total_floors: int | None
+    year_built: int | None
+    house_type: str | None
+    material_walls: str | None
+
+
+def _lookup_house_facts(
+    db: Session,
+    *,
+    target_house_id: int | None,
+    lat: float | None,
+    lon: float | None,
+) -> _HouseFacts | None:
+    """Читает total_floors/year_built/house_type/material_walls из `houses`.
+
+    Ключ — canonical target_house_id (уже резолвлен match_house_readonly выше
+    по estimate-функции); если он None — фолбэк на ближайший дом в радиусе 60м
+    от subject lat/lon (ST_DWithin + сортировка по расстоянию, LIMIT 1).
+
+    Best-effort: любая ошибка БД → None + WARNING, оценка не должна падать
+    (тот же паттерн try/except + db.rollback(), см. _fetch_anchor_comps выше).
+    """
+    try:
+        if target_house_id is not None:
+            row = (
+                db.execute(
+                    text(
+                        """
+                        SELECT id, total_floors, year_built, house_type, material_walls
+                        FROM houses
+                        WHERE id = CAST(:hid AS bigint)
+                        """
+                    ),
+                    {"hid": target_house_id},
+                )
+                .mappings()
+                .first()
+            )
+        elif lat is not None and lon is not None:
+            row = (
+                db.execute(
+                    text(
+                        """
+                        SELECT id, total_floors, year_built, house_type, material_walls
+                        FROM houses
+                        WHERE geom IS NOT NULL
+                          AND ST_DWithin(
+                                geom::geography,
+                                ST_MakePoint(:lon, :lat)::geography,
+                                60
+                          )
+                        ORDER BY geom::geography <-> ST_MakePoint(:lon, :lat)::geography
+                        LIMIT 1
+                        """
+                    ),
+                    {"lon": lon, "lat": lat},
+                )
+                .mappings()
+                .first()
+            )
+        else:
+            return None
+    except Exception as exc:  # pragma: no cover — defensive
+        logger.warning("house facts lookup failed (graceful): %s", exc)
+        try:
+            db.rollback()
+        except Exception:  # pragma: no cover — defensive
+            pass
+        return None
+    if row is None:
+        return None
+    # Guard типов строки: houses.id — NOT NULL bigint, любая другая форма (в т.ч.
+    # тестовый db-double без явного мока запроса) трактуется как «строки нет» —
+    # тот же best-effort, что и явная ошибка выше. Остальные поля best-effort
+    # каждое по отдельности — неожиданный тип поля не должен ронять facts целиком.
+    house_id = row["id"]
+    if not isinstance(house_id, int):
+        return None
+
+    def _as_int(value: Any) -> int | None:
+        return value if isinstance(value, int) else None
+
+    def _as_str(value: Any) -> str | None:
+        return value if isinstance(value, str) else None
+
+    return _HouseFacts(
+        house_id=house_id,
+        total_floors=_as_int(row["total_floors"]),
+        year_built=_as_int(row["year_built"]),
+        house_type=_as_str(row["house_type"]),
+        material_walls=_as_str(row["material_walls"]),
+    )
+
+
 def _band_haircut(anchor_ppm2: float) -> float:
     """asking→sold haircut, banded по ppm² (class-label в prod пуст — band на цену).
 
@@ -4040,7 +4146,39 @@ async def estimate_quality(
     #    gateway-таймауту; деградируем в None при превышении budget.
     target_year = payload.year_built
     target_house_type = payload.house_type
-    if target_year is None or target_house_type is None:
+    target_total_floors = payload.total_floors
+
+    # 2a. #3234: houses — наш справочник, приоритет ПЕРЕД OSM Overpass (year
+    #     86.8% / floors 98.3% покрытия на проде против 68 строк в
+    #     house_metadata-кэше). payload остаётся главнее в любом случае — houses
+    #     заполняет только то, что пользователь не указал.
+    if target_year is None or target_house_type is None or target_total_floors is None:
+        house_facts = await asyncio.to_thread(
+            _lookup_house_facts,
+            db,
+            target_house_id=target_house_id,
+            lat=geo.lat,
+            lon=geo.lon,
+        )
+        if house_facts is not None:
+            filled: list[str] = []
+            if target_year is None and house_facts.year_built is not None:
+                target_year = house_facts.year_built
+                filled.append(f"year_built={target_year}")
+            if target_house_type is None and house_facts.house_type is not None:
+                target_house_type = house_facts.house_type
+                filled.append(f"house_type={target_house_type}")
+            if target_total_floors is None and house_facts.total_floors is not None:
+                target_total_floors = house_facts.total_floors
+                filled.append(f"total_floors={target_total_floors}")
+            if filled:
+                logger.info(
+                    "estimate: house facts fallback house_id=%s → %s",
+                    house_facts.house_id,
+                    ", ".join(filled),
+                )
+
+    if target_year is None or target_house_type is None or target_total_floors is None:
         house_meta = await _with_budget(
             get_house_metadata(geo.lat, geo.lon, db),
             settings.estimate_house_meta_timeout_s,
@@ -4051,6 +4189,8 @@ async def estimate_quality(
                 target_year = house_meta.year_built
             if target_house_type is None:
                 target_house_type = house_meta.house_type
+            if target_total_floors is None:
+                target_total_floors = house_meta.total_floors
 
     # 2b. Mera-audit 2026-08-02: неправдоподобный год (payload user-input ge=1800/le=2100 в схеме,
     #     либо house_metadata OSM/кадастр — прод-инцидент year_built=1829) — на
@@ -4097,7 +4237,7 @@ async def estimate_quality(
             target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
-            total_floors=payload.total_floors,
+            total_floors=target_total_floors,
             cohort_year_min=cy_min,
             cohort_year_max=cy_max,
         )
@@ -4122,7 +4262,7 @@ async def estimate_quality(
             target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
-            total_floors=payload.total_floors,
+            total_floors=target_total_floors,
         )
     # #oblast-F: True only when there WAS a cohort (year_built) filter to drop —
     # surfaced later as the "снят фильтр по году постройки" relaxation label.
@@ -4142,7 +4282,7 @@ async def estimate_quality(
             target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
-            total_floors=payload.total_floors,
+            total_floors=target_total_floors,
         )
         if len(listings_wide) > len(listings):
             listings = listings_wide
@@ -4166,7 +4306,7 @@ async def estimate_quality(
             target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
-            total_floors=payload.total_floors,
+            total_floors=target_total_floors,
         )
         if len(listings_widearea) > len(listings):
             listings = listings_widearea
@@ -4225,7 +4365,7 @@ async def estimate_quality(
             target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
-            total_floors=payload.total_floors,
+            total_floors=target_total_floors,
         )
         if len(candidate) > len(listings):
             return candidate, tier

tradein-mvp/backend/tests/test_3234_house_facts_fallback.py

@@ -0,0 +1,294 @@
+"""#3234: `houses` как fallback-источник total_floors/year_built/house_type в estimate.
+
+Приоритет: payload (пользователь) → houses (наш справочник, локальный, ~87%
+покрытия года / ~98% этажности на проде) → OSM Overpass (get_house_metadata,
+внешний вызов с таймаутом, house_metadata-кэш — всего 68 строк на базу).
+
+payload НИКОГДА не переопределяется: houses/OSM заполняют только то, что
+payload не задал. get_house_metadata должен вызываться ТОЛЬКО если после
+payload+houses что-то всё ещё неизвестно (экономия внешнего вызова).
+
+year_built из houses обязан пройти через тот же _sanitize_build_year, что и
+payload/OSM (Mera-audit 2026-08-02, прод-инцидент year_built=1829 — см.
+test_estimator_implausible_year.py).
+
+Два уровня тестов:
+  1. _lookup_house_facts() изолированно — маршрутизация SQL (house_id vs
+     геофолбэк) + best-effort на ошибке БД (без сети).
+  2. estimate_quality() целиком, offline — зеркалит паттерн
+     test_estimator_client_coords.py: все внешние вызовы замоканы,
+     _fetch_analogs захватывает kwargs, чтобы проверить что именно уехало
+     вниз по пайплайну.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+
+# Settings требует DATABASE_URL при импорте app.core.config — до любого app-импорта.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import anyio
+
+# ── Level 1: _lookup_house_facts() изолированно (без estimate_quality) ─────
+
+
+def test_lookup_house_facts_uses_house_id_when_present() -> None:
+    """target_house_id задан → SQL идёт по id, геофолбэк (ST_DWithin) не участвует."""
+    from app.services.estimator import _lookup_house_facts
+
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = {
+        "id": 42,
+        "total_floors": 9,
+        "year_built": 1975,
+        "house_type": "panel",
+        "material_walls": None,
+    }
+    facts = _lookup_house_facts(db, target_house_id=42, lat=None, lon=None)
+
+    assert facts is not None
+    assert (facts.house_id, facts.total_floors, facts.year_built, facts.house_type) == (
+        42,
+        9,
+        1975,
+        "panel",
+    )
+    sql_text = str(db.execute.call_args.args[0])
+    assert "WHERE id = " in sql_text
+    assert "ST_DWithin" not in sql_text
+
+
+def test_lookup_house_facts_falls_back_to_geo_radius_when_no_house_id() -> None:
+    """target_house_id=None → SQL идёт по ST_DWithin(60м) + ближайший к lat/lon."""
+    from app.services.estimator import _lookup_house_facts
+
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = {
+        "id": 7,
+        "total_floors": 5,
+        "year_built": 1968,
+        "house_type": "brick",
+        "material_walls": None,
+    }
+    facts = _lookup_house_facts(db, target_house_id=None, lat=56.838, lon=60.595)
+
+    assert facts is not None
+    assert facts.house_id == 7
+    sql_text = str(db.execute.call_args.args[0])
+    assert "ST_DWithin" in sql_text
+    assert "60" in sql_text
+
+
+def test_lookup_house_facts_no_key_returns_none_without_query() -> None:
+    """Ни house_id, ни lat/lon — запроса к БД нет вообще, сразу None."""
+    from app.services.estimator import _lookup_house_facts
+
+    db = MagicMock()
+    facts = _lookup_house_facts(db, target_house_id=None, lat=None, lon=None)
+
+    assert facts is None
+    db.execute.assert_not_called()
+
+
+def test_lookup_house_facts_db_error_is_graceful() -> None:
+    """Любая ошибка БД → None + rollback, estimate не должен падать (best-effort)."""
+    from app.services.estimator import _lookup_house_facts
+
+    db = MagicMock()
+    db.execute.side_effect = RuntimeError("boom")
+    facts = _lookup_house_facts(db, target_house_id=42, lat=None, lon=None)
+
+    assert facts is None
+    db.rollback.assert_called_once()
+
+
+# ── Level 2: estimate_quality() целиком, offline ────────────────────────────
+
+
+def _make_fake_geo():
+    from app.services.geocoder import GeocodeResult
+
+    return GeocodeResult(
+        lat=56.838,
+        lon=60.595,
+        full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
+        provider="nominatim",
+    )
+
+
+def _make_payload(*, year_built=None, house_type=None, total_floors=None):
+    from app.schemas.trade_in import TradeInEstimateInput
+
+    return TradeInEstimateInput(
+        address="ЕКБ, ул. Учителей, 18",
+        area_m2=38.8,
+        rooms=1,
+        floor=4,
+        lat=56.838,
+        lon=60.595,
+        year_built=year_built,
+        house_type=house_type,
+        total_floors=total_floors,
+    )
+
+
+def _house_facts(*, house_id=777, total_floors=9, year_built=1975, house_type="panel"):
+    from app.services.estimator import _HouseFacts
+
+    return _HouseFacts(
+        house_id=house_id,
+        total_floors=total_floors,
+        year_built=year_built,
+        house_type=house_type,
+        material_walls=None,
+    )
+
+
+def _house_meta(*, total_floors, year_built, house_type):
+    from app.services.house_metadata import HouseMetadata
+
+    return HouseMetadata(
+        lat=56.838,
+        lon=60.595,
+        year_built=year_built,
+        total_floors=total_floors,
+        house_type=house_type,
+        total_units=None,
+        source="osm",
+    )
+
+
+def _run_estimate(payload, *, house_facts, house_meta):
+    """Прогоняет estimate_quality offline. Возвращает (result, _fetch_analogs mock,
+    get_house_metadata mock) — mocks выжившие после await для инспекции call_args."""
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    geocode_mock = AsyncMock(return_value=_make_fake_geo())
+    house_facts_mock = MagicMock(return_value=house_facts)
+    house_meta_mock = AsyncMock(return_value=house_meta)
+    analogs_mock = MagicMock(return_value=([], False, "W"))
+
+    async def _go():
+        with contextlib.ExitStack() as stack:
+            for cm in (
+                patch("app.services.estimator.geocode", new=geocode_mock),
+                patch("app.services.estimator._lookup_house_facts", new=house_facts_mock),
+                patch("app.services.estimator.get_house_metadata", new=house_meta_mock),
+                patch("app.services.estimator._fetch_analogs", new=analogs_mock),
+                patch("app.services.estimator._fetch_deals", return_value=[]),
+                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)),
+            ):
+                stack.enter_context(cm)
+            return await estimate_quality(payload, db)
+
+    result = anyio.run(_go)
+    return result, analogs_mock, house_meta_mock
+
+
+def _floors_seen(mock) -> set:
+    return {c.kwargs["total_floors"] for c in mock.call_args_list}
+
+
+def _years_seen(mock) -> set:
+    return {c.kwargs["year_built"] for c in mock.call_args_list}
+
+
+def _house_types_seen(mock) -> set:
+    return {c.kwargs["house_type"] for c in mock.call_args_list}
+
+
+def test_empty_payload_uses_houses_facts() -> None:
+    """payload пустой, houses знает всё → его значения уезжают в _fetch_analogs."""
+    payload = _make_payload()
+    facts = _house_facts(total_floors=9, year_built=1975, house_type="panel")
+    result, analogs_mock, house_meta_mock = _run_estimate(
+        payload, house_facts=facts, house_meta=None
+    )
+
+    assert analogs_mock.call_args_list, "estimate должен вызвать _fetch_analogs хотя бы раз"
+    assert _floors_seen(analogs_mock) == {9}
+    assert _years_seen(analogs_mock) == {1975}
+    assert _house_types_seen(analogs_mock) == {"panel"}
+    house_meta_mock.assert_not_awaited()
+    assert result.estimate_id is not None
+
+
+def test_payload_wins_over_conflicting_houses_facts() -> None:
+    """payload задан, houses знает ДРУГОЕ → побеждает payload (user главнее houses)."""
+    payload = _make_payload(year_built=2015, house_type="brick", total_floors=5)
+    facts = _house_facts(total_floors=9, year_built=1975, house_type="panel")
+    result, analogs_mock, house_meta_mock = _run_estimate(
+        payload, house_facts=facts, house_meta=None
+    )
+
+    assert _floors_seen(analogs_mock) == {5}
+    assert _years_seen(analogs_mock) == {2015}
+    assert _house_types_seen(analogs_mock) == {"brick"}
+    house_meta_mock.assert_not_awaited()
+    assert result.estimate_id is not None
+
+
+def test_neither_payload_nor_houses_falls_back_to_osm() -> None:
+    """Ни payload, ни houses ничего не знают → OSM get_house_metadata вызывается
+    (старое поведение сохранено, houses тут ничем не помогает)."""
+    payload = _make_payload()
+    meta = _house_meta(total_floors=5, year_built=1968, house_type="brick")
+    result, analogs_mock, house_meta_mock = _run_estimate(
+        payload, house_facts=None, house_meta=meta
+    )
+
+    house_meta_mock.assert_awaited()
+    assert _floors_seen(analogs_mock) == {5}
+    assert _years_seen(analogs_mock) == {1968}
+    assert _house_types_seen(analogs_mock) == {"brick"}
+    assert result.estimate_id is not None
+
+
+def test_payload_and_houses_cover_everything_skips_osm() -> None:
+    """payload частично + houses добивает остальное → OSM НЕ вызывается (экономия
+    внешнего бюджетируемого вызова, п.5 задачи)."""
+    payload = _make_payload(year_built=2015)  # house_type/total_floors придут из houses
+    facts = _house_facts(total_floors=9, year_built=1975, house_type="panel")
+    result, analogs_mock, house_meta_mock = _run_estimate(
+        payload, house_facts=facts, house_meta=None
+    )
+
+    house_meta_mock.assert_not_awaited()
+    assert _floors_seen(analogs_mock) == {9}
+    assert _years_seen(analogs_mock) == {2015}  # payload побеждает для года
+    assert _house_types_seen(analogs_mock) == {"panel"}
+    assert result.estimate_id is not None
+
+
+def test_implausible_year_from_houses_is_sanitized() -> None:
+    """houses отдал неправдоподобный год (1829, прод-паттерн) → _sanitize_build_year
+    отбивает его в None; total_floors/house_type из houses при этом всё равно
+    применяются (год — независимый гейт, не блокирует остальные поля)."""
+    payload = _make_payload()
+    facts = _house_facts(total_floors=9, year_built=1829, house_type="panel")
+    result, analogs_mock, house_meta_mock = _run_estimate(
+        payload, house_facts=facts, house_meta=None
+    )
+
+    assert _years_seen(analogs_mock) == {None}
+    assert _floors_seen(analogs_mock) == {9}
+    assert _house_types_seen(analogs_mock) == {"panel"}
+    house_meta_mock.assert_not_awaited()
+    assert result.estimate_id is not None