backend/app/api/v1/admin_scrape.py

@@ -684,12 +684,18 @@ def objective_coverage(
     """
     from app.services.objective_etl import get_sqlite_info
 
+    # #1964: `lots` = СЫРОЙ COUNT(*) — намеренно. Этот счётчик сравнивается с
+    # SQLite-источником Антона (ETL-fidelity: «свежее ли SQLite, чем БД»), а SQLite
+    # хранит сырые строки → дедуп-view сломал бы сравнение. Рядом отдаём
+    # `lots_physflat` (v_objective_lots_latest — по физлоту) для прозрачности
+    # инфляции (~2.91×: сырых ~1.75M vs физлотов ~603k).
     pg_row = (
         db.execute(
             text(
                 """
                 SELECT
                     (SELECT COUNT(*) FROM objective_lots) AS lots,
+                    (SELECT COUNT(*) FROM v_objective_lots_latest) AS lots_physflat,
                     (SELECT COUNT(*) FROM objective_corpus_room_month) AS crm,
                     (SELECT COUNT(*) FROM objective_complex_mapping) AS mappings,
                     (SELECT MAX(snapshot_date) FROM objective_lots) AS last_lot_snapshot,
@@ -706,6 +712,7 @@ def objective_coverage(
     return {
         "pg": {
             "lots": int(pg_row["lots"] or 0),
+            "lots_physflat": int(pg_row["lots_physflat"] or 0),
             "corp_room_month": int(pg_row["crm"] or 0),
             "mappings": int(pg_row["mappings"] or 0),
             "last_lot_snapshot": pg_row["last_lot_snapshot"].isoformat()

backend/app/api/v1/concepts.py

@@ -72,13 +72,33 @@ _DISTRICT_FOR_POINT_SQL = text(
 
 # Медиана цены продажи жилья из объявлений Objective по району + размер выборки.
 # price_per_m2_rub — РУБЛИ. Санитарный диапазон отсекает мусор. CAST psycopg v3.
+#
+# #1964 physflat-дедуп: objective_lots раздут ~2.91× (мульти objective_lot_id на
+# один физлот через пере-листинги) → sample_size и гейт n≥:_MIN_OBJECTIVE_SAMPLE
+# были по пере-листингам, вес медианы смещался. Дедупим INLINE через DISTINCT ON
+# (physflat-ключ, последний снапшот snapshot_date DESC, id DESC) — НЕ через общий
+# physflat-VIEW. Причина (deep-review #1964, прод-EXPLAIN): view не несёт
+# district-фильтр внутри, qual не проталкивается ниже DISTINCT ON → view сортирует
+# ВСЮ таблицу 1.76M (~5.8 s, request-path внутри analyze_parcel). Inline с district
+# В CTE → bitmap по району ~240k строк + sort (~1.7 s), индекс по district работает.
+# Сегмент-фильтры (price NOT NULL / диапазон) объективны по физлоту → применяем
+# ПОСЛЕ дедупа. Зеркало паттерна market_metrics._STOCK_SQL.
 _OBJECTIVE_MEDIAN_SQL = text(
     """
-    SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_per_m2_rub) AS median_ppm2,
-           COUNT(*)                                                      AS sample_size
+    WITH latest AS (
+        SELECT DISTINCT ON (
+            project_name, corpus_name, section, floor, lot_number
+        )
+            price_per_m2_rub
           FROM objective_lots
          WHERE district = CAST(:dn AS text)
-       AND price_per_m2_rub IS NOT NULL
+         ORDER BY project_name, corpus_name, section, floor, lot_number,
+                  snapshot_date DESC, id DESC
+    )
+    SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_per_m2_rub) AS median_ppm2,
+           COUNT(*)                                                      AS sample_size
+      FROM latest
+     WHERE price_per_m2_rub IS NOT NULL
        AND price_per_m2_rub BETWEEN CAST(:lo AS numeric) AND CAST(:hi AS numeric)
     """
 )

backend/app/api/v1/landing.py

@@ -69,12 +69,14 @@ def _query_stats(db: Session) -> dict[str, Any] | None:
         deals_total = int(row_deals or 0)
 
         # KPI 3: % objective_lots с ценой
+        # #1964: physflat-дедуп view — доля квартир с ценой по ФИЗИЧЕСКИМ лотам, а не
+        # по пере-листингам (сырой objective_lots раздут ~2.91× мульти lot_id/физлот).
         row_price = db.execute(
             text(
                 "SELECT"
                 "  COUNT(*) FILTER (WHERE price_per_m2_rub IS NOT NULL) * 100.0"
                 "    / NULLIF(COUNT(*), 0)"
-                " FROM objective_lots"
+                " FROM v_objective_lots_latest"
             )
         ).scalar()
         price_coverage_pct = round(float(row_price or 0.0), 1)

backend/app/api/v1/parcels.py

@@ -1869,24 +1869,52 @@ def analyze_parcel(
                            3000
                        )
             ),
-            obj_pricing AS (
-                -- Pushdown: агрегируем objective_lots ТОЛЬКО для ближних obj_id.
-                -- AVG/COUNT per domrf_obj_id независимы между группами, поэтому
-                -- ограничение scope даёт байт-идентичные значения для тех obj_id,
-                -- что реально джойнятся ниже (далёкие медианы раньше считались зря).
-                SELECT
-                    cm.domrf_obj_id,
-                    ROUND(AVG(ol.price_per_m2_rub)::numeric, 0)  AS avg_price_per_m2_rub,
-                    ROUND(AVG(ol.area_pd)::numeric, 1)            AS avg_area_pd,
-                    COUNT(*) FILTER (WHERE ol.is_sold)            AS units_sold,
-                    COUNT(*) FILTER (WHERE NOT ol.is_sold)        AS units_available,
-                    COUNT(*) FILTER (
-                        WHERE ol.price_per_m2_rub IS NOT NULL
-                    )                                             AS lots_with_price
+            nearby_projects AS (
+                -- Имена Objective-проектов ближних obj_id — scope для physflat-дедупа.
+                SELECT DISTINCT cm.domrf_obj_id, cm.objective_complex_name
                   FROM objective_complex_mapping cm
-                  JOIN objective_lots ol ON ol.project_name = cm.objective_complex_name
                  WHERE cm.domrf_obj_id IN (SELECT obj_id FROM nearby_obj)
-                 GROUP BY cm.domrf_obj_id
+            ),
+            obj_lots_latest AS (
+                -- #1964 physflat-дедуп: objective_lots раздут ~2.91× (мульти lot_id
+                -- на физлот) → units_sold/units_available карточки конкурента были
+                -- завышены. Дедупим INLINE (DISTINCT ON physflat-ключ, последний
+                -- снапшот), scope протолкнут через project_name ближних ЖК. НЕ через
+                -- общий physflat-VIEW: он материализует ВСЮ таблицу (join не проходит
+                -- ниже DISTINCT ON) → seq-scan+sort 1.76M (~5.7 s на request-path
+                -- analyze_parcel); inline scoped → ~доли секунды (прод-EXPLAIN #1964).
+                SELECT DISTINCT ON (
+                    ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number
+                )
+                    ol.project_name,
+                    ol.price_per_m2_rub,
+                    ol.area_pd,
+                    ol.is_sold
+                  FROM objective_lots ol
+                 WHERE ol.project_name IN (
+                           SELECT objective_complex_name FROM nearby_projects
+                       )
+                 ORDER BY ol.project_name, ol.corpus_name, ol.section, ol.floor,
+                          ol.lot_number, ol.snapshot_date DESC, ol.id DESC
+            ),
+            obj_pricing AS (
+                -- Pushdown: агрегируем ТОЛЬКО для ближних obj_id. AVG/COUNT per
+                -- domrf_obj_id независимы между группами → scope даёт идентичные
+                -- значения для джойнящихся obj_id. Маппинг прод 1:1 по domrf_obj_id
+                -- (303 строки = 303 distinct) → COUNT(*) по дедуп-физлотам корректен.
+                SELECT
+                    np.domrf_obj_id,
+                    ROUND(AVG(oll.price_per_m2_rub)::numeric, 0) AS avg_price_per_m2_rub,
+                    ROUND(AVG(oll.area_pd)::numeric, 1)          AS avg_area_pd,
+                    COUNT(*) FILTER (WHERE oll.is_sold)          AS units_sold,
+                    COUNT(*) FILTER (WHERE NOT oll.is_sold)      AS units_available,
+                    COUNT(*) FILTER (
+                        WHERE oll.price_per_m2_rub IS NOT NULL
+                    )                                            AS lots_with_price
+                  FROM nearby_projects np
+                  JOIN obj_lots_latest oll
+                    ON oll.project_name = np.objective_complex_name
+                 GROUP BY np.domrf_obj_id
             )
             SELECT o.obj_id,
                    o.comm_name,
@@ -2535,6 +2563,24 @@ def analyze_parcel(
                 dp_row = (
                     db.execute(
                         text("""
+                        -- #1964 physflat-дедуп: objective_lots раздут ~2.91× (мульти
+                        -- lot_id на физлот) → sample_size/вес медианы были по
+                        -- пере-листингам. Дедупим INLINE через DISTINCT ON
+                        -- (physflat-ключ, последний снапшот), district протолкнут В
+                        -- CTE. НЕ через v_objective_lots_latest: view не несёт
+                        -- district внутри → qual не проходит ниже DISTINCT ON →
+                        -- сортировка ВСЕЙ таблицы 1.76M (~5.8 s на request-path
+                        -- analyze_parcel). Inline → bitmap по району + sort (~1.7 s).
+                        WITH latest AS (
+                            SELECT DISTINCT ON (
+                                project_name, corpus_name, section, floor, lot_number
+                            )
+                                price_per_m2_rub
+                              FROM objective_lots
+                             WHERE district = CAST(:dn AS text)
+                             ORDER BY project_name, corpus_name, section, floor,
+                                      lot_number, snapshot_date DESC, id DESC
+                        )
                         SELECT
                             MIN(price_per_m2_rub)                     AS price_min,
                             MAX(price_per_m2_rub)                     AS price_max,
@@ -2542,9 +2588,8 @@ def analyze_parcel(
                                 ORDER BY price_per_m2_rub
                             )                                          AS price_median,
                             COUNT(*)                                   AS sample_size
-                          FROM objective_lots
-                         WHERE district = CAST(:dn AS text)
-                           AND price_per_m2_rub IS NOT NULL
+                          FROM latest
+                         WHERE price_per_m2_rub IS NOT NULL
                            AND price_per_m2_rub BETWEEN 30000 AND 600000
                     """),
                         {"dn": district_row["district_name"]},
@@ -2903,28 +2948,58 @@ def analyze_parcel(
                 grp_row = (
                     db.execute(
                         text("""
-                            SELECT
-                                percentile_cont(0.5) WITHIN GROUP (ORDER BY ol.price_per_m2_rub)
-                                    AS median,
-                                count(*) AS n,
-                                count(DISTINCT c.id) AS n_complexes
+                            -- #1964 physflat-дедуп: objective_lots раздут ~2.91×
+                            -- (мульти lot_id на физлот) → n/вес медианы/гейт n≥10
+                            -- были по пере-листингам. Дедупим INLINE через DISTINCT ON
+                            -- (physflat-ключ, последний снапшот), scope протолкнут В
+                            -- CTE через complex_id ближних ЖК. НЕ через
+                            -- v_objective_lots_latest: view материализует ВСЮ таблицу
+                            -- (qual/join не проходят ниже DISTINCT ON) → seq-scan+sort
+                            -- 1.76M (~6.4 s на request-path analyze_parcel). Inline:
+                            -- geo-index по complexes → Nested Loop bitmap
+                            -- objective_lots_complex_idx по ~186 ЖК → ~120 ms
+                            -- (прод-EXPLAIN deep-review #1964). Дедуп до price-фильтра:
+                            -- цена объективна по физлоту (последний снапшот).
+                            WITH nearby_cx AS (
+                                SELECT c.id
                                   FROM complexes c
-                            JOIN objective_lots ol
-                                ON ol.complex_id = c.id
-                               AND ol.price_per_m2_rub IS NOT NULL
                                  WHERE c.latitude IS NOT NULL
                                    AND c.longitude IS NOT NULL
                                    AND ST_DWithin(
                                            ST_SetSRID(
-                                          ST_MakePoint(c.longitude, c.latitude), 4326
+                                               ST_MakePoint(c.longitude, c.latitude),
+                                               4326
                                            )::geography,
                                            ST_SetSRID(
                                                ST_MakePoint(
-                                              CAST(:lon AS float), CAST(:lat AS float)
+                                                   CAST(:lon AS float),
+                                                   CAST(:lat AS float)
                                                ), 4326
                                            )::geography,
                                            CAST(:radius_m AS float)
                                        )
+                            ),
+                            latest AS (
+                                SELECT DISTINCT ON (
+                                    ol.project_name, ol.corpus_name, ol.section,
+                                    ol.floor, ol.lot_number
+                                )
+                                    ol.price_per_m2_rub,
+                                    ol.complex_id
+                                  FROM objective_lots ol
+                                 WHERE ol.complex_id IN (SELECT id FROM nearby_cx)
+                                 ORDER BY ol.project_name, ol.corpus_name, ol.section,
+                                          ol.floor, ol.lot_number,
+                                          ol.snapshot_date DESC, ol.id DESC
+                            )
+                            SELECT
+                                percentile_cont(0.5) WITHIN GROUP (
+                                    ORDER BY price_per_m2_rub
+                                )                                       AS median,
+                                count(*)                                AS n,
+                                count(DISTINCT complex_id)              AS n_complexes
+                            FROM latest
+                            WHERE price_per_m2_rub IS NOT NULL
                         """),
                         {
                             "lon": centroid_lon,

backend/app/services/forecasting/report_assembler.py

@@ -152,17 +152,31 @@ def _as_dict_list(items: Sequence[Any] | None) -> list[dict[str, Any]]:
 
 
 def _deal_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
-    """Число сделок (продаж) за окно — для deal_count #990. PURE.
+    """Число сделок (продаж) ЗА ОКНО — для deal_count #990. PURE.
 
-    Предпочитаем `market_metrics.n_sold` (§9.2 — прямой счётчик проданных лотов выборки);
-    fallback на `analyze.market_pulse.competitors_total` НЕ берём (это число ЖК, не
-    сделок — пошло бы в analog_count). Нет сигнала → None (#990 → тянет в low).
+    #1964: НЕ берём `market_metrics.n_sold` — это КУМУЛЯТИВНЫЙ all-time проданный сток
+    (десятки-сотни тысяч), а #990/confidence_engine помечает фактор «за {window} мес»
+    (6) и гейтит порогами 6-мес окна (high≥50). n_sold ≫ 50 ВСЕГДА → гейт бессмыслен и
+    подпись врёт. Берём оконное число = unit_velocity × window_months, где unit_velocity
+    (§9.2) = units_sold_window / window_months из _SALES_WINDOW_SQL (он
+    contract_date-windowed) → произведение восстанавливает units_sold за окно. Так
+    подпись «за 6 мес» честная и порог 50/15 осмысленный (прод EKB: ~24 876 за 6 мес
+    vs ~380 921 кумулятивно).
+
+    unit_velocity отсутствует (None / нет market_metrics) → None (#990 трактует как
+    «нет данных», тянет в low). window_months ≤ 0 / нет → None (нечем масштабировать).
+    Fallback на `analyze.market_pulse.competitors_total` НЕ берём (это число ЖК, не
+    сделок — пошло бы в analog_count).
     """
-    if market_metrics is not None:
-        n_sold = market_metrics.get("n_sold")
-        if isinstance(n_sold, int):
-            return n_sold
+    if market_metrics is None:
         return None
+    unit_velocity = market_metrics.get("unit_velocity")
+    window_months = market_metrics.get("window_months")
+    if not isinstance(unit_velocity, int | float):
+        return None
+    if not isinstance(window_months, int) or window_months <= 0:
+        return None
+    return round(float(unit_velocity) * window_months)
 
 
 def _analog_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
@@ -239,7 +253,8 @@ def _deal_count_months(market_metrics: dict[str, Any] | None) -> int | None:
     """Окно наблюдения для deal_count (мес) — для deal_count_months #990. PURE.
 
     Читает тот же `market_metrics.window_months` (§9.2), что и `_history_months` —
-    именно за это окно считается n_sold. Нет → None (#990 пропускает суффикс «за N мес»).
+    #1964: именно за это окно `_deal_count` считает сделки (unit_velocity ×
+    window_months). Нет → None (#990 пропускает суффикс «за N мес»).
     """
     if market_metrics is not None:
         window = market_metrics.get("window_months")

backend/app/services/forecasting/special_indices.py

@@ -828,13 +828,16 @@ def _unavailable(key: str, *, reason: str) -> SpecialIndex:
 # `market_metrics._SALES_WINDOW_SQL`: bind `:has_district` (boolean) переключает
 # ветку, `:districts` (text[]) — список микро. «Продан» = is_sold IS TRUE ИЛИ
 # contract_date заполнена. psycopg v3: CAST, НИКОГДА :x::type.
+# #1964: читаем из v_objective_lots_latest (physflat-дедуп view) — сырой
+# objective_lots раздут ~2.91× (мульти lot_id на физлот); доля ипотеки —
+# отношение, к дублям устойчива, но n_sold/n_mortgage теперь честные (по физлотам).
 _ARTIFICIAL_DEMAND_SQL = text(
     """
     WITH sold AS (
         SELECT
             ol.objective_lot_id,
             (ol.encumbrance_type IS NOT NULL OR ol.bank_name IS NOT NULL) AS has_mortgage
-          FROM objective_lots ol
+          FROM v_objective_lots_latest ol
          WHERE ol.premise_kind = CAST(:premise_kind AS text)
            AND (ol.is_sold IS TRUE OR ol.contract_date IS NOT NULL)
            AND (

backend/app/services/site_finder/competitors.py

@@ -491,10 +491,18 @@ _AVG_PRICE_SQL = text("""
 # COUNT(DISTINCT ol.objective_lot_id) — дедуп на уровне лота, fan-out-safe.
 # (_OBJECTIVE_PRICE_FALLBACK не прецедент: percentile/median dup-нечувствительны.)
 #
+# #1964: objective_lots — current-state UPSERT, но НЕ «ровно одна строка на лот».
+# UNIQUE = objective_lot_id, а Объектив за пере-листинги присваивает ОДНОМУ
+# физлоту (project,corpus,section,floor,lot) НЕСКОЛЬКО objective_lot_id → таблица
+# раздута ~2.91× → flats_sold завышался. Читаем из v_objective_lots_latest
+# (physflat-дедуп view, по строке на физлот). В view objective_lot_id уникален
+# (603k строк = 603k distinct lot_id), поэтому COUNT(DISTINCT ol.objective_lot_id)
+# == COUNT(*) по физлотам — НО оставляем DISTINCT: маппинг 1:N по domrf_obj_id
+# (см. выше) при будущем fan-out задвоил бы строки view, COUNT(*) был бы не
+# fan-out-safe. DISTINCT даёт И physflat-дедуп (через view), И fan-out-защиту.
+#
 # 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="квартира") — оба считают продажи жилых лотов.
@@ -508,7 +516,7 @@ _SOLD_COUNT_SQL = text("""
                OR LOWER(COALESCE(ol.status, '')) = 'продан'
         ) AS flats_sold
     FROM objective_complex_mapping cm
-    JOIN objective_lots ol
+    JOIN v_objective_lots_latest ol
         ON ol.project_name = cm.objective_complex_name
     WHERE cm.domrf_obj_id = ANY(:obj_ids)
       AND ol.premise_kind = :premise_kind

backend/app/services/site_finder/market_metrics.py

@@ -290,8 +290,11 @@ def _room_bucket(rooms_int: int | None) -> str:
 # только к форсайт-пути. Deep-review (#1959) подтвердил: безопасно для остальных
 # потребителей (absorption/MoS/sell_through — отношения; velocity — saturated),
 # т.к. inflation симметрична в числителе/знаменателе и съедается шкалой. Дедуп
-# здесь касается ТОЛЬКО выборки внутри метрик — дедуп самой сырой таблицы
-# objective_lots (для всех platform-консьюмеров) остаётся отдельной задачей #1964.
+# здесь касается ТОЛЬКО выборки внутри метрик; #1964 ввёл общий view
+# v_objective_lots_latest для ОСТАЛЬНЫХ current-state консьюмеров — он использует
+# ТОТ ЖЕ physflat-ключ/направления (project,corpus,section,floor,lot_number;
+# snapshot_date DESC, id DESC), что и этот inline DISTINCT ON (инвариант общий).
+# Этот блок НЕ рефакторим на view (zero behavior change, лишний риск).
 #
 # Сегмент-фильтры (#1959): obj_class (class в objective_lots — lowercase, матчим
 # LOWER=LOWER) + room_bucket. ROOM-bucket — Source-B room_area-вокабуляр

backend/app/services/site_finder/supply_layers.py

@@ -184,6 +184,12 @@ def _l2_confidence(n_objects_total: int, n_objects_with_free_flats: int) -> Conf
 # Доступно = выставлено и НЕ продано. is_sold распознаём И флагом, И статусом 'продан'
 # / наличием contract_date (Объектив заполняет неконсистентно — зеркало _STOCK_SQL).
 # Группируем по (district, class). class — lowercase в lots (м.68 коммент).
+#
+# #1964: читаем из v_objective_lots_latest (physflat-дедуп view), НЕ из сырого
+# objective_lots. Сырая таблица раздута ~2.91× (мульти objective_lot_id на один
+# физлот через пере-листинги) → units_available по району был завышен ~3-4.6×
+# (прод: Юго-Западный комфорт 58 606 → 12 620) → дефицит-форсайт переоценивал
+# открытое предложение L1. View отдаёт по строке на физический лот.
 _L1_OPEN_SQL = text(
     """
     WITH lots AS (
@@ -196,7 +202,7 @@ _L1_OPEN_SQL = text(
                 OR ol.contract_date IS NOT NULL
                 OR LOWER(COALESCE(ol.status, '')) = 'продан'
             ) AS sold_now
-          FROM objective_lots ol
+          FROM v_objective_lots_latest ol
          WHERE ol.premise_kind = :premise_kind
            AND (
                 CAST(:has_district AS boolean) IS FALSE

backend/tests/api/v1/test_analyze_competitors_status.py

@@ -174,12 +174,36 @@ class TestObjPricingPushdown:
             "совпадал с выводимыми конкурентами"
         )
 
-    def test_aggregation_unchanged(self) -> None:
-        """Агрегатные выражения (AVG/COUNT FILTER) не тронуты — менялся только scope."""
+    def test_aggregation_per_domrf_obj_id(self) -> None:
+        """Агрегаты (AVG/COUNT FILTER) считаются per domrf_obj_id (логика цен #1332).
+
+        #1964: источник агрегатов сменился с сырого objective_lots (alias ol) на
+        physflat-дедуп CTE obj_lots_latest (alias oll) — см. test_obj_pricing_*_physflat
+        ниже. Сами агрегатные выражения и группировка per-obj_id неизменны.
+        """
         sql = self._competitor_sql()
-        # Логика цен/fallback (#1332) неизменна: те же агрегаты per domrf_obj_id.
-        assert "ROUND(AVG(ol.price_per_m2_rub)::numeric, 0) AS avg_price_per_m2_rub" in sql
-        assert "COUNT(*) FILTER (WHERE ol.price_per_m2_rub IS NOT NULL )" in sql.replace(
+        assert "ROUND(AVG(oll.price_per_m2_rub)::numeric, 0) AS avg_price_per_m2_rub" in sql
+        assert "lots_with_price" in sql
+        assert "COUNT(*) FILTER (WHERE oll.is_sold) AS units_sold" in sql
+        assert "COUNT(*) FILTER (WHERE NOT oll.is_sold) AS units_available" in sql
+        assert "GROUP BY np.domrf_obj_id" in sql
+
+    def test_obj_pricing_dedups_physflat_inline(self) -> None:
+        """#1964: obj_pricing агрегирует physflat-дедуп набор (DISTINCT ON), НЕ сырой.
+
+        objective_lots раздут ~2.91× (мульти objective_lot_id на физлот через
+        пере-листинги) → units_sold/units_available карточки конкурента были завышены.
+        Дедупим INLINE (DISTINCT ON physflat-ключ, последний снапшот) со scope по
+        project_name ближних ЖК. НЕ через v_objective_lots_latest: view материализует
+        ВСЮ таблицу 1.76M (join не проходит ниже DISTINCT ON → seq-scan+sort ~5.7 s на
+        request-path analyze_parcel; inline scoped → ~0.4 s, прод-EXPLAIN #1964).
+        """
+        sql = self._competitor_sql()
+        assert "obj_lots_latest AS (" in sql, "должен быть physflat-дедуп CTE (#1964)"
+        assert "DISTINCT ON ( ol.project_name, ol.corpus_name, ol.section," in sql.replace(
             "\n", " "
-        ) or "lots_with_price" in sql
-        assert "GROUP BY cm.domrf_obj_id" in sql
+        ), "obj_lots_latest должен дедупить по physflat-ключу"
+        assert "snapshot_date DESC, ol.id DESC" in sql, "берём последний снапшот физлота"
+        assert (
+            "v_objective_lots_latest" not in sql
+        ), "request-path: view материализует всю таблицу — нужен inline DISTINCT ON (#1964)"

backend/tests/api/v1/test_parcel_competitors.py

@@ -449,9 +449,9 @@ def test_competitors_avg_price_populated() -> None:
         )
         assert resp.status_code == 200, resp.text
         comp = resp.json()["competitors"][0]
-        assert comp["avg_price_per_m2"] == pytest.approx(150_000.0), (
-            "avg_price_per_m2 должен быть не None — регрессия #227 status='sold' filter"
-        )
+        assert comp["avg_price_per_m2"] == pytest.approx(
+            150_000.0
+        ), "avg_price_per_m2 должен быть не None — регрессия #227 status='sold' filter"
         # OBJ-3 #307: domrf-hit → price_source='domrf'.
         assert comp["price_source"] == "domrf"
     finally:
@@ -471,9 +471,7 @@ def test_competitors_objective_price_fallback_fills_gap() -> None:
     rows = [_obj_row(obj_id=1)]
     # domrf price пуст → fallback на objective median.
     obj_price_rows = [_obj_price_row(obj_id=1, price=136_000.0)]
-    db = _make_db(
-        coord=_coord_row(), obj_rows=rows, price_rows=[], obj_price_rows=obj_price_rows
-    )
+    db = _make_db(coord=_coord_row(), obj_rows=rows, price_rows=[], obj_price_rows=obj_price_rows)
 
     from app.core.db import get_db
 
@@ -597,14 +595,30 @@ def test_sold_count_sql_is_fanout_safe() -> None:
     НЕ domrf_obj_id → один obj_id может иметь несколько mapping-строк (то же имя,
     разный objective_group) → JOIN по project_name задвоил бы лоты. DISTINCT по
     objective_lot_id дедуплицирует на уровне лота. Регресс к COUNT(*) → fail.
+
+    #1964: даже после перехода на v_objective_lots_latest (где objective_lot_id
+    уникален → COUNT(DISTINCT)==COUNT(*) по физлотам) DISTINCT СОХРАНЯЕТСЯ: будущий
+    fan-out маппинга задвоил бы строки view, и COUNT(*) перестал бы быть fan-out-safe.
     """
     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(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_reads_physflat_view() -> None:
+    """#1964: sold-count читает physflat-дедуп view, НЕ сырой objective_lots.
+
+    Сырой objective_lots раздут ~2.91× (мульти objective_lot_id на один физлот через
+    пере-листинги) → flats_sold завышался. Репойнт на v_objective_lots_latest даёт по
+    строке на физический лот. Регресс к `FROM objective_lots` (без _latest) → fail.
+    """
+    sql = _sold_sql_text()
+    assert "v_objective_lots_latest" in sql, "должен читать physflat-дедуп view (#1964)"
+    assert "JOIN objective_lots " not in sql, "не должен джойнить сырой objective_lots"
+
+
 def test_sold_count_sql_filters_premise_kind() -> None:
     """premise_kind-фильтр (как в MarketMetrics._STOCK_SQL) — без него проданные
     паркинги/кладовые/нежилое считались бы проданными квартирами (over-count)."""

backend/tests/services/forecasting/test_report_assembler.py

@@ -752,8 +752,18 @@ class TestSignalExtractionHelpers:
         assert _as_dict_or(_FakeMetrics(1.0, 2, 3))["obj_count"] == 2
         assert _as_dict_or(42) is None  # мусор → None (graceful)
 
-    def test_deal_count_from_market_metrics(self) -> None:
-        assert _deal_count({}, {"n_sold": 88}) == 88
+    def test_deal_count_is_windowed_not_cumulative(self) -> None:
+        # #1964: deal_count = unit_velocity × window_months (продажи ЗА ОКНО), НЕ
+        # кумулятивный n_sold. confidence_engine помечает фактор «за N мес» и гейтит
+        # порогами окна — кумулятивный n_sold (≫50) сделал бы гейт бессмысленным и
+        # подпись лживой.
+        assert _deal_count({}, {"unit_velocity": 8.2, "window_months": 18}) == round(8.2 * 18)
+        # n_sold ИГНОРИРУЕТСЯ (кумулятив, не оконный) — даже когда присутствует.
+        assert _deal_count({}, {"n_sold": 99000, "unit_velocity": 5.0, "window_months": 6}) == 30
+        # нет unit_velocity / window / market_metrics → None (нет оконного сигнала)
+        assert _deal_count({}, {"n_sold": 88}) is None
+        assert _deal_count({}, {"unit_velocity": 8.2}) is None
+        assert _deal_count({}, {"unit_velocity": 8.2, "window_months": 0}) is None
         assert _deal_count({}, None) is None
 
     def test_analog_count_prefers_obj_count(self) -> None:

backend/tests/services/generative/test_market_price_lookup.py

@@ -48,6 +48,24 @@ class _FakeSession:
 _WKT = "POINT (60.6 56.83)"
 
 
+def test_objective_median_sql_dedups_inline_not_via_view() -> None:
+    """#1964: _OBJECTIVE_MEDIAN_SQL дедупит INLINE (DISTINCT ON physflat-ключ), НЕ
+    через v_objective_lots_latest.
+
+    Это REQUEST-PATH запрос (внутри analyze/концепт-генерации). View не несёт
+    district-фильтр внутри → qual не проталкивается ниже DISTINCT ON → view сортирует
+    ВСЮ таблицу 1.76M (~5.8 s, прод-EXPLAIN deep-review #1964). Inline DISTINCT ON с
+    district В CTE → bitmap по району + sort (~1.7 s). Регресс к
+    `FROM v_objective_lots_latest` (whole-table materialize) → fail.
+    """
+    sql = str(concepts._OBJECTIVE_MEDIAN_SQL)
+    assert "DISTINCT ON (" in sql, "должен дедупить физлоты inline"
+    assert "snapshot_date DESC" in sql, "берём последний снапшот физлота"
+    assert (
+        "v_objective_lots_latest" not in sql
+    ), "request-path: view материализует всю таблицу — нужен inline DISTINCT ON (#1964)"
+
+
 def test_objective_median_selected_when_sample_large_enough() -> None:
     # Район найден + Objective n>=10 → выбирается медиана Objective.
     db = _FakeSession(

backend/tests/services/site_finder/test_supply_layers.py

@@ -229,6 +229,17 @@ class TestLayer1Open:
         assert "CAST(:districts AS text[])" in sql
         _assert_no_double_colon_cast(sql)
 
+    def test_sql_reads_physflat_view(self) -> None:
+        # #1964: L1 открытое предложение читает physflat-дедуп view, НЕ сырой
+        # objective_lots (раздут ~2.91× мульти lot_id/физлот → units_available был
+        # завышен ~3-4.6× → дефицит-форсайт переоценивал L1). Регресс к
+        # `FROM objective_lots` → fail.
+        db = _mock_db([])
+        compute_layer1_open(db, district="Автовокзал")
+        sql = _executed_sql(db)
+        assert "v_objective_lots_latest" in sql, "L1 должен читать physflat-дедуп view"
+        assert "FROM objective_lots " not in sql, "L1 не должен читать сырой objective_lots"
+
     def test_params(self) -> None:
         db = _mock_db([])
         compute_layer1_open(db, district="Автовокзал", premise_kind="квартира")

data/sql/175_v_objective_lots_latest.sql

@@ -0,0 +1,67 @@
+-- Контекст: #1964 (эпик #1953) — physflat-дедуп current-state потребителей
+-- objective_lots через общий VIEW.
+--
+-- objective_lots — current-state UPSERT (UNIQUE objective_lot_id, ON CONFLICT DO
+-- UPDATE), НЕ snapshot-append: всего 5 distinct snapshot_date. Инфляция строк
+-- (~2.91×: прод 2026-06-28 — 1 753 283 сырых квартир-строки vs 603 049 физлотов)
+-- возникает потому, что Объектив за время пере-листингов присваивает ОДНОМУ
+-- физическому лоту (project_name, corpus_name, section, floor, lot_number)
+-- НЕСКОЛЬКО objective_lot_id. История лотов живёт в ОТДЕЛЬНОЙ objective_lots_history
+-- (её НЕ трогаем — это корректный time-series).
+--
+-- View отдаёт ТОТ ЖЕ набор колонок, что и objective_lots (ol.* — прод-проверка:
+-- 51 base cols == 51 view cols), по ОДНОЙ строке на физлот — последний снапшот
+-- (snapshot_date DESC, tie-break id DESC). objective_lot_id в результате уникален
+-- (603 049 строк = 603 049 distinct objective_lot_id, 0 NULL) → потребители,
+-- которым нужна fan-out-защита по маппингу (competitors._SOLD_COUNT_SQL), могут
+-- использовать COUNT(DISTINCT objective_lot_id) поверх view.
+--
+-- ⚠ ПЛАН/ИНДЕКС (deep-review #1964, прод-EXPLAIN 2026-06-28): partial-индекс
+-- objective_lots_physflat_latest_idx (миграция 173, WHERE premise_kind='квартира')
+-- НЕ обслуживает этот view. Внутри view НЕТ premise-фильтра, а внешний
+-- WHERE premise_kind/district потребителя НЕ проталкивается НИЖЕ DISTINCT ON →
+-- индекс-173 не применим. View материализует ВСЮ таблицу: Parallel Seq Scan +
+-- external Sort 1.76M строк → ~5.8 s (count по одному району).
+-- ПОЛНЫЙ индекс (project,corpus,section,floor,lot_number,snapshot_date DESC,id DESC)
+-- НЕ помогает и НЕ добавлен: DISTINCT ON отдаёт ol.* (51 кол., width≈945) →
+-- index-only-unique невозможен; 605k heap-проб через 142 MB индекс ДОРОЖЕ
+-- seq-scan+sort (планировщик его игнорирует: forced index-path всё равно ~3.9 s).
+-- 142 MB write-amp ради нуля на request-path + замедление bulk-INSERT objective-
+-- scrape — не берём. Подтверждает исходное решение #1964 «новый индекс НЕ добавляем».
+--
+-- Следствие: view применяем ТОЛЬКО к BATCH/широким консьюмерам, где полная
+-- материализация амортизирована или кэширована (supply_layers L1 группирует ВСЕ
+-- районы; competitors._SOLD_COUNT_SQL — по mapping; special_indices; admin; landing).
+-- REQUEST-PATH консьюмеры внутри analyze_parcel с фильтром по ОДНОМУ району
+-- (concepts._OBJECTIVE_MEDIAN_SQL, parcels district-price/geo-median) view НЕ
+-- используют — у них inline DISTINCT ON с district, протолкнутым В CTE (bitmap по
+-- району ~240k строк + sort ~19 MB → ~1.7 s vs 5.8 s через view). Дедуп платится в
+-- любом случае (старый 303 ms был быстрым потому что считал РАЗДУТЫЕ сырые строки).
+--
+-- #1959 уже устранил инфляцию ВНУТРИ compute_market_metrics (inline DISTINCT ON в
+-- _STOCK_SQL/_SALES_WINDOW_SQL) и его 9 forecast/scoring-вызовов — НЕ дублируем.
+--
+-- НЕ трогаем (time-series, корректны): sales_series._SOURCE_B_SQL,
+-- objective_lots_history, mv_sales_tracker_*.
+--
+-- Порядок: миграция ПЕРВОЙ (deploy first), затем backend-код, читающий view.
+-- Idempotent: CREATE OR REPLACE VIEW. Self-wrapped BEGIN/COMMIT.
+-- Прод DRY-RUN (BEGIN; CREATE OR REPLACE VIEW; SELECT count; ROLLBACK):
+-- view kvartira = 603 049 (vs 1 753 283 raw), колонки 51==51 ✓.
+BEGIN;
+
+CREATE OR REPLACE VIEW v_objective_lots_latest AS
+SELECT DISTINCT ON (project_name, corpus_name, section, floor, lot_number) ol.*
+  FROM objective_lots ol
+ ORDER BY project_name, corpus_name, section, floor, lot_number,
+          snapshot_date DESC, id DESC;
+
+COMMENT ON VIEW v_objective_lots_latest IS
+    '#1964: physflat-deduped current state of objective_lots — one row per '
+    '(project_name, corpus_name, section, floor, lot_number) at the latest '
+    'snapshot (snapshot_date DESC, id DESC). objective_lots stores ~2.91x '
+    'inflated rows (multiple objective_lot_id per physical flat across '
+    're-listings); this view collapses to ~603k physical flats. History lives '
+    'in objective_lots_history (do not use this view for time-series).';
+
+COMMIT;