fix(tradein/estimator): kitchen_area_m2/ceiling_height_m/is_apartments comp-scoring (#2012)

Three independent feature flags, all default OFF pending live A/B
backtest (no DB access in worker/main sessions this round — see PR
body for exact before/after commands to run once available):

- estimate_kitchen_area_signal_enabled / estimate_ceiling_height_signal_enabled:
  house_type/year_built have a user-input target to compare candidates
  against; kitchen_area_m2/ceiling_height_m have neither a target
  (TradeInEstimateInput doesn't collect them) nor ground-truth on
  rosreestr deals (backtest DealSample lacks these columns). Inventing
  an external "typical" constant would be unfounded, so the penalty is
  self-referential: deviation from the CANDIDATE POOL's own median in
  Tier H/W relevance_score. NULL-safe (no penalty, excluded from
  median) and sparse-safe (fully skipped when <5 non-null values in
  the pool).
- estimate_is_apartments_filter_enabled: hard-filter in _COMMON_WHERE +
  Tier W, symmetric to the existing novostroyki-guard (#1186). NULL
  passes through untouched, only explicit true excluded.

Review fixup: _stratify_candidates requires its input pre-sorted by
relevance_score (its per-source guaranteed-quota selection depends on
that order). _apply_kitchen_ceiling_signal mutates relevance_score
in-place after the SQL fetch but before stratify — added an explicit
re-sort in both call sites (Tier H, Tier W) so the diversity-quota
selection sees the post-adjustment order once either flag ships live.
Currently a no-op (flags off), but was a latent footgun for the first
PR to flip one on.

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

@@ -307,6 +307,61 @@ class Settings(BaseSettings):
     # кросс-постингом ×3. ENV: ESTIMATE_DEDUP_ANALOGS_ENABLED (=false откатывает).
     estimate_dedup_analogs_enabled: bool = True
 
+    # ── #2012: kitchen_area_m2 / ceiling_height_m / is_apartments comp-scoring ──
+    # Follow-up к #2007/#2008/#2009 (промоутят поля в колонки). До этой правки
+    # estimator читал house_type ТОЛЬКО как soft-penalty, а kitchen_area_m2 /
+    # ceiling_height_m / is_apartments НЕ читал вовсе для отбора/скоринга
+    # аналогов. Три НЕЗАВИСИМЫХ флага (по одному на фичу, все default OFF —
+    # см. scripts/backtest_estimator.py --engine full для A/B измерения; каждый
+    # PR/issue #2012 обязан задокументировать MAPE/coverage/calibration до
+    # включения любого в default ON):
+    #
+    # kitchen_area_m2 / ceiling_height_m ("мягкие корректировки"): в отличие от
+    # house_type/year_built (сравниваются с target_house_type/target_year,
+    # известными из payload или OSM house_metadata fallback) — у kitchen/ceiling
+    # НЕТ target-значения: ни TradeInEstimateInput (форма пользователя), ни
+    # `deals` (backtest ground truth, rosreestr ДКП) их не несут. Поэтому
+    # релевантность штрафуется отклонением кандидата от МЕДИАНЫ САМОГО ПУЛА
+    # кандидатов текущего запроса (self-referential), а НЕ сравнением с внешней
+    # "типичной" константой — так сигнал не зависит от непроверенных допущений
+    # о типичном размере кухни/высоте потолка. NULL-safe и sparse-safe вдвойне:
+    # (1) кандидат без значения колонки не штрафуется и не участвует в подсчёте
+    # медианы пула; (2) если кандидатов пула с непустым значением меньше
+    # estimate_kitchen_ceiling_signal_min_n — сигнал пропускается ЦЕЛИКОМ для
+    # всего пула (слишком мало данных для честной "типичной" медианы — риск шума
+    # на sparse-колонках, который явно называет issue #2012: kitchen 4-99%,
+    # ceiling ~10% покрытия по источникам).
+    # ENV: ESTIMATE_KITCHEN_AREA_SIGNAL_ENABLED, ESTIMATE_CEILING_HEIGHT_SIGNAL_ENABLED.
+    estimate_kitchen_area_signal_enabled: bool = False
+    estimate_ceiling_height_signal_enabled: bool = False
+    # Масштаб (м² / м): во сколько "единиц отклонения" превращается 1.0 очко
+    # relevance_score — симметрично house_type-штрафу (1.5 очка за несовпадение)
+    # и year_built-штрафу (abs(delta)/12.0). Кухня ~3м² и потолок ~0.3м —
+    # консервативные масштабы, дающие умеренный штраф на типичном разбросе пула.
+    estimate_kitchen_area_scale: float = 3.0
+    estimate_ceiling_height_scale: float = 0.3
+    # Максимальный штраф за отклонение (та же единица очков, что house_type=1.5) —
+    # клампим, чтобы редкий выброс пула (напр. кухня 25м² в студийной подборке)
+    # не выбрасывал кандидата из top-50 целиком одним лишь этим сигналом.
+    estimate_kitchen_ceiling_signal_max_penalty: float = 1.0
+    # Минимум кандидатов пула с НЕ-NULL значением колонки, чтобы доверять её
+    # медиане как "типичной" для этого пула (иначе сигнал пропускается — см. риск
+    # sparse-coverage выше).
+    estimate_kitchen_ceiling_signal_min_n: int = 5
+    #
+    # is_apartments (#2008): концептуально ОТДЕЛЬНАЯ фича — не "мягкая
+    # корректировка", а hard-filter сегмент-guard, симметричный novostroyki-guard
+    # #1186 (`listing_segment`) в _COMMON_WHERE. Апартаменты — юридически иной
+    # статус недвижимости (не жилое помещение, нет постоянной регистрации по
+    # месту жительства), заметно иная ценовая модель vs обычная квартира. Target
+    # trade-in объект почти всегда обычная квартира (TradeInEstimateInput не
+    # даёт признака "апартаменты"), поэтому при включении флага HARD-исключаем
+    # явно known is_apartments=true кандидатов из вторичка-пула. NULL-safe:
+    # неизвестный статус (подавляющее большинство строк, sparse coverage)
+    # участвует БЕЗ штрафа — фильтруются только явные True.
+    # ENV: ESTIMATE_IS_APARTMENTS_FILTER_ENABLED.
+    estimate_is_apartments_filter_enabled: bool = False
+
     # ── #1871 P2: split-дома wide-corridor disclosure (default ON, порог 1.2) ──
     # Tier A (same-building) матчит по address-regex (намеренно НЕ house_id — дом
     # дробится на несколько house_id). На split-доме разной этажности comp_min..max

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

@@ -25,6 +25,7 @@ import json
 import logging
 import math
 import re
+import statistics
 import time
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass
@@ -4126,13 +4127,84 @@ def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any
     return result[:50]
 
 
+def _adjust_relevance_by_pool_deviation(
+    candidates: list[dict[str, Any]],
+    *,
+    key: str,
+    scale: float,
+    max_penalty: float,
+    min_n: int,
+) -> None:
+    """#2012 comp-scoring: penalize deviation from the CANDIDATE POOL's own median.
+
+    kitchen_area_m2 / ceiling_height_m have NO target value to compare against —
+    unlike house_type/year_built (compared to target_house_type/target_year,
+    sourced from payload or OSM house_metadata fallback), neither
+    ``TradeInEstimateInput`` (user form) nor ``deals`` (backtest ground truth,
+    rosreestr ДКП) carry a kitchen/ceiling value for the property being priced.
+    So instead of a target-diff soft-penalty, this penalizes a candidate's
+    deviation from the MEDIAN of the same pool of candidates being scored right
+    now (self-referential) — no external "typical" constant is assumed.
+
+    Mutates each candidate's ``relevance_score`` IN PLACE (same convention as the
+    SQL house_type/year_built CASE terms: higher = less relevant). NULL-safe:
+    a candidate missing ``key`` is neither penalized nor counted toward the pool
+    median. Sparse-safe: if fewer than ``min_n`` candidates carry a non-NULL
+    value, the signal is skipped for the ENTIRE pool — too few examples to trust
+    a "typical" median (see #2012 sparse-coverage risk: kitchen 4-99%, ceiling
+    ~10% coverage across sources).
+    """
+    values = [float(c[key]) for c in candidates if c.get(key) is not None]
+    if len(values) < min_n:
+        return
+    pool_median = statistics.median(values)
+    for c in candidates:
+        v = c.get(key)
+        if v is None:
+            continue
+        penalty = min(abs(float(v) - pool_median) / scale, max_penalty)
+        c["relevance_score"] = (c.get("relevance_score") or 0.0) + penalty
+
+
+def _apply_kitchen_ceiling_signal(candidates: list[dict[str, Any]]) -> None:
+    """#2012: apply the kitchen_area_m2 / ceiling_height_m comp-scoring signals.
+
+    Each is an INDEPENDENT feature flag (default OFF — see config.py). No-op
+    when both flags are off (byte-identical to pre-#2012 behaviour). Called only
+    from the Tier H / Tier W paths of ``_fetch_analogs`` — Tier S (same
+    building) is intentionally excluded, symmetric with house_type/year_built,
+    which also don't participate in Tier S relevance (fixed 0.0 there).
+    """
+    if settings.estimate_kitchen_area_signal_enabled:
+        _adjust_relevance_by_pool_deviation(
+            candidates,
+            key="kitchen_area_m2",
+            scale=settings.estimate_kitchen_area_scale,
+            max_penalty=settings.estimate_kitchen_ceiling_signal_max_penalty,
+            min_n=settings.estimate_kitchen_ceiling_signal_min_n,
+        )
+    if settings.estimate_ceiling_height_signal_enabled:
+        _adjust_relevance_by_pool_deviation(
+            candidates,
+            key="ceiling_height_m",
+            scale=settings.estimate_ceiling_height_scale,
+            max_penalty=settings.estimate_kitchen_ceiling_signal_max_penalty,
+            min_n=settings.estimate_kitchen_ceiling_signal_min_n,
+        )
+
+
 _ANALOG_SELECT_COLS = """
     source, source_url, address, lat, lon,
     rooms, area_m2, floor, total_floors,
     price_rub, price_per_m2,
     listing_date, days_on_market, photo_urls,
     scraped_at,
-    building_cadastral_number
+    building_cadastral_number,
+    -- #2012: comp-scoring сигналы (kitchen_area_m2/ceiling_height_m). Только
+    -- Tier H/W применяют их (см. _apply_kitchen_ceiling_signal) — Tier S
+    -- (same building) не трогают, симметрично house_type/year_built, которые
+    -- тоже не участвуют в Tier S relevance (там фиксированный 0.0).
+    kitchen_area_m2, ceiling_height_m
 """
 
 _COMMON_WHERE = """
@@ -4156,6 +4228,17 @@ _COMMON_WHERE = """
     -- медиану ₽/м². NULL сегмент пропускаем (rosreestr/avito/yandex без сегмента —
     -- это вторичка или неклассифицированный объект).
     AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
+    -- #2012 is_apartments hard-filter (флаг estimate_is_apartments_filter_enabled,
+    -- default OFF pending backtest). Флаг выключен ⇒ CAST(... ) IS NOT TRUE ⇒
+    -- условие прозрачно (byte-identical старому поведению). Включён ⇒ исключает
+    -- ТОЛЬКО явные is_apartments=true (апартаменты — иной ценовой/юридический
+    -- сегмент). NULL (подавляющее большинство строк, sparse coverage) проходит
+    -- без штрафа — не наказываем отсутствие данных.
+    AND (
+        CAST(:is_apartments_filter AS boolean) IS NOT TRUE
+        OR is_apartments IS NULL
+        OR is_apartments = false
+    )
 """
 # Note: Tier W has its own inline copy of the cohort clause (PR #519 line
 # ~1280). Не удалять — Tier W не использует _COMMON_WHERE из-за inline
@@ -4226,6 +4309,11 @@ def _fetch_analogs(
     NULL допускается чтобы не отсеивать листинги с неизвестным годом
     (типично для Avito anonymous-address объявлений).
 
+    #2012 comp-scoring (все флаги default OFF, см. config.py):
+      - is_apartments hard-filter в WHERE (все тиры, симметрично novostroyki-guard).
+      - kitchen_area_m2 / ceiling_height_m soft self-referential pool-median
+        penalty в Tier H/W (не Tier S) — см. _apply_kitchen_ceiling_signal.
+
     Returns:
         (list_of_listings_as_dicts, fallback_radius_used_flag, tier)
         tier: 'S' | 'H' | 'W'
@@ -4245,6 +4333,8 @@ def _fetch_analogs(
         "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
         "cohort_year_min": cohort_year_min,
         "cohort_year_max": cohort_year_max,
+        # #2012: is_apartments hard-filter — see _COMMON_WHERE comment above.
+        "is_apartments_filter": settings.estimate_is_apartments_filter_enabled,
     }
 
     # ── Tier S (canonical): same building via house_id_fk ─────────────────────
@@ -4424,7 +4514,8 @@ def _fetch_analogs(
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls,
                     scraped_at, distance_m, relevance_score,
-                    building_cadastral_number
+                    building_cadastral_number,
+                    kitchen_area_m2, ceiling_height_m
                 FROM base
                 WHERE rn_addr <= :max_per_addr
                 {dup_filter}
@@ -4456,6 +4547,8 @@ def _fetch_analogs(
         )
 
         tier_h = [dict(r) for r in tier_h_rows]
+        _apply_kitchen_ceiling_signal(tier_h)
+        tier_h.sort(key=lambda r: r.get("relevance_score") or 0.0)
         if len(tier_h) >= 5:
             logger.info(
                 "analogs tier=H year=%d±15 tf=%d-%d → %d results",
@@ -4487,6 +4580,7 @@ def _fetch_analogs(
                     listing_date, days_on_market, photo_urls,
                     scraped_at,
                     building_cadastral_number,
+                    kitchen_area_m2, ceiling_height_m,
                     id,
                     ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                         AS distance_m,
@@ -4552,6 +4646,13 @@ def _fetch_analogs(
                   -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
                   -- Tier W: исключаем новостройки из comp-пула (sync с _COMMON_WHERE).
                   AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
+                  -- #2012 is_apartments hard-filter, sync с _COMMON_WHERE (см. комментарий
+                  -- там же). Флаг выключен ⇒ прозрачно (byte-identical старому поведению).
+                  AND (
+                      CAST(:is_apartments_filter AS boolean) IS NOT TRUE
+                      OR is_apartments IS NULL
+                      OR is_apartments = false
+                  )
                   -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после
                   -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin
                   -- (geom IS NULL → не matches). geocode-missing-listings backfill
@@ -4568,6 +4669,7 @@ def _fetch_analogs(
                 listing_date, days_on_market, photo_urls,
                 scraped_at,
                 building_cadastral_number,
+                kitchen_area_m2, ceiling_height_m,
                 distance_m,
                 relevance_score
             FROM base
@@ -4594,6 +4696,7 @@ def _fetch_analogs(
                 "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
                 "cohort_year_min": cohort_year_min,  # NEW
                 "cohort_year_max": cohort_year_max,  # NEW
+                "is_apartments_filter": settings.estimate_is_apartments_filter_enabled,  # #2012
             },
         )
         .mappings()
@@ -4601,6 +4704,8 @@ def _fetch_analogs(
     )
 
     candidates: list[dict[str, Any]] = [dict(r) for r in tier_w_rows]
+    _apply_kitchen_ceiling_signal(candidates)
+    candidates.sort(key=lambda r: r.get("relevance_score") or 0.0)
     logger.info("analogs tier=W radius=%dm → %d candidates", radius_m, len(candidates))
     return _stratify_candidates(candidates), radius_m > DEFAULT_RADIUS_M, "W"
 

tradein-mvp/backend/tests/test_estimator_kitchen_ceiling_2012.py

@@ -0,0 +1,248 @@
+"""#2012 — kitchen_area_m2 / ceiling_height_m / is_apartments comp-scoring.
+
+Follow-up к #2007/#2008/#2009 (промоутят поля в колонки). До этой правки
+estimator читал house_type ТОЛЬКО как soft-penalty, а kitchen_area_m2 /
+ceiling_height_m / is_apartments НЕ читал вовсе для отбора/скоринга аналогов.
+
+Три независимых флага, все default OFF:
+  - estimate_kitchen_area_signal_enabled / estimate_ceiling_height_signal_enabled
+    ("мягкие корректировки") — pure-Python self-referential pool-median
+    deviation penalty (см. _adjust_relevance_by_pool_deviation). НЕТ target-
+    значения для сравнения (ни TradeInEstimateInput, ни `deals` его не несут),
+    поэтому — в отличие от house_type/year_built — штраф считается от МЕДИАНЫ
+    ПУЛА кандидатов, а не от target.
+  - estimate_is_apartments_filter_enabled — hard-filter в _COMMON_WHERE (+ Tier W
+    inline copy), симметричный novostroyki-guard #1186. SQL-фрагмент проверяется
+    на сгенерированном тексте (mock db, паттерн test_estimator_radius_dedup_1871.py)
+    — полный radius-путь требует PostGIS+БД.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+
+import app.services.estimator as est
+
+# --------------------------------------------------------------------------- #
+# _adjust_relevance_by_pool_deviation — pure, no DB
+# --------------------------------------------------------------------------- #
+
+
+def _cands(values: list[float | None], key: str = "kitchen_area_m2") -> list[dict]:
+    return [{key: v, "relevance_score": 0.0} for v in values]
+
+
+def test_pool_deviation_null_safe_missing_key_not_penalized_or_counted() -> None:
+    # 5 candidates carry a value (>= min_n), 2 don't -> those 2 stay untouched.
+    cands = _cands([9.0, 9.0, 9.0, 9.0, 9.0, None, None])
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=3.0, max_penalty=1.0, min_n=5
+    )
+    for c in cands[:5]:
+        assert c["relevance_score"] == 0.0  # at the pool median -> no penalty
+    for c in cands[5:]:
+        assert c["relevance_score"] == 0.0  # missing value -> untouched, not 0-diff
+
+
+def test_pool_deviation_sparse_safe_skips_when_below_min_n() -> None:
+    # Only 3 candidates carry a value, min_n=5 -> signal skipped for the WHOLE pool.
+    cands = _cands([5.0, 20.0, 5.0])
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=3.0, max_penalty=1.0, min_n=5
+    )
+    assert all(c["relevance_score"] == 0.0 for c in cands)
+
+
+def test_pool_deviation_penalizes_far_from_median() -> None:
+    # Median of [8, 9, 9, 9, 10] = 9. Deviant 20 -> penalty = |20-9|/3.0 = 3.667,
+    # clamped to max_penalty=1.0.
+    cands = _cands([8.0, 9.0, 9.0, 9.0, 10.0, 20.0])
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=3.0, max_penalty=1.0, min_n=5
+    )
+    assert cands[0]["relevance_score"] == pytest.approx(1.0 / 3.0)  # |8-9|/3
+    assert cands[1]["relevance_score"] == 0.0  # at median
+    assert cands[-1]["relevance_score"] == 1.0  # clamped
+
+
+def test_pool_deviation_max_penalty_clamp() -> None:
+    cands = _cands([1.0, 1.0, 1.0, 1.0, 1.0, 100.0])
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=1.0, max_penalty=0.5, min_n=5
+    )
+    assert cands[-1]["relevance_score"] == 0.5
+
+
+def test_pool_deviation_additive_not_overwriting_existing_score() -> None:
+    """Mutation ADDS to relevance_score (e.g. house_type SQL penalty already there),
+    it never overwrites it."""
+    cands = [
+        {"kitchen_area_m2": 9.0, "relevance_score": 1.5},  # e.g. house_type mismatch
+        {"kitchen_area_m2": 9.0, "relevance_score": 1.5},
+        {"kitchen_area_m2": 9.0, "relevance_score": 0.0},
+        {"kitchen_area_m2": 9.0, "relevance_score": 0.0},
+        {"kitchen_area_m2": 20.0, "relevance_score": 0.0},
+    ]
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=3.0, max_penalty=1.0, min_n=5
+    )
+    assert cands[0]["relevance_score"] == pytest.approx(1.5)  # at median, +0
+    assert cands[-1]["relevance_score"] == pytest.approx(1.0)  # median=9, |20-9|/3 clamped to 1.0
+
+
+def test_pool_deviation_missing_relevance_score_key_defaults_to_zero() -> None:
+    cands = [{"kitchen_area_m2": v} for v in [8.0, 9.0, 9.0, 9.0, 10.0]]
+    est._adjust_relevance_by_pool_deviation(
+        cands, key="kitchen_area_m2", scale=3.0, max_penalty=1.0, min_n=5
+    )
+    assert all("relevance_score" in c for c in cands)
+
+
+# --------------------------------------------------------------------------- #
+# _apply_kitchen_ceiling_signal — flag wiring
+# --------------------------------------------------------------------------- #
+
+
+def _mixed_pool() -> list[dict]:
+    return [
+        {"kitchen_area_m2": 8.0, "ceiling_height_m": 2.7, "relevance_score": 0.0},
+        {"kitchen_area_m2": 9.0, "ceiling_height_m": 2.7, "relevance_score": 0.0},
+        {"kitchen_area_m2": 9.0, "ceiling_height_m": 2.7, "relevance_score": 0.0},
+        {"kitchen_area_m2": 9.0, "ceiling_height_m": 2.7, "relevance_score": 0.0},
+        {"kitchen_area_m2": 20.0, "ceiling_height_m": 4.5, "relevance_score": 0.0},
+    ]
+
+
+def test_apply_signal_noop_when_both_flags_off() -> None:
+    cands = _mixed_pool()
+    with (
+        patch.object(est.settings, "estimate_kitchen_area_signal_enabled", False),
+        patch.object(est.settings, "estimate_ceiling_height_signal_enabled", False),
+    ):
+        est._apply_kitchen_ceiling_signal(cands)
+    assert all(c["relevance_score"] == 0.0 for c in cands)
+
+
+def test_apply_signal_kitchen_only() -> None:
+    cands = _mixed_pool()
+    with (
+        patch.object(est.settings, "estimate_kitchen_area_signal_enabled", True),
+        patch.object(est.settings, "estimate_ceiling_height_signal_enabled", False),
+    ):
+        est._apply_kitchen_ceiling_signal(cands)
+    assert cands[-1]["relevance_score"] > 0.0  # kitchen outlier (20) penalized
+    for c in cands[1:4]:
+        assert c["relevance_score"] == 0.0  # at pool median (9.0) -> untouched
+
+
+def test_apply_signal_ceiling_only() -> None:
+    cands = _mixed_pool()
+    with (
+        patch.object(est.settings, "estimate_kitchen_area_signal_enabled", False),
+        patch.object(est.settings, "estimate_ceiling_height_signal_enabled", True),
+    ):
+        est._apply_kitchen_ceiling_signal(cands)
+    assert cands[-1]["relevance_score"] > 0.0  # ceiling outlier penalized
+    for c in cands[:4]:
+        assert c["relevance_score"] == 0.0
+
+
+# --------------------------------------------------------------------------- #
+# Defaults
+# --------------------------------------------------------------------------- #
+
+
+def test_new_flags_default_off() -> None:
+    from app.core.config import settings
+
+    assert settings.estimate_kitchen_area_signal_enabled is False
+    assert settings.estimate_ceiling_height_signal_enabled is False
+    assert settings.estimate_is_apartments_filter_enabled is False
+
+
+# --------------------------------------------------------------------------- #
+# is_apartments hard-filter — SQL-fragment (mock db, no PostGIS/DB needed)
+# --------------------------------------------------------------------------- #
+
+
+def _capture_tier_sql_and_params(*, is_apartments_filter: bool) -> list[tuple[str, dict]]:
+    """Runs _fetch_analogs with a mock db, returns (sql_text, params) per tier.
+
+    target_house_id + short_addr + year/floors are all set so SQL for ALL four
+    tiers renders (each tier returns [] -> fallthrough to the next).
+    """
+    captured: list[tuple[str, dict]] = []
+    db = MagicMock()
+
+    def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
+        params = args[1] if len(args) > 1 else {}
+        captured.append((str(args[0].text), dict(params)))
+        result = MagicMock()
+        result.mappings.return_value.all.return_value = []
+        return result
+
+    db.execute.side_effect = side_effect
+
+    with patch.object(est.settings, "estimate_is_apartments_filter_enabled", is_apartments_filter):
+        est._fetch_analogs(
+            db,
+            lat=56.83,
+            lon=60.6,
+            rooms=2,
+            area=50.0,
+            radius_m=2000,
+            full_address="г Екатеринбург, ул Малышева, д 30",
+            year_built=2010,
+            house_type="монолит",
+            total_floors=20,
+            target_house_id=123,
+        )
+    return captured
+
+
+def test_all_four_tiers_render_with_is_apartments_guard() -> None:
+    calls = _capture_tier_sql_and_params(is_apartments_filter=False)
+    assert len(calls) == 4, "ожидаем S-canonical, S-fallback, H, W"
+
+
+def test_is_apartments_bind_param_present_in_every_tier() -> None:
+    calls = _capture_tier_sql_and_params(is_apartments_filter=False)
+    for i, (sql, params) in enumerate(calls):
+        assert ":is_apartments_filter" in sql, f"tier#{i} без is_apartments-гварда"
+        assert "is_apartments_filter" in params, f"tier#{i}: параметр не передан"
+
+
+def test_is_apartments_null_safe_guard_sql_present() -> None:
+    calls = _capture_tier_sql_and_params(is_apartments_filter=True)
+    for i, (sql, _params) in enumerate(calls):
+        assert "IS NOT TRUE" in sql, f"tier#{i}: гейт по флагу отсутствует"
+        assert "is_apartments IS NULL" in sql, f"tier#{i}: NULL-safe пропуск отсутствует"
+        assert "is_apartments = false" in sql, f"tier#{i}: явный False-пропуск отсутствует"
+
+
+def test_is_apartments_param_value_reflects_setting() -> None:
+    calls_off = _capture_tier_sql_and_params(is_apartments_filter=False)
+    for i, (_sql, params) in enumerate(calls_off):
+        assert params["is_apartments_filter"] is False, f"tier#{i}"
+
+    calls_on = _capture_tier_sql_and_params(is_apartments_filter=True)
+    for i, (_sql, params) in enumerate(calls_on):
+        assert params["is_apartments_filter"] is True, f"tier#{i}"
+
+
+def test_no_bind_param_double_colon_cast() -> None:
+    """psycopg3-инвариант: только column::type (не :bind::type)."""
+    import re
+
+    bad = re.compile(r":[a-z_]+::[a-z]")
+    for i, (sql, _params) in enumerate(_capture_tier_sql_and_params(is_apartments_filter=True)):
+        assert not bad.search(sql), f"tier#{i}: найден запрещённый :bind::type"
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))