perf(objective_lots): inline DISTINCT ON for 3 view-fullscan consumers + cache landing stats (#1953)

#1964 repointed landing/supply_layers/special_indices to v_objective_lots_latest,
whose DISTINCT ON is an optimization barrier — outer WHERE can't push below it, so
the planner full Parallel-Seq-Scans 1.76M rows + external merge sorts (~6-8s) per call.

- landing._query_stats: add 1h in-process TTL cache (homepage /landing/stats was
  uncached; KPI3 full-view-aggregate measured 6.7s/load). Threadpool-aware lock,
  DB query outside lock, only successful result cached. Mirrors weather_cache.
- supply_layers._L1_OPEN_SQL: view->inline DISTINCT ON, stable filters
  (premise_kind/district) pushed below dedup -> Bitmap on district_class_idx.
  district-scoped 5.9s->0.45s (~13x).
- special_indices._ARTIFICIAL_DEMAND_SQL: view->inline 2-layer CTE; stable filters
  before dedup, changing filters (is_sold/contract_date/class) AFTER dedup.
  class has 248 transient physflats -> before-dedup drifted +193 on standard.
  district-scoped 6.7s->0.57s (~12x), EKB-wide 8.1s->5.5s.

Result bit-identical to view, verified on prod across all classes/districts.
Uses existing covering / district_class indexes (no migration).

backend/app/api/v1/landing.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import threading
+import time
 from datetime import date
 from typing import Annotated, Any
 
@@ -18,6 +20,32 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+# ---------------------------------------------------------------------------
+# In-process TTL-кэш результата _query_stats (перф-батч #2, #1953-followup)
+# ---------------------------------------------------------------------------
+# /landing/stats бьётся на КАЖДЫЙ load хоумпейджа и НЕ кэширован. KPI 3 (доля квартир
+# с ценой) делает полный агрегат по physflat-дедупу objective_lots: price_per_m2_rub
+# нет в covering-INCLUDE и фильтра нет → ~6.7 s heap-скан 603k физлотов на каждый
+# запрос. Данные дневной гранулярности (snapshot_date) → 1 ч staleness приемлем.
+#
+# Кэшируем ВЕСЬ dict (все 5 KPI), не только KPI 3 — там ещё pg_class deals + 2
+# MAX(snapshot_date). Только УСПЕШНЫЙ результат (None/fallback НЕ кэшируем — пустая
+# dev-БД / сбой запроса не должны залипать на час). Потокобезопасно (threading.Lock,
+# удерживается только на check/store); analyze идёт в threadpool под Uvicorn. Стиль —
+# зеркало app.services.weather_cache. time.monotonic устойчив к скачкам системных часов.
+_STATS_TTL_S: float = 3600.0  # 1 час — дневная гранулярность данных
+
+# (result_dict, expires_at_monotonic_seconds). Только успешный результат.
+_STATS_CACHE: dict[str, Any] | None = None
+_STATS_EXPIRES_AT: float = 0.0
+_STATS_LOCK = threading.Lock()
+
+
+def _now() -> float:
+    """Монотонный таймер (тесты подменяют этот helper, а не сам `time`)."""
+    return time.monotonic()
+
+
 # ---------------------------------------------------------------------------
 # Response schema
 # ---------------------------------------------------------------------------
@@ -129,6 +157,30 @@ def _query_stats(db: Session) -> dict[str, Any] | None:
         return None
 
 
+def _query_stats_cached(db: Session) -> dict[str, Any] | None:
+    """`_query_stats` с in-process TTL-кэшем (TTL=1 ч). См. блок-комментарий вверху.
+
+    Кэширует ТОЛЬКО успешный dict; None (сбой/пустая БД) не кэшируется — следующий
+    запрос попробует снова. Lock удерживается лишь на check/store; сам DB-запрос идёт
+    ВНЕ lock'а (не сериализуем одновременные cold-start, last-write wins — запрос
+    идемпотентен, TTL длинный).
+    """
+    global _STATS_CACHE, _STATS_EXPIRES_AT
+    now = _now()
+    with _STATS_LOCK:
+        if _STATS_CACHE is not None and _STATS_EXPIRES_AT > now:
+            return _STATS_CACHE
+    # MISS / истёк → DB-запрос ВНЕ lock'а.
+    result = _query_stats(db)
+    if result is None:
+        # Не кэшируем fallback — пусть следующий запрос повторит попытку.
+        return None
+    with _STATS_LOCK:
+        _STATS_CACHE = result
+        _STATS_EXPIRES_AT = _now() + _STATS_TTL_S
+    return result
+
+
 @router.get("/landing/stats", response_model=LandingStatsOut)
 def landing_stats(
     db: Annotated[Session, Depends(get_db)],
@@ -144,7 +196,7 @@ def landing_stats(
     - paradox: строка-парадокс портфеля для hero CTA
     - stale: true если данные недоступны (DB ошибка или пустая БД)
     """
-    result = _query_stats(db)
+    result = _query_stats_cached(db)
     if result is None:
         return LandingStatsOut(**_FALLBACK_DATA)
     return LandingStatsOut(**result)

backend/app/services/forecasting/special_indices.py

@@ -828,25 +828,62 @@ 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 теперь честные (по физлотам).
+# #1964: считаем по ФИЗИЧЕСКИМ лотам (дедуп) — сырой objective_lots раздут ~2.91×
+# (мульти lot_id на физлот); доля ипотеки — отношение, к дублям устойчива, но
+# n_sold/n_mortgage теперь честные (по физлотам).
+#
+# Перф-батч #2 (#1953-followup): дедупим INLINE через DISTINCT ON (physflat-ключ,
+# последний снапшот), НЕ через v_objective_lots_latest. Причина: view (DISTINCT ON ol.*)
+# — оптимизационный барьер: WHERE снаружи НЕ проталкивается ниже DISTINCT ON →
+# планировщик материализует ВСЮ таблицу (1.76M, Parallel Seq Scan + external merge sort
+# 66-81 МБ/воркер) ПЕРЕД фильтром (~5.4-6.4 s district-scoped, прод-EXPLAIN). Inline:
+# bitmap по objective_lots_district_class_idx + узкий sort (~3-5 МБ/воркер) ~0.37 s
+# (≈15×).
+#
+# ⚠️ КОРРЕКТНОСТЬ дедупа (слои фильтров РАЗНЫЕ — не путать):
+#   • premise_kind — ПОЛНОСТЬЮ стабилен (0 mixed-групп на проде) → в WHERE ДО DISTINCT ON.
+#   • district — почти стабилен (4 transient-группы на 605k физлотов). Протолкнут ДО
+#     DISTINCT ON ОСОЗНАННО: это index-драйвер objective_lots_district_class_idx; увод
+#     фильтра ПОСЛЕ дедупа = full seq-scan, убивает перф-фикс. Принятый компромисс
+#     паттерна parcels.py/#1964 (≤4 возможных дрейфа при district-scoped запросе).
+#   • class — МЕНЯЮЩИЙСЯ (248 transient-групп на проде): класс физлота дрейфует между
+#     снапшотами. Фильтр lower(class)=… ДО DISTINCT ON ловил бы СТАРЫЙ снапшот под
+#     старым классом (прод-замер: бизнес 16637→16657 +20, стандарт 128721→128914 +193
+#     vs view). View фильтрует class на ПОСЛЕДНЕМ снапшоте → class-фильтр применяем
+#     ПОСЛЕ дедупа (в sold-CTE). Тогда эквивалентность view бит-в-бит по ВСЕМ классам.
+#   • sold-фильтр (is_sold / contract_date) — тоже МЕНЯЮЩИЙСЯ: ПОСЛЕ дедупа (на
+#     ПОСЛЕДНЕМ снапшоте), как делал view. Протолкнуть ДО DISTINCT ON СЛОМАЛО БЫ
+#     семантику (включил бы когда-либо-проданные + дедуп среди sold-снапшотов).
 _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 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)
+    WITH latest AS (
+        SELECT DISTINCT ON (
+            project_name, corpus_name, section, floor, lot_number
+        )
+            objective_lot_id,
+            is_sold,
+            contract_date,
+            class,
+            encumbrance_type,
+            bank_name
+          FROM objective_lots
+         WHERE premise_kind = CAST(:premise_kind AS text)
            AND (
                 CAST(:has_district AS boolean) IS FALSE
-                OR ol.district = ANY(CAST(:districts AS text[]))
+                OR district = ANY(CAST(:districts AS text[]))
                )
+         ORDER BY project_name, corpus_name, section, floor,
+                  lot_number, snapshot_date DESC, id DESC
+    ),
+    sold AS (
+        SELECT
+            objective_lot_id,
+            (encumbrance_type IS NOT NULL OR bank_name IS NOT NULL) AS has_mortgage
+          FROM latest
+         WHERE (is_sold IS TRUE OR contract_date IS NOT NULL)
            AND (
                 CAST(:obj_class AS text) IS NULL
-                OR lower(ol.class) = lower(CAST(:obj_class AS text))
+                OR lower(class) = lower(CAST(:obj_class AS text))
                )
     )
     SELECT

backend/app/services/site_finder/supply_layers.py

@@ -185,29 +185,46 @@ def _l2_confidence(n_objects_total: int, n_objects_with_free_flats: int) -> Conf
 # / наличием 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 отдаёт по строке на физический лот.
+# #1964: дедуп до ФИЗИЧЕСКОГО лота. Сырая objective_lots раздута ~2.91× (мульти
+# objective_lot_id на один физлот через пере-листинги) → units_available по району
+# был завышен ~3-4.6× (прод: Юго-Западный комфорт 58 606 → 12 620) → дефицит-форсайт
+# переоценивал открытое предложение L1.
+#
+# Перф-батч #2 (#1953-followup): дедупим INLINE через DISTINCT ON (physflat-ключ,
+# последний снапшот), НЕ через v_objective_lots_latest. Причина: view селектит ol.* и
+# его DISTINCT ON — оптимизационный барьер: WHERE снаружи (premise_kind/district) НЕ
+# проталкивается ниже DISTINCT ON → планировщик материализует ВСЮ таблицу (1.76M,
+# Parallel Seq Scan width≈891 + external merge sort 76-101 МБ/воркер) ПЕРЕД фильтром
+# (~6.0 s, прод-EXPLAIN). Inline протолкивает СТАБИЛЬНЫЕ фильтры (premise_kind, district
+# — НЕ меняются между снапшотами физлота: премайз-кайнд 0 mixed-групп на проде, район ~4
+# transient на 605k) В WHERE ДО DISTINCT ON → bitmap по objective_lots_district_class_idx
+# + узкий sort (~5 МБ/воркер) ~0.47 s district-scoped (≈13× быстрее; EKB-wide остаётся
+# seq-scan, но width≈99 вместо 891 → ~4.6 s). Эквивалентность view-vs-inline доказана
+# бит-в-бит на проде (все district×class пары, 0 расхождений) — зеркало parcels.py.
+# sold_now (is_sold/contract_date/status='продан') — МЕНЯЮЩИЕСЯ атрибуты → считаем
+# ПОСЛЕ дедупа (на последнем снапшоте), как делал view.
 _L1_OPEN_SQL = text(
     """
     WITH lots AS (
-        SELECT
-            ol.district,
-            ol.class AS obj_class,
-            ol.area_pd,
+        SELECT DISTINCT ON (
+            project_name, corpus_name, section, floor, lot_number
+        )
+            district,
+            class AS obj_class,
+            area_pd,
             (
-                ol.is_sold IS TRUE
-                OR ol.contract_date IS NOT NULL
-                OR LOWER(COALESCE(ol.status, '')) = 'продан'
+                is_sold IS TRUE
+                OR contract_date IS NOT NULL
+                OR LOWER(COALESCE(status, '')) = 'продан'
             ) AS sold_now
-          FROM v_objective_lots_latest ol
-         WHERE ol.premise_kind = :premise_kind
+          FROM objective_lots
+         WHERE premise_kind = CAST(:premise_kind AS text)
            AND (
                 CAST(:has_district AS boolean) IS FALSE
-                OR ol.district = ANY(CAST(:districts AS text[]))
+                OR district = ANY(CAST(:districts AS text[]))
                )
+         ORDER BY project_name, corpus_name, section, floor,
+                  lot_number, snapshot_date DESC, id DESC
     )
     SELECT
         district,