diff --git a/backend/app/services/site_finder/competitors.py b/backend/app/services/site_finder/competitors.py
index 72252775..57be4814 100644
--- a/backend/app/services/site_finder/competitors.py
+++ b/backend/app/services/site_finder/competitors.py
@@ -470,28 +470,48 @@ _AVG_PRICE_SQL = text("""
 # UNDERCOUNT (в репортнутом кейсе ~1600 ед. недосчёта): почти все конкуренты
 # получали flats_sold≈0 → stage_at_horizon вырождался в нейтраль/«ничего не
 # продано», и блок конкурентов РАСХОДИЛСЯ с MarketMetrics, который считает sold
-# по objective_lots.contract_date.
+# по objective_lots.
 #
-# FIX: считаем sold ИЗ ТОГО ЖЕ ИСТОЧНИКА, что MarketMetrics (_STOCK_SQL /
-# _SALES_WINDOW_SQL): objective_lots.contract_date IS NOT NULL — «Дата договора»
-# заполнена у 100% проданных лотов (cumulative-sold, как n_sold в _STOCK_SQL),
-# поэтому совпадает с рыночными метриками и не недосчитывает.
+# FIX: считаем sold ИЗ ТОГО ЖЕ ИСТОЧНИКА И ПО ТОМУ ЖЕ ПРЕДИКАТУ, что
+# MarketMetrics (_STOCK_SQL): лот продан, если
+#   is_sold IS TRUE OR contract_date IS NOT NULL OR LOWER(status)='продан'
+# (Объектив заполняет эти поля неконсистентно, поэтому берём ИХ ИЛИ; см.
+# market_metrics._STOCK_SQL:285-289). Так flats_sold побайтово зеркалит n_sold.
 #
-# Мост domrf-obj → лоты — РОВНО тот же, что в _OBJECTIVE_PRICE_FALLBACK_SQL
+# Мост domrf-obj → лоты — тот же, что в _OBJECTIVE_PRICE_FALLBACK_SQL
 # (primary_price CTE): domrf_kn_objects.obj_id == objective_complex_mapping
 # .domrf_obj_id → objective_complex_mapping.objective_complex_name ==
-# objective_lots.project_name. Маппинг 1:1 (прод: 263 строки, 263 distinct
-# obj_id, 0 obj_id с >1 именем) → fan-out по obj_id нет. objective_lots —
-# current-state UPSERT (ON CONFLICT (objective_lot_id) DO UPDATE, ровно одна
-# строка на лот) → snapshot-дедуп НЕ нужен (как и в market_metrics._STOCK_SQL).
+# objective_lots.project_name.
+#
+# ВАЖНО (code-review #1927): objective_complex_mapping UNIQUE — это
+# (objective_complex_name, objective_group), а НЕ domrf_obj_id (68_schema:321).
+# Значит ОДИН obj_id может иметь НЕСКОЛЬКО mapping-строк (то же имя, разный
+# objective_group) → наивный COUNT(*) по cm JOIN ol задвоил бы лоты (GROUP BY
+# схлопывает выходные строки, но не дубли ВНУТРИ агрегата). Поэтому
+# COUNT(DISTINCT ol.objective_lot_id) — дедуп на уровне лота, fan-out-safe.
+# (_OBJECTIVE_PRICE_FALLBACK не прецедент: percentile/median dup-нечувствительны.)
+#
+# premise_kind='квартира' — как в _STOCK_SQL: без него проданные паркинги/
+# кладовые/нежилое считались бы как проданные квартиры (over-count).
+# objective_lots — current-state UPSERT (ON CONFLICT (objective_lot_id) DO
+# UPDATE, ровно одна строка на лот) → snapshot-дедуп НЕ нужен (как в _STOCK_SQL).
+#
+# Дефолт premise_kind зеркалит market_metrics.compute_market_metrics
+# (premise_kind="квартира") — оба считают продажи жилых лотов.
+_SOLD_PREMISE_KIND = "квартира"
 _SOLD_COUNT_SQL = text("""
     SELECT
         cm.domrf_obj_id AS obj_id,
-        COUNT(*) FILTER (WHERE ol.contract_date IS NOT NULL) AS flats_sold
+        COUNT(DISTINCT ol.objective_lot_id) FILTER (
+            WHERE ol.is_sold IS TRUE
+               OR ol.contract_date IS NOT NULL
+               OR LOWER(COALESCE(ol.status, '')) = 'продан'
+        ) AS flats_sold
     FROM objective_complex_mapping cm
     JOIN objective_lots ol
         ON ol.project_name = cm.objective_complex_name
     WHERE cm.domrf_obj_id = ANY(:obj_ids)
+      AND ol.premise_kind = :premise_kind
     GROUP BY cm.domrf_obj_id
 """)
 
@@ -723,15 +743,20 @@ def get_competitors(
     except Exception:
         logger.warning("competitors: avg_price query failed, continuing without prices")
 
-    # ── 4·sold. flats_sold из objective_lots.contract_date (#1926) ───────────
+    # ── 4·sold. flats_sold из objective_lots (#1926) ─────────────────────────
     # flats_sold нужен для stage_at_horizon (available = flat_count - flats_sold).
-    # Источник выровнен с MarketMetrics (contract_date 100%-заполнен у проданных),
-    # см. _SOLD_COUNT_SQL — ушли от domrf_kn_flats.status (0.2%, недосчёт).
-    # Отдельный db.execute (не ломает мок-тесты _AVG_PRICE_SQL); _row_get
-    # graceful: отсутствующий ключ flats_sold → None → нейтральный stage.
+    # Источник И предикат выровнены с MarketMetrics._STOCK_SQL (is_sold/
+    # contract_date/status='продан', premise_kind='квартира'), см. _SOLD_COUNT_SQL
+    # — ушли от domrf_kn_flats.status (0.2%, недосчёт). Отдельный db.execute (не
+    # ломает мок-тесты _AVG_PRICE_SQL); _row_get graceful: отсутствующий ключ
+    # flats_sold → None → нейтральный stage.
     sold_count_map: dict[int, int] = {}
     try:
-        sold_rows = db.execute(_SOLD_COUNT_SQL, {"obj_ids": obj_ids}).mappings().all()
+        sold_rows = (
+            db.execute(_SOLD_COUNT_SQL, {"obj_ids": obj_ids, "premise_kind": _SOLD_PREMISE_KIND})
+            .mappings()
+            .all()
+        )
         for r in sold_rows:
             oid = int(r["obj_id"])
             sold = _row_get(r, "flats_sold")
diff --git a/backend/tests/api/v1/test_parcel_competitors.py b/backend/tests/api/v1/test_parcel_competitors.py
index e5da088d..6540d6ad 100644
--- a/backend/tests/api/v1/test_parcel_competitors.py
+++ b/backend/tests/api/v1/test_parcel_competitors.py
@@ -394,14 +394,16 @@ def test_competitors_sold_pct_null() -> None:
         app.dependency_overrides.clear()
 
 
-def test_competitors_flats_sold_from_contract_date() -> None:
-    """flats_sold/sold_pct заполняются из objective_lots.contract_date (#1926).
+def test_competitors_flats_sold_from_objective_lots() -> None:
+    """flats_sold/sold_pct заполняются из objective_lots (#1926).
 
     Регрессия аудита #1871/#1926: РАНЬШЕ flats_sold считался из
     domrf_kn_flats.status (LIKE '%прод%'), заполненного ~0.2% → массовый
-    недосчёт и расхождение с MarketMetrics. Теперь источник — contract_date
-    (100%-заполнен у проданных), как в market_metrics._STOCK_SQL. Здесь
-    sold-count запрос возвращает 120 проданных из 200 → sold_pct 60.0%.
+    недосчёт и расхождение с MarketMetrics. Теперь источник И предикат — как
+    в market_metrics._STOCK_SQL (is_sold OR contract_date OR status='продан',
+    premise_kind='квартира'). Здесь sold-count запрос возвращает 120 проданных
+    из 200 → sold_pct 60.0%. (SQL-структуру предиката/fan-out/premise_kind
+    проверяют test_sold_count_sql_* ниже — мок тут SQL не исполняет.)
     """
     rows = [_obj_row(obj_id=1, flat_count=200)]
     sold_rows = [_sold_row(obj_id=1, flats_sold=120)]
@@ -571,3 +573,49 @@ def test_competitors_is_active_flag() -> None:
         assert comps[3]["is_active"] is False
     finally:
         app.dependency_overrides.clear()
+
+
+# ── _SOLD_COUNT_SQL структура (#1926 / code-review #1927) ───────────────────────
+# Mock-тесты выше НЕ исполняют реальный SQL (db.execute замокан) → не ловят
+# fan-out маппинга, отсутствие premise_kind-фильтра и рассинхрон sold-предиката
+# с MarketMetrics. Эти проверки — статический guard на текст _SOLD_COUNT_SQL:
+# регресс к COUNT(*) / потеря premise_kind / усечение предиката упадёт здесь.
+# (Полноценный fan-out-тест с двумя mapping-строками на один obj_id требует
+# живой БД — он в tests/integration/test_phantom_columns.py, skip без TEST_DB.)
+
+
+def _sold_sql_text() -> str:
+    from app.services.site_finder.competitors import _SOLD_COUNT_SQL
+
+    return str(_SOLD_COUNT_SQL)
+
+
+def test_sold_count_sql_is_fanout_safe() -> None:
+    """COUNT(DISTINCT objective_lot_id) — маппинг 1:N по obj_id не задваивает.
+
+    objective_complex_mapping UNIQUE = (objective_complex_name, objective_group),
+    НЕ domrf_obj_id → один obj_id может иметь несколько mapping-строк (то же имя,
+    разный objective_group) → JOIN по project_name задвоил бы лоты. DISTINCT по
+    objective_lot_id дедуплицирует на уровне лота. Регресс к COUNT(*) → fail.
+    """
+    sql = _sold_sql_text()
+    assert "COUNT(DISTINCT ol.objective_lot_id)" in sql, (
+        "fan-out guard: маппинг не unique по domrf_obj_id — нужен COUNT(DISTINCT lot)"
+    )
+    assert "COUNT(*)" not in sql, "COUNT(*) задвоит лоты при 1:N маппинге"
+
+
+def test_sold_count_sql_filters_premise_kind() -> None:
+    """premise_kind-фильтр (как в MarketMetrics._STOCK_SQL) — без него проданные
+    паркинги/кладовые/нежилое считались бы проданными квартирами (over-count)."""
+    sql = _sold_sql_text()
+    assert "ol.premise_kind = :premise_kind" in sql
+
+
+def test_sold_count_sql_predicate_matches_market_metrics() -> None:
+    """sold-предикат побайтово совпадает с MarketMetrics: is_sold OR contract_date
+    OR status='продан' (Объектив заполняет поля неконсистентно → берём их ИЛИ)."""
+    sql = _sold_sql_text()
+    assert "ol.is_sold IS TRUE" in sql
+    assert "ol.contract_date IS NOT NULL" in sql
+    assert "LOWER(COALESCE(ol.status, '')) = 'продан'" in sql
diff --git a/backend/tests/integration/test_phantom_columns.py b/backend/tests/integration/test_phantom_columns.py
index ca8074bd..080ab108 100644
--- a/backend/tests/integration/test_phantom_columns.py
+++ b/backend/tests/integration/test_phantom_columns.py
@@ -333,23 +333,29 @@ class TestObjectiveTables:
         )
 
     def test_sold_count_query(self, phantom_check_session: Session) -> None:
-        """competitors._SOLD_COUNT_SQL (#1926) — mapping → objective_lots.contract_date.
+        """competitors._SOLD_COUNT_SQL (#1926, code-review #1927) — mapping → lots.
 
-        flats_sold выровнен с MarketMetrics: COUNT проданных лотов по
-        objective_lots.contract_date (100%-заполнен), мост через
-        objective_complex_mapping (domrf_obj_id → objective_complex_name ==
-        objective_lots.project_name).
+        flats_sold выровнен с MarketMetrics._STOCK_SQL: предикат
+        is_sold/contract_date/status='продан' + premise_kind='квартира'.
+        COUNT(DISTINCT objective_lot_id) — fan-out-safe (mapping НЕ unique по
+        domrf_obj_id). Мост objective_complex_mapping.domrf_obj_id →
+        objective_complex_name == objective_lots.project_name.
         """
         _explain(
             phantom_check_session,
             """
             SELECT
                 cm.domrf_obj_id AS obj_id,
-                COUNT(*) FILTER (WHERE ol.contract_date IS NOT NULL) AS flats_sold
+                COUNT(DISTINCT ol.objective_lot_id) FILTER (
+                    WHERE ol.is_sold IS TRUE
+                       OR ol.contract_date IS NOT NULL
+                       OR LOWER(COALESCE(ol.status, '')) = 'продан'
+                ) AS flats_sold
             FROM objective_complex_mapping cm
             JOIN objective_lots ol
                 ON ol.project_name = cm.objective_complex_name
             WHERE cm.domrf_obj_id = ANY(ARRAY[1, 2, 3])
+              AND ol.premise_kind = 'квартира'
             GROUP BY cm.domrf_obj_id
             """,
         )