sprint1: nspd scraper industrialization, per-bucket elasticity, cadastre cross-check, sentry releases

- NSPD-skraper переехал в backend/app/services/scrapers/nspd_kn.py +
  Celery task scrape_nspd_region (beat: 20-е февраля/мая/авг/нояб).
  Redis lock 3h, WAF auto-retry, heartbeat в nspd_scrape_runs.
- Recommend_mix Tier 3: per-bucket elasticity через регрессию по
  «доминирующему bucket» каждого ЖК. Weighted-elasticity для inverse-mode.
  UI показывает разброс эластичностей и переключение regression/fallback.
- Cadastre vs market cross-check: spatial-join cad_buildings →
  ekb_districts_geom; cadastre_vs_market_pct в scope, аномалии
  (>+50% / <-30%) подсвечены в UI.
- Sentry release tracking (#4): IMAGE_TAG → backend/.env.runtime →
  sentry_sdk.init(release=...). Compose v2 env_file optional path.

Schemas: 63_schema_nspd_runs.sql (cad_buildings + nspd_scrape_runs/log
формализуют то, что уже жило в проде через 61_import_nspd_batch.py),
64_v_zk_rosreestr_velocity.sql (refresh с cad_buildings).

.github/workflows/deploy.yml

@@ -92,6 +92,12 @@ jobs:
             git fetch origin main
             git reset --hard origin/main
 
+            # Sentry release tracking — записываем git-sha в .env.runtime
+            # (отдельный файл, чтобы не трогать ручной .env с секретами).
+            # backend/worker/beat подхватывают его через env_file (см. compose).
+            mkdir -p backend
+            printf 'SENTRY_RELEASE=%s\n' "$IMAGE_TAG" > backend/.env.runtime
+
             export IMAGE_TAG="$IMAGE_TAG"
             docker compose -f docker-compose.prod.yml pull
             docker compose -f docker-compose.prod.yml up -d

.gitignore

@@ -21,6 +21,7 @@ out/
 .env
 .env.local
 .env.*.local
+.env.runtime
 .mcp.json
 
 # IDE

backend/.env.runtime.example

@@ -0,0 +1,6 @@
+# Auto-generated runtime overlay (deploy.yml пишет git-sha сюда).
+# Backend / worker / beat подхватывают через docker-compose env_file.
+# В локальной разработке файл создаётся пустым (compose скиппит env_file который не существует
+# при опциональной форме `${...:-}`, но мы используем явный список — поэтому
+# держим этот файл в репо как пустой шаблон, чтобы compose не падал).
+SENTRY_RELEASE=

backend/app/core/config.py

@@ -8,6 +8,10 @@ class Settings(BaseSettings):
     redis_url: str = "redis://localhost:6379/0"
     cors_origins: list[str] = ["http://localhost:3000"]
     sentry_dsn: str | None = None
+    # Release tag для Sentry — обычно git short sha, проставляется
+    # deploy.yml в backend/.env.runtime (см. workflow). Локально оставляем
+    # пустым — Sentry припишет 'unknown'.
+    sentry_release: str | None = None
     environment: str = "dev"
 
     # External APIs (Stage 2)
@@ -30,5 +34,14 @@ class Settings(BaseSettings):
     # Empty string = endpoint disabled.
     scrape_admin_token: str = ""
 
+    # NSPD-scraper schedule (кадастровые кварталы / здания).
+    # По умолчанию: 20-е число февраля/мая/августа/ноября в 03:30 МСК
+    # (после квартальных публикаций rosreestr_deals).
+    scrape_nspd_cron: str = "30 3 20 2,5,8,11 *"
+    # Регионы для NSPD-sweep (comma-separated rosreestr region codes).
+    scrape_nspd_default_regions: str = "66"
+    # Пауза между NSPD-запросами в мс. <600мс — высокий риск WAF 403.
+    scrape_nspd_rate_ms: int = 600
+
 
 settings = Settings()

backend/app/main.py

@@ -15,6 +15,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         sentry_sdk.init(
             dsn=settings.sentry_dsn,
             environment=settings.environment,
+            release=settings.sentry_release,
             traces_sample_rate=0.1,
         )
     yield

backend/app/services/analytics_queries.py

@@ -6,7 +6,6 @@ Region 66 = Sverdlovskaya oblast. Developer 6208_0 = PRINZIP.
 
 from __future__ import annotations
 
-import math
 from decimal import Decimal
 from typing import Any
 
@@ -1251,17 +1250,80 @@ def _city_avg_poi_score(db: Session, *, region_code: int = 66) -> float | None:
     return _f(row["avg_score"]) if row else None
 
 
+def _district_cadastre_baseline(db: Session, *, district_name: str) -> dict[str, Any]:
+    """Медианная кадастровая стоимость ₽/м² жилых строений в районе через
+    spatial-join cad_buildings → ekb_districts_geom. Возвращает None полей,
+    если в районе нет cad_buildings со cost_value.
+
+    Используется как cross-check для market price из rosreestr_deals:
+    cadastre_vs_market_pct > +50% (рынок сильно дороже кадастра, переоценка)
+    или < -30% (рынок дешевле кадастра, аномалия) → warning badge на UI.
+    """
+    row = (
+        db.execute(
+            text(
+                """
+                WITH district_geom AS (
+                    SELECT geom
+                      FROM ekb_districts_geom
+                     WHERE district_name = :dn
+                     LIMIT 1
+                ),
+                buildings_in AS (
+                    SELECT
+                        cb.cost_value / NULLIF(cb.area, 0) AS price_per_m2
+                      FROM cad_buildings cb
+                      JOIN district_geom dg
+                        ON ST_Intersects(dg.geom, cb.geom)
+                     WHERE cb.cost_value IS NOT NULL
+                       AND cb.area IS NOT NULL
+                       AND cb.area >= 100
+                       AND (cb.floors IS NOT NULL AND cb.floors >= 3
+                            OR cb.purpose ILIKE '%многокв%')
+                       AND (cb.cost_value / NULLIF(cb.area, 0))
+                           BETWEEN 5000 AND 500000
+                )
+                SELECT
+                    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_per_m2)
+                        AS median_per_m2,
+                    COUNT(*)::bigint AS n
+                  FROM buildings_in
+                """
+            ),
+            {"dn": district_name},
+        )
+        .mappings()
+        .first()
+    )
+    if not row or row["n"] == 0:
+        return {"median_per_m2": None, "buildings_n": 0}
+    return {
+        "median_per_m2": _f(row["median_per_m2"]),
+        "buildings_n": int(row["n"]),
+    }
+
+
 def _current_mortgage_rate(db: Session) -> tuple[float | None, str | None]:
-    """Последняя средневзвешенная ипотечная ставка из cbr_mortgage_series.
+    """Последняя средневзвешенная ставка ИЖК из cbr_mortgage_series.
-    Возвращает (rate_pct, period_label)."""
+
+    ВАЖНО: возвращаем СРЕДНЕВЗВЕШЕННУЮ С льготами (семейная/IT/ДВ-ипотека) —
+    это ~7-8%. РЫНОЧНАЯ ставка без льгот в БД отсутствует (она ~20% по
+    публикациям ЦБ Янв 2026, но в наших cbr_mortgage_series этой серии нет).
+
+    Старый ILIKE '%ипотечн%жилищн%' случайно матчил «долю ипотечных кредитов
+    на ИЖС» (5.57% на ИЖС — НЕ ставка). Теперь строго matchим
+    'Средневзвешенная ставка по ипотечным жилищным' + 'в рублях, %'.
+    """
     row = (
         db.execute(
             text(
                 """
                 SELECT value, period
                   FROM cbr_mortgage_series
-                 WHERE title ILIKE '%ипотечн%жилищн%'
+                 WHERE title ILIKE 'Средневзвешенная ставка по ипотечным жилищным%'
+                   AND title ILIKE '%в рублях, %'
                    AND value IS NOT NULL
+                   AND value BETWEEN 1 AND 30  -- защита от мусорных
                  ORDER BY period DESC
                  LIMIT 1
                 """
@@ -1392,6 +1454,111 @@ def _elasticity_coef(
     }
 
 
+def _elasticity_per_bucket_coef(
+    db: Session,
+    *,
+    region_code: int,
+    district_name: str,
+    target_class: str | None,
+    fallback: dict[str, Any],
+) -> dict[str, dict[str, Any]]:
+    """Per-bucket эластичность (Tier 3): группируем sale_graph-наблюдения по
+    «доминирующему bucket» каждого ЖК (mode total_area из domrf_kn_flats),
+    регрессия log-log для каждой группы. Студии vs 80+ м² реагируют на цену
+    по-разному.
+
+    Returns: dict[bucket_pretty → {elasticity, r2, n, source}]. Если в bucket'е
+    меньше 30 точек или регрессия слабая (R²<0.05 либо positive slope) — берём
+    общую эластичность из `fallback` со source='fallback_global'.
+    """
+    where_class = "AND o.obj_class = :cls" if target_class else ""
+    params: dict[str, Any] = {"rc": region_code, "dn": district_name}
+    if target_class:
+        params["cls"] = target_class
+    rows = (
+        db.execute(
+            text(
+                f"""
+                WITH obj_pool AS (
+                    SELECT o.obj_id
+                      FROM domrf_kn_objects o
+                     WHERE o.region_cd = :rc
+                       AND o.district_name = :dn
+                       {where_class}
+                ),
+                obj_bucket AS (
+                    -- Доминирующий bucket каждого ЖК = mode total_area среди
+                    -- его flats. Если flats пусты — ЖК не учитывается.
+                    SELECT
+                        f.obj_id,
+                        CASE
+                            WHEN PERCENTILE_CONT(0.5) WITHIN GROUP
+                                (ORDER BY f.total_area) < 30 THEN '1-Студия'
+                            WHEN PERCENTILE_CONT(0.5) WITHIN GROUP
+                                (ORDER BY f.total_area) < 45 THEN '2-1-к'
+                            WHEN PERCENTILE_CONT(0.5) WITHIN GROUP
+                                (ORDER BY f.total_area) < 60 THEN '3-2-к'
+                            WHEN PERCENTILE_CONT(0.5) WITHIN GROUP
+                                (ORDER BY f.total_area) < 80 THEN '4-3-к'
+                            ELSE                                  '5-80+ м²'
+                        END AS bucket
+                      FROM domrf_kn_flats f
+                      JOIN obj_pool p ON p.obj_id = f.obj_id
+                     WHERE f.total_area IS NOT NULL
+                       AND f.total_area BETWEEN 15 AND 200
+                     GROUP BY f.obj_id
+                ),
+                pts AS (
+                    SELECT
+                        ob.bucket,
+                        LN(sg.realised)::float8   AS y,
+                        LN(sg.price_avg)::float8  AS x
+                      FROM domrf_kn_sale_graph sg
+                      JOIN obj_bucket ob ON ob.obj_id = sg.obj_id
+                     WHERE sg.type = 'apartments'
+                       AND sg.realised IS NOT NULL AND sg.realised > 0
+                       AND sg.price_avg IS NOT NULL AND sg.price_avg > 0
+                       AND sg.report_month >= NOW() - INTERVAL '36 months'
+                )
+                SELECT bucket,
+                       regr_slope(y, x)  AS slope,
+                       regr_r2(y, x)     AS r2,
+                       COUNT(*)::bigint  AS n
+                  FROM pts
+                 GROUP BY bucket
+                """
+            ),
+            params,
+        )
+        .mappings()
+        .all()
+    )
+
+    out: dict[str, dict[str, Any]] = {}
+    fallback_e = float(fallback["elasticity"])
+    by_bucket = {r["bucket"]: r for r in rows}
+    for bucket_id, bucket_pretty in _BUCKET_PRETTY.items():
+        r = by_bucket.get(bucket_id)
+        n_b = int(r["n"]) if r and r["n"] is not None else 0
+        slope = _f(r["slope"]) if r else None
+        r2 = _f(r["r2"]) if r else None
+        if n_b >= 30 and slope is not None and r2 is not None and r2 >= 0.05 and slope < 0:
+            out[bucket_pretty] = {
+                "elasticity": round(slope, 4),
+                "r2": round(r2, 4),
+                "n": n_b,
+                "source": "regression",
+            }
+        else:
+            out[bucket_pretty] = {
+                "elasticity": fallback_e,
+                "r2": round(r2, 4) if r2 is not None else 0.0,
+                "n": n_b,
+                "source": "fallback_global",
+            }
+    return out
+
+
 def recommend_mix(
     db: Session,
     *,
@@ -1621,18 +1788,8 @@ def recommend_mix(
         district_name=district_row["district_name"],
         target_class=target_class_for_geo,
     )
-    market_vel_pm = vel["realised_per_month_median"] or vel["realised_per_month_avg"]
+    sale_graph_vel_pm = vel["realised_per_month_median"] or vel["realised_per_month_avg"]
-    if market_vel_pm is None:
+    velocity_source = "sale_graph" if sale_graph_vel_pm is not None else "rosreestr_fallback"
-        # Fallback: derive from city-wide rosreestr deals (distribute per bucket
-        # by share). Coarser, but lets the calculator work anywhere.
-        warnings.append(
-            "Нет sale_graph данных для этого района и класса —"
-            " темп считается по rosreestr-сделкам (грубее)."
-        )
-        market_vel_pm = (total_deals / max(effective_window, 1)) if total_deals else 0.0
-        velocity_source = "rosreestr_fallback"
-    else:
-        velocity_source = "sale_graph"
 
     elast = _elasticity_coef(
         db,
@@ -1648,8 +1805,20 @@ def recommend_mix(
             " (недостаточно для регрессии)."
         )
 
+    # Tier 3: per-bucket эластичность. Регрессия sale_graph по
+    # «доминирующему bucket» каждого ЖК. Если для bucket'а данных мало —
+    # подставляем общую elasticity. Малые сегменты (1-2 студии в районе)
+    # таким образом не выкидываются — используем общую кривую.
+    elast_per_bucket = _elasticity_per_bucket_coef(
+        db,
+        region_code=region_code,
+        district_name=district_row["district_name"],
+        target_class=target_class_for_geo,
+        fallback=elast,
+    )
+
     # 5b-1) N активных конкурентов с каскадным fallback (район+класс →
-    # район → регион). Используется для нормировки рыночной velocity.
+    # район → регион). Используется как divisor в rosreestr-fallback ветке.
     competitors, competitors_scope = _active_competitors_count(
         db,
         region_code=region_code,
@@ -1672,14 +1841,31 @@ def recommend_mix(
             f" нормировка по {competitors} ЖК в {scope_label}."
         )
 
-    # 5b-2) Per-bucket market velocity (сделок/мес для каждого размерного
+    # 5b-2) market_vel_pm = «что продаёт ОДИН активный ЖК района за месяц».
-    # сегмента из rosreestr — НЕ city-wide, а РЕАЛЬНАЯ интенсивность сегмента).
+    # ИСТОЧНИК ИСТИНЫ — sale_graph (median realised per ЖК). При отсутствии —
-    # Студии/1к — обычно выше, 80+ — ниже.
+    # rosreestr-fallback: city-wide deals/mo / N_competitors → per-ЖК proxy.
-    bucket_market_velocities = {
+    # Это критично: per-ЖК baseline должен иметь правильную размерность
-        _BUCKET_PRETTY.get(r["bucket"], r["bucket"]): (
+    # (~3-7 кв/мес для ЕКБ ЖК), иначе months_to_sellout получается
-            int(r["deals"] or 0) / max(effective_window, 1)
+    # нереалистично коротким.
+    if sale_graph_vel_pm is not None:
+        market_vel_pm = sale_graph_vel_pm
+    else:
+        warnings.append(
+            "Нет sale_graph данных для этого района и класса —"
+            " темп считается по rosreestr-сделкам ÷ конкуренты (грубее)."
         )
-        for r in bucket_rows
+        market_vel_pm = (
+            (total_deals / max(effective_window, 1) / max(competitors, 1))
+            if total_deals and competitors
+            else 0.0
+        )
+
+    # 5b-2.5) Per-bucket market velocity = market_vel_pm × share / 100.
+    # Аллоцируем единый per-ЖК baseline на размерные сегменты по shares
+    # (одинаковая модель для sale_graph и rosreestr_fallback). Студии/1к
+    # получат больший абсолютный темп если их share высокая в районе.
+    bucket_market_velocities = {
+        b["bucket"]: market_vel_pm * (b["share_pct"] / 100.0) for b in buckets
     }
 
     # 5b-2.5) Дополнительные district-specific signals (Tier 2):
@@ -1699,6 +1885,11 @@ def recommend_mix(
 
     poi_score = _district_poi_score(db, district_name=district_row["district_name"])
     city_avg_poi = _city_avg_poi_score(db, region_code=region_code)
+
+    # Cadastre cross-check: медианная кадастровая стоимость ₽/м² района через
+    # cad_buildings → ekb_districts spatial-join. Аномалии (рынок vs кадастр)
+    # выводятся как warning-цена в RecommendVelocityPanel.
+    cadastre = _district_cadastre_baseline(db, district_name=district_row["district_name"])
     poi_factor = (
         1 + (poi_score - city_avg_poi) / max(city_avg_poi, 1) * 0.05
         if (poi_score is not None and city_avg_poi is not None and city_avg_poi > 0)
@@ -1708,23 +1899,31 @@ def recommend_mix(
     mortgage_rate, mortgage_period = _current_mortgage_rate(db)
 
     # 5b-3) Per-bucket project velocity at price_factor=1.0:
-    #   bucket_market_v   = темп РЫНКА для bucket'а (deals/mo по всему региону)
+    #   bucket_market_v   = market_vel_pm × bucket.share/100 — доля per-ЖК
-    #   normalisation     = sqrt(N_competitors) — power-law эффективные
+    #                       темпа, аллоцированная на размерный сегмент.
-    #                       конкуренты (sqrt компромисс между ÷1 и ÷N).
+    #                       market_vel_pm УЖЕ per-ЖК (median sale_graph либо
-    #   project_velocity  = bucket_market_v / sqrt(N) × sat_factor × trend_factor
+    #                       rosreestr/N_competitors), доп. нормировка не нужна.
+    #   project_velocity  = bucket_market_v × sat_factor × trend_factor
     #                       sat — зрелый рынок ускоряет; trend — текущая
     #                       динамика (горит/остывает).
     #   adjusted          = project_velocity × price_factor^elasticity
     #   months_to_sellout = units_planned / adjusted
     # Цена тоже корректируется на poi_factor (развитость района = премиум).
     pf_pow = price_factor**elasticity if price_factor > 0 else 1.0
-    competitors_norm = math.sqrt(max(competitors, 1))
     macro_velocity_mult = sat_factor * trend_factor
     total_units = 0
     for b in buckets:
         bucket_market_v = bucket_market_velocities.get(b["bucket"], 0.0)
-        bucket_velocity = round(bucket_market_v / competitors_norm * macro_velocity_mult, 4)
+        bucket_velocity = round(bucket_market_v * macro_velocity_mult, 4)
         b["velocity_per_month"] = bucket_velocity
+        # Per-bucket эластичность: ключ — pretty-имя (b["bucket"] уже pretty).
+        be = elast_per_bucket.get(b["bucket"]) or {}
+        bucket_elasticity = float(be.get("elasticity", elasticity))
+        bucket_pf_pow = price_factor**bucket_elasticity if price_factor > 0 else 1.0
+        b["elasticity"] = bucket_elasticity
+        b["elasticity_r2"] = be.get("r2", 0.0)
+        b["elasticity_n"] = be.get("n", 0)
+        b["elasticity_source"] = be.get("source", "fallback_global")
         # POI-корректировка на цену (на ВСЕ p25/median/p75)
         b["price_median_per_m2"] = round(b["price_median_per_m2"] * poi_factor, 2)
         b["price_p25_per_m2"] = round(b["price_p25_per_m2"] * poi_factor, 2)
@@ -1733,7 +1932,7 @@ def recommend_mix(
             # Revenue тоже пересчитываем после POI-correction (linear scale).
             if b["revenue_planned_rub"] is not None:
                 b["revenue_planned_rub"] = round(b["revenue_planned_rub"] * poi_factor, 2)
-            adjusted_velocity = bucket_velocity * pf_pow
+            adjusted_velocity = bucket_velocity * bucket_pf_pow
             b["months_to_sellout"] = (
                 round(b["units_planned"] / adjusted_velocity, 1) if adjusted_velocity > 0 else None
             )
@@ -1747,18 +1946,26 @@ def recommend_mix(
         weighted_avg_price = round(weighted_avg_price * poi_factor, 2)
 
     # 5c) Inverse mode: target_months → required price_factor.
-    # required_velocity = total_units / target_months
+    # Tier 3: используем weighted-by-units эластичность (per-bucket эластичности
-    # base_velocity_total = sum(bucket_velocity)  (at price_factor=1)
+    # → агрегатная только когда нужна одна цифра). При smooth-buckets разница
-    # required_pf^elasticity = required_velocity / base_velocity_total
+    # с глобальной невелика, но если bucket-mix сильно перекошен в одну сторону —
-    # → required_pf = (required_velocity / base_velocity_total)^(1/elasticity)
+    # weighted-эластичность точнее отражает поведение портфеля.
     required_price_factor: float | None = None
+    weighted_elasticity = elasticity
+    if total_units > 0:
+        weighted_elasticity = (
+            sum(
+                (b.get("elasticity") or elasticity) * (b.get("units_planned") or 0) for b in buckets
+            )
+            / total_units
+        )
     if target_months and total_units > 0:
         base_total_velocity = sum(b["velocity_per_month"] or 0 for b in buckets)
-        if base_total_velocity > 0 and elasticity != 0:
+        if base_total_velocity > 0 and weighted_elasticity != 0:
             required_velocity = total_units / target_months
             ratio = required_velocity / base_total_velocity
             try:
-                required_price_factor = round(ratio ** (1.0 / elasticity), 4)
+                required_price_factor = round(ratio ** (1.0 / weighted_elasticity), 4)
             except Exception:
                 required_price_factor = None
             if required_price_factor and required_price_factor < 0.7:
@@ -1781,11 +1988,19 @@ def recommend_mix(
             sold_24mo += frac * up
         liquidity_24mo = round(sold_24mo / total_units * 100, 1)
 
-    # 5e) Aggregate KPIs
+    # 5e) Aggregate KPIs. Total months_to_sellout считаем как сумму
+    # bucket-уровневых adjusted velocities (каждая со своим pf_pow по своей
+    # эластичности) — иначе перекос в bucket-mix искажает агрегат.
     months_to_sellout_total: float | None = None
     base_total_v = sum(b["velocity_per_month"] or 0 for b in buckets)
-    if total_units > 0 and base_total_v > 0:
+    adjusted_total_v = 0.0
-        months_to_sellout_total = round(total_units / (base_total_v * pf_pow), 1)
+    for b in buckets:
+        v = b.get("velocity_per_month") or 0
+        be = b.get("elasticity")
+        bpf = price_factor**be if (be is not None and price_factor > 0) else pf_pow
+        adjusted_total_v += v * bpf
+    if total_units > 0 and adjusted_total_v > 0:
+        months_to_sellout_total = round(total_units / adjusted_total_v, 1)
     avg_ticket = (
         round(total_revenue / total_units, 2) if (have_revenue and total_units > 0) else None
     )
@@ -1835,8 +2050,9 @@ def recommend_mix(
     if avg_ticket:
         headline_parts.append(f"ср. чек {round(avg_ticket / 1_000_000, 1)} М ₽")
     if base_total_v > 0:
-        # Малая velocity — формат с 2 десятыми (0.07 кв/мес для ЖК-доли).
+        # Tempo = sum bucket-adjusted velocities (каждая со своим pf_pow по своей
-        tempo = base_total_v * pf_pow
+        # эластичности). Это согласовано с months_to_sellout_total выше.
+        tempo = adjusted_total_v if adjusted_total_v > 0 else base_total_v * pf_pow
         headline_parts.append(
             f"темп {tempo:.2f} кв/мес" if tempo < 1 else f"темп {tempo:.1f} кв/мес"
         )
@@ -1881,6 +2097,24 @@ def recommend_mix(
             "elasticity_r2": elast["r2"],
             "elasticity_n": elast["n"],
             "elasticity_source": elast["source"],
+            "elasticity_weighted": (round(weighted_elasticity, 4) if total_units > 0 else None),
+            "elasticity_per_bucket": elast_per_bucket,
+            "cadastre_median_per_m2": (
+                round(cadastre["median_per_m2"], 0)
+                if cadastre["median_per_m2"] is not None
+                else None
+            ),
+            "cadastre_buildings_n": cadastre["buildings_n"],
+            "cadastre_vs_market_pct": (
+                round(
+                    (district_median - cadastre["median_per_m2"])
+                    / cadastre["median_per_m2"]
+                    * 100.0,
+                    1,
+                )
+                if (cadastre["median_per_m2"] and cadastre["median_per_m2"] > 0 and district_median)
+                else None
+            ),
             "price_factor_applied": round(price_factor, 4),
             "required_price_factor": required_price_factor,
             "target_months": target_months,

backend/app/services/scrapers/nspd_kn.py

@@ -0,0 +1,536 @@
+"""NSPD scraper — кадастровые кварталы и здания через nspd.gov.ru API.
+
+Endpoints:
+    GET /api/geoportal/v2/search/geoportal?thematicSearchId=2&query={cad}
+        → polygon кадастрового квартала (1 feature с label = cad-номер)
+    GET /api/geoportal/v2/search/geoportal?thematicSearchId=1&query={cad}
+        → объекты внутри квартала (здания, земля, линейные).
+        Фильтруем categoryName='Здания' и cad_num LIKE '<region>:<district>:%'.
+
+WAF nspd.gov.ru банит burst-запросы и нероссийские IP. Стратегия:
+- rate-limit 600мс между запросами (≤100 req/min — наблюдаемый предел)
+- exponential backoff на 403 (60s + 30s × attempt)
+- запускать с RU IP (production worker должен быть в RU/CIS-zone)
+
+Этот модуль — рефакторинг data/sql/62_scrape_nspd_full.py из stand-alone скрипта
+в библиотечную функцию для Celery. Логика идентична, но:
+- работает через SQLAlchemy Session (FK к cad_quarters_geom/cad_buildings)
+- пишет heartbeat и progress в nspd_scrape_runs
+- логирует структурированно в nspd_scrape_log
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import ssl
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+NSPD_BASE = "https://nspd.gov.ru/api/geoportal/v2/search/geoportal"
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+    "Accept-Language": "ru-RU,ru;q=0.9",
+    "Referer": "https://nspd.gov.ru/map",
+}
+SSL_CTX = ssl._create_unverified_context()
+
+DEFAULT_RATE_MS = 600
+DEFAULT_HEARTBEAT_EVERY = 5  # quarters
+DEFAULT_COMMIT_EVERY = 10
+DEFAULT_RETRIES = 5
+DEFAULT_TIMEOUT_S = 30
+
+# region_code (rosreestr) → cad-prefix фильтр для cad_buildings.
+# 66 = Свердловская обл., 66:41 = ЕКБ. Для других регионов добавлять mapping.
+REGION_CAD_PREFIX: dict[int, str] = {
+    66: "66:41:",  # Екатеринбург
+}
+
+
+# ── HTTP layer ────────────────────────────────────────────────────────────────
+
+
+class WafBlockedError(RuntimeError):
+    """403 после всех retry — WAF банит, прогон не продолжаем."""
+
+
+def nspd_fetch(
+    thematic_search_id: int,
+    query: str,
+    *,
+    retries: int = DEFAULT_RETRIES,
+    timeout: int = DEFAULT_TIMEOUT_S,
+    on_403: Any = None,
+) -> dict | None:
+    """Fetch NSPD с retry/backoff. on_403 callable(attempt) — для трекинга WAF."""
+    qs = urllib.parse.urlencode({"thematicSearchId": thematic_search_id, "query": query})
+    url = f"{NSPD_BASE}?{qs}"
+    req = urllib.request.Request(url, headers=HEADERS)
+    last_err: Exception | None = None
+    for attempt in range(retries):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r:
+                return json.loads(r.read().decode("utf-8"))
+        except urllib.error.HTTPError as e:
+            last_err = e
+            if e.code == 404:
+                return None
+            if e.code == 403:
+                if on_403:
+                    on_403(attempt)
+                wait = 60 + 30 * attempt
+                logger.warning(
+                    "WAF 403 для %s (попытка %d/%d), пауза %ds", query, attempt + 1, retries, wait
+                )
+                time.sleep(wait)
+                continue
+            logger.warning("HTTP %s для %s (попытка %d)", e.code, query, attempt + 1)
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            last_err = e
+            logger.warning("Сетевая ошибка для %s: %s (попытка %d)", query, e, attempt + 1)
+        time.sleep(min(2**attempt, 30))
+    if isinstance(last_err, urllib.error.HTTPError) and last_err.code == 403:
+        raise WafBlockedError(f"WAF banned after {retries} retries on {query}")
+    logger.error("FAILED %s после %d попыток: %s", query, retries, last_err)
+    return None
+
+
+# ── Geometry helpers ──────────────────────────────────────────────────────────
+
+
+def poly_to_wkt(geom: dict | None) -> str | None:
+    if not geom:
+        return None
+
+    def ring(r: list) -> str:
+        return "(" + ",".join(f"{p[0]} {p[1]}" for p in r) + ")"
+
+    t = geom.get("type")
+    if t == "Polygon":
+        return "POLYGON(" + ",".join(ring(r) for r in geom["coordinates"]) + ")"
+    if t == "MultiPolygon":
+        return (
+            "MULTIPOLYGON("
+            + ",".join("(" + ",".join(ring(r) for r in p) + ")" for p in geom["coordinates"])
+            + ")"
+        )
+    if t == "Point":
+        return f"POINT({geom['coordinates'][0]} {geom['coordinates'][1]})"
+    return None
+
+
+# ── SQL helpers ───────────────────────────────────────────────────────────────
+
+
+def _to_int(x: Any) -> int | None:
+    if x is None or x == "":
+        return None
+    try:
+        return int(x)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_num(x: Any) -> float | None:
+    if x is None or x == "":
+        return None
+    try:
+        return float(x)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_date(x: Any) -> str | None:
+    if not x:
+        return None
+    s = str(x).split(" ")[0]
+    return s or None
+
+
+def get_pending_cads(db: Session, region_code: int) -> list[str]:
+    """Cad-кварталы с ДДУ-сделками в регионе минус уже scraped."""
+    rows = db.execute(
+        text(
+            """
+            SELECT DISTINCT quarter_cad_number
+              FROM rosreestr_deals
+             WHERE region_code = :rc
+               AND doc_type = 'ДДУ'
+               AND realestate_type_code = '002001003000'
+               AND quarter_cad_number IS NOT NULL
+               AND quarter_cad_number <> ''
+               AND quarter_cad_number !~ '^00:00:'
+               AND quarter_cad_number !~ ':0000000$'
+            """
+        ),
+        {"rc": region_code},
+    ).all()
+    pending = {r[0] for r in rows}
+    done = {r[0] for r in db.execute(text("SELECT cad_number FROM cad_quarters_geom")).all()}
+    return sorted(pending - done)
+
+
+def insert_quarter(db: Session, cad: str, wkt: str, raw_props: dict | None) -> None:
+    db.execute(
+        text(
+            """
+            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, source)
+            VALUES (
+                :cad,
+                ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromText(:wkt), 3857), 4326))
+                    ::geometry(MultiPolygon, 4326),
+                CAST(:props AS jsonb),
+                'nspd'
+            )
+            ON CONFLICT (cad_number) DO UPDATE SET
+                geom = EXCLUDED.geom,
+                raw_props = COALESCE(EXCLUDED.raw_props, cad_quarters_geom.raw_props),
+                fetched_at = NOW()
+            """
+        ),
+        {
+            "cad": cad,
+            "wkt": wkt,
+            "props": json.dumps(raw_props, ensure_ascii=False) if raw_props else None,
+        },
+    )
+
+
+def insert_buildings(db: Session, qcad: str, features: list[dict], cad_prefix: str) -> int:
+    n = 0
+    for f in features:
+        props_outer = f.get("properties") or {}
+        if props_outer.get("categoryName") != "Здания":
+            continue
+        opts = props_outer.get("options") or {}
+        cn = opts.get("cad_num")
+        if not cn or not cn.startswith(cad_prefix):
+            continue
+        wkt = poly_to_wkt(f.get("geometry"))
+        if not wkt:
+            continue
+        db.execute(
+            text(
+                """
+                INSERT INTO cad_buildings (
+                    cad_num, quarter_cad_num, geom, purpose, building_name,
+                    readable_address, area, floors, year_built, year_commisioning,
+                    cost_value, registration_date,
+                    status, ownership_type, cultural_heritage, underground_floors,
+                    build_record_area, build_record_type, common_data_status, obj_type,
+                    raw_props
+                )
+                VALUES (
+                    :cad_num, :qcad,
+                    ST_Transform(ST_SetSRID(ST_GeomFromText(:wkt), 3857), 4326),
+                    :purpose, :name, :addr, :area, :floors, :yb, :yc,
+                    :cost, :reg_date,
+                    :status, :ownership, :cultural, :underground,
+                    :build_rec_area, :build_rec_type, :common_status, :obj_type,
+                    CAST(:raw_props AS jsonb)
+                )
+                ON CONFLICT (cad_num) DO UPDATE SET
+                    cost_value = EXCLUDED.cost_value,
+                    area = EXCLUDED.area,
+                    year_built = EXCLUDED.year_built,
+                    year_commisioning = EXCLUDED.year_commisioning,
+                    status = EXCLUDED.status,
+                    ownership_type = EXCLUDED.ownership_type,
+                    cultural_heritage = EXCLUDED.cultural_heritage,
+                    underground_floors = EXCLUDED.underground_floors,
+                    build_record_area = EXCLUDED.build_record_area,
+                    build_record_type = EXCLUDED.build_record_type,
+                    common_data_status = EXCLUDED.common_data_status,
+                    obj_type = EXCLUDED.obj_type,
+                    raw_props = COALESCE(EXCLUDED.raw_props, cad_buildings.raw_props),
+                    fetched_at = NOW()
+                """
+            ),
+            {
+                "cad_num": cn,
+                "qcad": qcad,
+                "wkt": wkt,
+                "purpose": opts.get("purpose"),
+                "name": opts.get("building_name"),
+                "addr": opts.get("readable_address"),
+                "area": _to_num(opts.get("area")),
+                "floors": _to_int(opts.get("floors")),
+                "yb": _to_int(opts.get("year_built")),
+                "yc": _to_int(opts.get("year_commisioning")),
+                "cost": _to_num(opts.get("cost_value")),
+                "reg_date": _to_date(opts.get("registration_date")),
+                "status": opts.get("status"),
+                "ownership": opts.get("ownership_type"),
+                "cultural": opts.get("cultural_heritage"),
+                "underground": _to_int(opts.get("underground_floors")),
+                "build_rec_area": _to_num(opts.get("build_record_area")),
+                "build_rec_type": opts.get("build_record_type"),
+                "common_status": opts.get("common_data_status"),
+                "obj_type": opts.get("obj_type"),
+                "raw_props": json.dumps(opts, ensure_ascii=False) if opts else None,
+            },
+        )
+        n += 1
+    return n
+
+
+# ── Run lifecycle ─────────────────────────────────────────────────────────────
+
+
+def _start_run(db: Session, region_code: int, triggered_by: str, pending_count: int) -> int:
+    row = db.execute(
+        text(
+            """
+            INSERT INTO nspd_scrape_runs (region_code, triggered_by, pending_count, status)
+            VALUES (:rc, :tb, :pc, 'running')
+            RETURNING run_id
+            """
+        ),
+        {"rc": region_code, "tb": triggered_by, "pc": pending_count},
+    ).scalar_one()
+    db.commit()
+    return int(row)
+
+
+def _heartbeat(db: Session, run_id: int, **counts: int) -> None:
+    sets = ["heartbeat_at = NOW()"]
+    params: dict[str, Any] = {"rid": run_id}
+    for k, v in counts.items():
+        sets.append(f"{k} = :{k}")
+        params[k] = v
+    db.execute(
+        text(f"UPDATE nspd_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
+        params,
+    )
+    db.commit()
+
+
+def _finish_run(
+    db: Session,
+    run_id: int,
+    *,
+    status: str,
+    error: str | None = None,
+    **counts: int,
+) -> None:
+    sets = ["finished_at = NOW()", "status = :status"]
+    params: dict[str, Any] = {"rid": run_id, "status": status, "error": error}
+    sets.append("error = :error")
+    for k, v in counts.items():
+        sets.append(f"{k} = :{k}")
+        params[k] = v
+    db.execute(
+        text(f"UPDATE nspd_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
+        params,
+    )
+    db.commit()
+
+
+def _log(
+    db: Session,
+    run_id: int | None,
+    *,
+    level: str,
+    stage: str,
+    message: str,
+    cad: str | None = None,
+) -> None:
+    try:
+        db.execute(
+            text(
+                """
+                INSERT INTO nspd_scrape_log (run_id, level, stage, cad_number, message)
+                VALUES (:rid, :lvl, :st, :cad, :msg)
+                """
+            ),
+            {"rid": run_id, "lvl": level, "st": stage, "cad": cad, "msg": message[:1000]},
+        )
+        db.commit()
+    except Exception as e:
+        logger.warning("nspd_scrape_log insert failed: %s", e)
+
+
+# ── Main entrypoint ───────────────────────────────────────────────────────────
+
+
+def run_region_scrape(
+    *,
+    region_code: int,
+    triggered_by: str = "beat",
+    limit: int | None = None,
+    rate_ms: int = DEFAULT_RATE_MS,
+    commit_every: int = DEFAULT_COMMIT_EVERY,
+    heartbeat_every: int = DEFAULT_HEARTBEAT_EVERY,
+) -> dict[str, Any]:
+    """Полный sweep одного региона. Безопасен для повторных запусков —
+    pending = (rosreestr cads) − (cad_quarters_geom cads)."""
+    cad_prefix = REGION_CAD_PREFIX.get(region_code)
+    if not cad_prefix:
+        raise ValueError(f"region_code={region_code} unknown — добавь в REGION_CAD_PREFIX")
+
+    db = SessionLocal()
+    run_id: int | None = None
+    try:
+        pending = get_pending_cads(db, region_code)
+        if limit:
+            pending = pending[:limit]
+        run_id = _start_run(db, region_code, triggered_by, len(pending))
+        _log(
+            db,
+            run_id,
+            level="info",
+            stage="discover",
+            message=f"region={region_code} pending={len(pending)} cads",
+        )
+
+        if not pending:
+            _finish_run(db, run_id, status="done", quarters_ok=0, quarters_failed=0, buildings_ok=0)
+            return {"run_id": run_id, "pending": 0, "ok": 0, "failed": 0, "buildings": 0}
+
+        ok = 0
+        failed = 0
+        n_buildings = 0
+        n_requests = 0
+        n_waf = 0
+        started = time.time()
+
+        for i, cn in enumerate(pending, 1):
+            try:
+                j2 = nspd_fetch(2, cn, on_403=lambda _a: None)
+                n_requests += 1
+                qf = None
+                q_props: dict | None = None
+                if j2:
+                    for f in j2.get("data", {}).get("features") or []:
+                        if (f.get("properties") or {}).get("label") == cn:
+                            qf = f
+                            q_props = (f.get("properties") or {}).get("options") or {}
+                            break
+                wkt = poly_to_wkt(qf.get("geometry") if qf else None) if qf else None
+                if wkt:
+                    insert_quarter(db, cn, wkt, q_props)
+                    ok += 1
+                else:
+                    failed += 1
+                    _log(
+                        db,
+                        run_id,
+                        level="warn",
+                        stage="quarter_fetch",
+                        cad=cn,
+                        message="no polygon returned",
+                    )
+
+                j1 = nspd_fetch(1, cn, on_403=lambda _a: None)
+                n_requests += 1
+                if j1:
+                    features = j1.get("data", {}).get("features") or []
+                    n_buildings += insert_buildings(db, cn, features, cad_prefix)
+
+                if i % commit_every == 0:
+                    db.commit()
+                if i % heartbeat_every == 0:
+                    _heartbeat(
+                        db,
+                        run_id,
+                        quarters_ok=ok,
+                        quarters_failed=failed,
+                        buildings_ok=n_buildings,
+                        requests_count=n_requests,
+                        waf_429_count=n_waf,
+                    )
+
+            except WafBlockedError as e:
+                _log(
+                    db,
+                    run_id,
+                    level="error",
+                    stage="quarter_fetch",
+                    cad=cn,
+                    message=f"WAF blocked: {e}",
+                )
+                _finish_run(
+                    db,
+                    run_id,
+                    status="failed",
+                    error=str(e),
+                    quarters_ok=ok,
+                    quarters_failed=failed,
+                    buildings_ok=n_buildings,
+                    requests_count=n_requests,
+                    waf_429_count=n_waf + 1,
+                )
+                raise
+            except Exception as e:
+                failed += 1
+                db.rollback()
+                _log(
+                    db,
+                    run_id,
+                    level="error",
+                    stage="quarter_fetch",
+                    cad=cn,
+                    message=f"{type(e).__name__}: {e}",
+                )
+
+            time.sleep(rate_ms / 1000.0)
+
+        db.commit()
+        elapsed = time.time() - started
+        _finish_run(
+            db,
+            run_id,
+            status="done",
+            quarters_ok=ok,
+            quarters_failed=failed,
+            buildings_ok=n_buildings,
+            requests_count=n_requests,
+            waf_429_count=n_waf,
+        )
+        _log(
+            db,
+            run_id,
+            level="info",
+            stage="done",
+            message=(
+                f"elapsed={elapsed:.0f}s ok={ok} failed={failed} "
+                f"buildings={n_buildings} reqs={n_requests}"
+            ),
+        )
+        return {
+            "run_id": run_id,
+            "pending": len(pending),
+            "ok": ok,
+            "failed": failed,
+            "buildings": n_buildings,
+            "requests": n_requests,
+            "elapsed_s": int(elapsed),
+            "started_at": datetime.utcfromtimestamp(started).isoformat() + "Z",
+        }
+    except WafBlockedError:
+        raise
+    except Exception as e:
+        if run_id:
+            try:
+                _finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}")
+            except Exception:
+                pass
+        raise
+    finally:
+        db.close()

backend/app/workers/celery_app.py

@@ -30,11 +30,18 @@ def _default_regions() -> list[int]:
     return [int(x.strip()) for x in settings.scrape_kn_default_regions.split(",") if x.strip()]
 
 
+def _nspd_default_regions() -> list[int]:
+    return [int(x.strip()) for x in settings.scrape_nspd_default_regions.split(",") if x.strip()]
+
+
 celery_app = Celery(
     "gendesign",
     broker=settings.redis_url,
     backend=settings.redis_url,
-    include=["app.workers.tasks.scrape_kn"],
+    include=[
+        "app.workers.tasks.scrape_kn",
+        "app.workers.tasks.scrape_nspd",
+    ],
 )
 celery_app.conf.timezone = "Europe/Moscow"
 
@@ -50,6 +57,24 @@ celery_app.conf.beat_schedule = {
     for rc in _default_regions()
 }
 
+# NSPD-скрейп — квартально, после публикации rosreestr_deals.
+# Pending = (новые quarter_cad_number из rosreestr_deals) − (cad_quarters_geom).
+# При limit=None прогон захватит все накопленные cad-кварталы за квартал.
+celery_app.conf.beat_schedule.update(
+    {
+        f"nspd-region-{rc}": {
+            "task": "tasks.scrape_nspd.scrape_nspd_region",
+            "schedule": _parse_cron(settings.scrape_nspd_cron),
+            "kwargs": {
+                "region_code": rc,
+                "triggered_by": "beat",
+                "rate_ms": settings.scrape_nspd_rate_ms,
+            },
+        }
+        for rc in _nspd_default_regions()
+    }
+)
+
 
 @worker_ready.connect
 def _resume_zombie_runs(sender=None, **_kwargs) -> None:
@@ -122,3 +147,32 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None:
             logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid)
         except Exception as e:
             logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e)
+
+    # NSPD-runs: помечаем зомби (>15 мин без heartbeat — NSPD сам по себе
+    # медленнее kn, дольше дельта). Resume не делаем — следующий beat
+    # подберёт pending (idempotent через cad_quarters_geom).
+    db = SessionLocal()
+    try:
+        db.execute(
+            text(
+                """
+                UPDATE nspd_scrape_runs
+                   SET status = 'zombie',
+                       finished_at = NOW(),
+                       error = COALESCE(error,
+                           'auto-zombie at worker_ready')
+                 WHERE status = 'running'
+                   AND COALESCE(heartbeat_at, started_at)
+                       < NOW() - INTERVAL '15 minutes'
+                """
+            )
+        )
+        db.commit()
+    except Exception as e:
+        logger.warning("worker_ready nspd zombie scan failed: %s", e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+    finally:
+        db.close()

backend/app/workers/tasks/scrape_nspd.py

@@ -0,0 +1,104 @@
+"""Celery task wrapper для NSPD-скрейпа кадастровых кварталов."""
+
+from __future__ import annotations
+
+import logging
+from contextlib import contextmanager
+from typing import Any
+
+import redis
+
+from app.core.config import settings
+from app.services.scrapers.nspd_kn import WafBlockedError, run_region_scrape
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# NSPD-скрейпы 700+ cad-кварталов на регион занимают ~2 часа (600мс/req × 2 req/cad).
+# Lock TTL 3 часа — потолок с запасом, не оставляет зомби-lock на долго.
+_LOCK_TTL_SECONDS = 3 * 60 * 60
+
+
+def _lock_key(region_code: int) -> str:
+    return f"scrape:nspd:lock:{region_code}"
+
+
+def force_release_lock(region_code: int) -> bool:
+    """Принудительно удалить Redis-lock для emergency manual trigger."""
+    key = _lock_key(region_code)
+    r = redis.Redis.from_url(settings.redis_url)
+    try:
+        return bool(r.delete(key))
+    except Exception as e:
+        logger.warning("force_release_lock %s failed: %s", key, e)
+        return False
+
+
+@contextmanager
+def _region_lock(region_code: int):
+    key = _lock_key(region_code)
+    r = redis.Redis.from_url(settings.redis_url)
+    acquired = r.set(key, "1", nx=True, ex=_LOCK_TTL_SECONDS)
+    try:
+        yield bool(acquired)
+    finally:
+        if acquired:
+            try:
+                r.delete(key)
+            except Exception:
+                pass
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.scrape_nspd.scrape_nspd_region",
+    max_retries=2,
+    autoretry_for=(WafBlockedError,),
+    retry_backoff=600,  # 10 мин начальный, удваивается
+    retry_backoff_max=7200,  # потолок 2 часа
+)
+def scrape_nspd_region(
+    self: Any,
+    region_code: int = 66,
+    triggered_by: str = "beat",
+    limit: int | None = None,
+    rate_ms: int = 600,
+) -> dict[str, Any]:
+    """Запуск NSPD-скрейпа одного региона. По расписанию: 20-е число
+    февраля/мая/августа/ноября (после квартальных публикаций rosreestr).
+
+    Args:
+        region_code: код региона rosreestr (66 = Свердл).
+        triggered_by: 'beat' | 'manual' | 'resume' — для журналирования.
+        limit: ограничить N cad-кварталов (smoke-тест).
+        rate_ms: пауза между запросами (по умолчанию 600мс — наблюдаемый предел WAF).
+
+    Returns:
+        dict с run_id и итогами (pending/ok/failed/buildings/requests/elapsed_s),
+        либо {'skipped': True, 'reason': 'lock_held'} если уже идёт прогон.
+    """
+    with _region_lock(region_code) as got_lock:
+        if not got_lock:
+            logger.warning(
+                "scrape_nspd_region SKIPPED region=%s — другой sweep идёт (Redis lock)",
+                region_code,
+            )
+            return {
+                "skipped": True,
+                "reason": "lock_held",
+                "region_code": region_code,
+                "lock_key": _lock_key(region_code),
+            }
+
+        logger.info(
+            "scrape_nspd_region start region=%s trigger=%s limit=%s",
+            region_code,
+            triggered_by,
+            limit,
+        )
+        return run_region_scrape(
+            region_code=region_code,
+            triggered_by=triggered_by,
+            limit=limit,
+            rate_ms=rate_ms,
+        )

data/sql/58_fetch_cad_quarters_nspd.py

@@ -0,0 +1,281 @@
+"""Fetch кадастровые квартал-полигоны из NSPD и загрузить в PostGIS.
+
+Usage:
+    cd backend && uv run python ../data/sql/58_fetch_cad_quarters_nspd.py [options]
+
+Options:
+    --scope ekb|region   ekb (default): только cad-кварталы с ДДУ-сделками
+                         в ЕКБ (~674 кварталов). region: вся Свердл (~10 853).
+    --batch-size N       commit каждые N кварталов (default 25).
+    --rate-limit-ms N    пауза между запросами в мс (default 700).
+    --resume             пропустить cad-кварталы уже в cad_quarters_geom.
+    --max N              ограничить N запросами (для smoke-теста).
+    --dry-run            fetch+parse only, skip DB upsert.
+
+ВАЖНО: NSPD блокирует запросы с не-российских IP (WAF). Скрипт нужно запускать
+с машины в РФ. Если получаешь HTTP 403 + 'Client IP: ...' — это блок гео.
+
+Endpoint: GET https://nspd.gov.ru/api/geoportal/v2/search/geoportal
+          ?thematicSearchId=1&query={cad_number}
+Returns: GeoJSON FeatureCollection. Геометрия MultiPolygon в EPSG:3857 (web
+         mercator). Скрипт переводит в EPSG:4326 через ST_Transform на стороне
+         БД при INSERT.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+# Add backend to path so we can import app.* in standalone CLI mode.
+ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(ROOT / "backend"))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger(__name__)
+
+NSPD_URL = "https://nspd.gov.ru/api/geoportal/v2/search/geoportal"
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+    "Accept-Language": "ru-RU,ru;q=0.9",
+    "Referer": "https://nspd.gov.ru/map",
+}
+
+
+def fetch_cad_quarter(cad_number: str, *, retries: int = 3, timeout: int = 30) -> dict | None:
+    """Запрос NSPD API для одного cad-номера.
+
+    Returns: GeoJSON FeatureCollection or None если квартал не найден / ошибка.
+    """
+    qs = urllib.parse.urlencode(
+        {
+            "thematicSearchId": 1,  # 1 = кадастровые квартал
+            "query": cad_number,
+        }
+    )
+    url = f"{NSPD_URL}?{qs}"
+    req = urllib.request.Request(url, headers=HEADERS)
+    last_err: Exception | None = None
+    for attempt in range(retries):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return json.loads(resp.read().decode("utf-8"))
+        except urllib.error.HTTPError as e:
+            if e.code == 403:
+                # WAF: гео-блок не лечится retry. Бросаем выше — оператор решит.
+                body = e.read().decode("utf-8", errors="ignore")[:300]
+                log.error("HTTP 403 from NSPD (likely geo-WAF). Body: %s", body)
+                raise
+            if e.code == 404:
+                return None  # cad-квартал не найден — норма для глухих номеров.
+            last_err = e
+            log.warning("HTTP %d for %s (attempt %d/%d)", e.code, cad_number, attempt + 1, retries)
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            last_err = e
+            log.warning("Network err for %s: %s (attempt %d/%d)", cad_number, e, attempt + 1, retries)
+        time.sleep(2 ** attempt)
+    log.error("Failed %s after %d retries: %s", cad_number, retries, last_err)
+    return None
+
+
+def extract_geometry(geojson: dict | None, cad_number: str) -> tuple[dict | None, dict | None]:
+    """Из GeoJSON выбрать первый MultiPolygon (или Polygon) под точный
+    cad-номер. Returns (geometry_dict, raw_properties)."""
+    if not geojson or "features" not in geojson:
+        return None, None
+    for feat in geojson["features"]:
+        props = feat.get("properties") or {}
+        # NSPD возвращает кадастровый номер в одном из этих полей.
+        cn = (
+            props.get("cn")
+            or props.get("cadastralNumber")
+            or props.get("cad_num")
+            or props.get("CAD_NUMBER")
+        )
+        if cn and str(cn).strip() == cad_number:
+            geom = feat.get("geometry")
+            if not geom:
+                continue
+            # Принимаем Polygon → MultiPolygon (для unification).
+            if geom.get("type") == "Polygon":
+                geom = {"type": "MultiPolygon", "coordinates": [geom["coordinates"]]}
+            return geom, props
+    # Fallback: если features=1 и cn совпадает по prefix — берём первую.
+    if len(geojson["features"]) == 1:
+        feat = geojson["features"][0]
+        geom = feat.get("geometry")
+        if geom:
+            if geom.get("type") == "Polygon":
+                geom = {"type": "MultiPolygon", "coordinates": [geom["coordinates"]]}
+            return geom, feat.get("properties") or {}
+    return None, None
+
+
+def candidates_to_fetch(scope: str, resume: bool) -> list[str]:
+    """Возвращает список cad-номеров для скрейпа из rosreestr_deals."""
+    from sqlalchemy import text
+
+    from app.core.db import SessionLocal
+
+    db = SessionLocal()
+    try:
+        if scope == "ekb":
+            rows = db.execute(
+                text(
+                    """
+                    SELECT DISTINCT quarter_cad_number
+                      FROM rosreestr_deals
+                     WHERE region_code = 66
+                       AND doc_type = 'ДДУ'
+                       AND realestate_type_code = '002001003000'
+                       AND (district ILIKE '%Екатеринбург%' OR city ILIKE '%Екатеринбург%')
+                       AND quarter_cad_number IS NOT NULL
+                       AND quarter_cad_number <> ''
+                    """
+                )
+            ).all()
+        else:  # region
+            rows = db.execute(
+                text(
+                    """
+                    SELECT DISTINCT quarter_cad_number
+                      FROM rosreestr_deals
+                     WHERE region_code = 66
+                       AND quarter_cad_number IS NOT NULL
+                       AND quarter_cad_number <> ''
+                    """
+                )
+            ).all()
+        cads = sorted({r[0] for r in rows if r[0]})
+
+        if resume:
+            done = {
+                r[0]
+                for r in db.execute(
+                    text("SELECT cad_number FROM cad_quarters_geom")
+                ).all()
+            }
+            cads = [c for c in cads if c not in done]
+            log.info("Resume: skipping %d already-loaded quarters", len(done))
+
+        return cads
+    finally:
+        db.close()
+
+
+def upsert_quarter(db, cad_number: str, geom_dict: dict, props: dict) -> None:
+    """INSERT или UPDATE в cad_quarters_geom. Полигон приходит в EPSG:3857
+    (web mercator); конвертируем в 4326 на стороне БД через ST_Transform."""
+    from sqlalchemy import text
+
+    db.execute(
+        text(
+            """
+            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at, source)
+            VALUES (
+                :cn,
+                ST_Multi(
+                    ST_Transform(
+                        ST_SetSRID(ST_GeomFromGeoJSON(:gj), 3857),
+                        4326
+                    )
+                )::geometry(MultiPolygon, 4326),
+                CAST(:props AS jsonb),
+                NOW(),
+                'nspd'
+            )
+            ON CONFLICT (cad_number) DO UPDATE SET
+                geom = EXCLUDED.geom,
+                raw_props = EXCLUDED.raw_props,
+                fetched_at = NOW()
+            """
+        ),
+        {
+            "cn": cad_number,
+            "gj": json.dumps(geom_dict),
+            "props": json.dumps(props, ensure_ascii=False),
+        },
+    )
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description="Fetch+load NSPD cad-quarter polygons")
+    p.add_argument("--scope", choices=["ekb", "region"], default="ekb")
+    p.add_argument("--batch-size", type=int, default=25)
+    p.add_argument("--rate-limit-ms", type=int, default=700)
+    p.add_argument("--resume", action="store_true", help="Skip already-loaded cad-numbers")
+    p.add_argument("--max", type=int, default=None, help="Limit N requests (smoke test)")
+    p.add_argument("--dry-run", action="store_true")
+    args = p.parse_args()
+
+    cads = candidates_to_fetch(args.scope, resume=args.resume)
+    if args.max:
+        cads = cads[: args.max]
+    log.info("Plan to fetch %d cad-quarters (scope=%s)", len(cads), args.scope)
+    if not cads:
+        log.info("Nothing to do.")
+        return 0
+
+    if args.dry_run:
+        log.info("Dry-run: would fetch %s", cads[:5])
+        return 0
+
+    from sqlalchemy import text  # noqa: F401
+
+    from app.core.db import SessionLocal
+
+    db = SessionLocal()
+    ok = 0
+    skipped = 0
+    failed = 0
+    try:
+        for i, cn in enumerate(cads, 1):
+            try:
+                gj = fetch_cad_quarter(cn)
+                geom, props = extract_geometry(gj, cn)
+                if geom is None:
+                    log.warning("[%d/%d] %s — no geometry returned", i, len(cads), cn)
+                    skipped += 1
+                else:
+                    upsert_quarter(db, cn, geom, props or {})
+                    ok += 1
+                    if i % 10 == 0 or i == len(cads):
+                        log.info("[%d/%d] %s ✓ (ok=%d skip=%d fail=%d)",
+                                 i, len(cads), cn, ok, skipped, failed)
+            except urllib.error.HTTPError as e:
+                if e.code == 403:
+                    db.commit()
+                    log.error(
+                        "Aborting: NSPD WAF блокирует запросы (HTTP 403). "
+                        "Скрипт нужно запускать с российского IP. Saved %d so far.",
+                        ok,
+                    )
+                    return 2
+                failed += 1
+            except Exception as e:  # noqa: BLE001
+                failed += 1
+                log.exception("[%d/%d] %s ✗ %s", i, len(cads), cn, e)
+            if i % args.batch_size == 0:
+                db.commit()
+                log.info("  → committed batch (i=%d, ok=%d)", i, ok)
+            time.sleep(args.rate_limit_ms / 1000.0)
+        db.commit()
+        log.info("Done: ok=%d, skipped=%d, failed=%d", ok, skipped, failed)
+        return 0
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data/sql/58_schema_cad_quarters.sql

@@ -0,0 +1,37 @@
+-- Cad-quarter polygons (NSPD) + obj_id ↔ cad_quarter mapping.
+--
+-- Purpose: связать domrf_kn_objects (lat/lon) и rosreestr_deals (quarter_cad_number)
+-- через PostGIS spatial-join. Это даёт per-ЖК аппроксимацию rosreestr-сделок
+-- (распределение пропорционально flat_count внутри квартала).
+--
+-- Source: NSPD nspd.gov.ru thematicSearchId=1 (кадастровые кварталы).
+-- Coverage: ~674 кварталов в ЕКБ с ДДУ-сделками на квартиры (см. 58_fetch_cad_quarters_nspd.py).
+
+CREATE EXTENSION IF NOT EXISTS postgis;
+
+CREATE TABLE IF NOT EXISTS cad_quarters_geom (
+    cad_number    TEXT PRIMARY KEY,
+    geom          geometry(MultiPolygon, 4326) NOT NULL,
+    raw_props     jsonb,
+    fetched_at    timestamptz NOT NULL DEFAULT NOW(),
+    source        TEXT NOT NULL DEFAULT 'nspd'
+);
+
+CREATE INDEX IF NOT EXISTS cad_quarters_geom_gist
+    ON cad_quarters_geom USING GIST (geom);
+
+-- На domrf_kn_objects добавляем cad_quarter (заполняется backfill-скриптом
+-- 59_backfill_obj_cad_quarter.sql через ST_Contains по lat/lon).
+ALTER TABLE domrf_kn_objects
+    ADD COLUMN IF NOT EXISTS cad_quarter TEXT;
+
+CREATE INDEX IF NOT EXISTS domrf_kn_objects_cad_quarter_idx
+    ON domrf_kn_objects(cad_quarter);
+
+COMMENT ON TABLE cad_quarters_geom IS
+    'Кадастровые кварталы РФ (полигоны NSPD). Используется для spatial-join '
+    'rosreestr_deals (по quarter_cad_number) → domrf_kn_objects (по lat/lon → ST_Contains).';
+
+COMMENT ON COLUMN domrf_kn_objects.cad_quarter IS
+    'Кадастровый квартал, в который попадают координаты ЖК. Заполняется '
+    '59_backfill_obj_cad_quarter.sql через ST_Contains(cad_quarters_geom.geom, lat/lon).';

data/sql/59_backfill_obj_cad_quarter.sql

@@ -0,0 +1,27 @@
+-- Backfill domrf_kn_objects.cad_quarter через ST_Contains spatial-join.
+--
+-- Логика: каждый ЖК с lat/lon → найти cad-квартал, в полигон которого попадает
+-- точка. Если ЖК на границе двух кварталов — берём первый (ST_Contains
+-- однозначен, ST_Intersects дал бы пересечения).
+--
+-- Запускать ПОСЛЕ 58_fetch_cad_quarters_nspd.py.
+
+UPDATE domrf_kn_objects o
+   SET cad_quarter = cq.cad_number
+  FROM cad_quarters_geom cq
+ WHERE o.region_cd = 66
+   AND o.latitude IS NOT NULL
+   AND o.longitude IS NOT NULL
+   AND ST_Contains(
+           cq.geom,
+           ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)
+       )
+   AND (o.cad_quarter IS NULL OR o.cad_quarter <> cq.cad_number);
+
+-- Контрольные цифры: сколько ЖК привязано / без привязки.
+SELECT
+    COUNT(*) FILTER (WHERE cad_quarter IS NOT NULL) AS with_cad,
+    COUNT(*) FILTER (WHERE cad_quarter IS NULL)     AS without_cad,
+    COUNT(DISTINCT cad_quarter)                     AS unique_quarters_used
+  FROM domrf_kn_objects
+ WHERE region_cd = 66 AND district_name IS NOT NULL;

data/sql/60_v_zk_rosreestr_velocity.sql

@@ -0,0 +1,83 @@
+-- View: per-ЖК velocity и средняя цена через rosreestr-сделки в cad-квартале,
+-- распределённые пропорционально flat_count.
+--
+-- Логика: rosreestr_deals не имеет obj_id, но имеет quarter_cad_number.
+-- domrf_kn_objects.cad_quarter заполнен через spatial-join (см. 59).
+-- Для каждого квартала: суммируем сделки за 12 мес, распределяем по ЖК
+-- внутри квартала пропорционально их flat_count.
+--
+-- estimated_velocity_pm — оценка скорости продаж конкретного ЖК (кв/мес).
+-- estimated_price_th_per_m2 — медианная цена сделок в квартале (тыс. ₽/м²).
+--
+-- ВАЖНО: это АППРОКСИМАЦИЯ. Если в одном квартале несколько ЖК, доли точные
+-- только когда они продают равномерно. Для крупных одиночных ЖК (квартал
+-- содержит только их) — точно.
+
+CREATE OR REPLACE VIEW v_zk_rosreestr_velocity AS
+WITH zk_pool AS (
+    -- Берём последний snapshot каждого активного ЖК.
+    SELECT DISTINCT ON (o.obj_id)
+           o.obj_id,
+           o.comm_name,
+           o.dev_name,
+           o.district_name,
+           o.flat_count,
+           o.cad_quarter,
+           o.obj_class
+      FROM domrf_kn_objects o
+     WHERE o.region_cd = 66
+       AND o.cad_quarter IS NOT NULL
+       AND o.flat_count IS NOT NULL
+       AND o.flat_count > 0
+       AND o.site_status = 'Строящиеся'
+     ORDER BY o.obj_id, o.snapshot_date DESC
+),
+quarter_totals AS (
+    -- Сумма flat_count активных ЖК в каждом квартале (denominator для split).
+    SELECT cad_quarter, SUM(flat_count)::numeric AS total_flats
+      FROM zk_pool
+     GROUP BY cad_quarter
+),
+quarter_deals AS (
+    -- Rosreestr-сделки за 12 мес по каждому кварталу.
+    SELECT quarter_cad_number AS cad_quarter,
+           COUNT(*)                                            AS deals_12mo,
+           AVG(price_per_sqm)                                  AS avg_price_pm,
+           PERCENTILE_CONT(0.5) WITHIN GROUP
+               (ORDER BY price_per_sqm)                        AS median_price_pm,
+           AVG(area)                                           AS avg_area_sqm
+      FROM rosreestr_deals
+     WHERE region_code = 66
+       AND doc_type = 'ДДУ'
+       AND realestate_type_code = '002001003000'
+       AND area > 10 AND area <= 200
+       AND price_per_sqm BETWEEN 30000 AND 1000000
+       AND period_start_date >= NOW() - INTERVAL '12 months'
+     GROUP BY quarter_cad_number
+)
+SELECT
+    z.obj_id,
+    z.comm_name,
+    z.dev_name,
+    z.district_name,
+    z.obj_class,
+    z.flat_count,
+    z.cad_quarter,
+    qt.total_flats                                              AS quarter_total_flats,
+    qd.deals_12mo                                               AS quarter_deals_12mo,
+    -- estimated deals = quarter_deals × my_flats / quarter_total_flats
+    ROUND(qd.deals_12mo * z.flat_count::numeric / qt.total_flats, 1)
+                                                                AS estimated_deals_12mo,
+    ROUND(qd.deals_12mo * z.flat_count::numeric / qt.total_flats / 12.0, 2)
+                                                                AS estimated_velocity_pm,
+    ROUND(qd.median_price_pm / 1000.0, 1)                       AS median_price_th_per_m2,
+    ROUND(qd.avg_area_sqm, 1)                                   AS avg_deal_area_sqm
+  FROM zk_pool z
+  JOIN quarter_totals qt ON qt.cad_quarter = z.cad_quarter
+  JOIN quarter_deals qd ON qd.cad_quarter = z.cad_quarter
+;
+
+COMMENT ON VIEW v_zk_rosreestr_velocity IS
+    'Per-ЖК аппроксимация скорости продаж и медианной цены через rosreestr_deals '
+    'квартала, распределённые по ЖК пропорционально flat_count. Источник для '
+    'cross-validation domrf_kn_sale_graph и обнаружения ЖК-аутлайеров.';

data/sql/61_import_nspd_batch.py

@@ -0,0 +1,208 @@
+"""Импорт NSPD-сырого батча (cad_quarters + cad_buildings) из JSONL в Postgres.
+
+Поддерживает 2 формата JSONL:
+- v1 (старый): {cad, q_wkt, buildings: [{cn, p, n, a, ar, f, yb, yc, c, rd, w}]}
+- v2 (новый): {cad, q_wkt, q_props, buildings: [{cn, w, props}]} — props это полный
+  options-объект из NSPD (сохраняется как raw_props в БД).
+
+Подключение через SSH-тунель:
+    ssh -N -L 15432:localhost:5432 gendesign
+    psql порт: localhost:15432, db gendesign, user gendesign, pwd <см .env>
+
+Usage:
+    cd backend && uv run python ../data/sql/61_import_nspd_batch.py <path/to/jsonl>
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+import psycopg
+
+PG_DSN = os.environ.get(
+    "TUNNEL_DSN",
+    "postgresql://gendesign:2J2SBPMKuS998fiwhtQqDhMI@localhost:15432/gendesign",
+)
+
+
+def _to_int(x):
+    if x is None or x == "":
+        return None
+    try:
+        return int(x)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_num(x):
+    if x is None or x == "":
+        return None
+    try:
+        return float(x)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_date(x):
+    if not x:
+        return None
+    s = str(x)
+    return s[:10] if len(s) >= 10 else None
+
+
+def insert_quarters(cur, quarters: list[dict]) -> int:
+    inserted = 0
+    for q in quarters:
+        if not q.get("q_wkt"):
+            continue
+        raw = q.get("q_props")  # full options dict from NSPD (v2)
+        cur.execute(
+            """
+            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, source)
+            VALUES (
+                %s,
+                ST_Multi(ST_Transform(
+                    ST_SetSRID(ST_GeomFromText(%s), 3857),
+                    4326
+                ))::geometry(MultiPolygon, 4326),
+                CAST(%s AS jsonb),
+                'nspd'
+            )
+            ON CONFLICT (cad_number) DO UPDATE SET
+                geom = EXCLUDED.geom,
+                raw_props = COALESCE(EXCLUDED.raw_props, cad_quarters_geom.raw_props),
+                fetched_at = NOW()
+            """,
+            (
+                q["cad"],
+                q["q_wkt"],
+                json.dumps(raw, ensure_ascii=False) if raw else None,
+            ),
+        )
+        inserted += 1
+    return inserted
+
+
+def insert_buildings(cur, buildings: list[tuple[dict, str]]) -> int:
+    """buildings: [(building_dict, parent_quarter_cad), ...]"""
+    inserted = 0
+    for b, qcad in buildings:
+        if not b.get("w"):
+            continue
+        # v2 has full props; v1 has cherry-picked top-level keys
+        props = b.get("props") or {}
+        # Prefer v2 (props), fallback to v1 (top-level)
+        cad_num = props.get("cad_num") or b.get("cn")
+        purpose = props.get("purpose") or b.get("p")
+        name = props.get("building_name") or b.get("n")
+        addr = props.get("readable_address") or b.get("a")
+        area = _to_num(props.get("area") if props else b.get("ar"))
+        floors = props.get("floors") or b.get("f")
+        year_built = _to_int(props.get("year_built") if props else b.get("yb"))
+        year_comm = _to_int(props.get("year_commisioning") if props else b.get("yc"))
+        cost = _to_num(props.get("cost_value") if props else b.get("c"))
+        reg_date = _to_date(props.get("registration_date") if props else b.get("rd"))
+        # New v2-only fields
+        status = props.get("status")
+        ownership = props.get("ownership_type")
+        cultural = props.get("cultural_heritage_object") or props.get("cultural_heritage_val")
+        if cultural is not None and not isinstance(cultural, str):
+            cultural = json.dumps(cultural, ensure_ascii=False)
+        underground = props.get("underground_floors")
+        build_rec_area = _to_num(props.get("build_record_area"))
+        build_rec_type = props.get("build_record_type_value")
+        common_status = props.get("common_data_status")
+        obj_type = props.get("type")
+        raw_props_json = json.dumps(props, ensure_ascii=False) if props else None
+
+        cur.execute(
+            """
+            INSERT INTO cad_buildings (
+                cad_num, quarter_cad_num, geom, purpose, building_name,
+                readable_address, area, floors, year_built, year_commisioning,
+                cost_value, registration_date,
+                status, ownership_type, cultural_heritage, underground_floors,
+                build_record_area, build_record_type, common_data_status, obj_type,
+                raw_props
+            )
+            VALUES (
+                %s, %s,
+                ST_Transform(ST_SetSRID(ST_GeomFromText(%s), 3857), 4326),
+                %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s, %s, %s, %s,
+                CAST(%s AS jsonb)
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                cost_value = EXCLUDED.cost_value,
+                area = EXCLUDED.area,
+                year_built = EXCLUDED.year_built,
+                year_commisioning = EXCLUDED.year_commisioning,
+                status = EXCLUDED.status,
+                ownership_type = EXCLUDED.ownership_type,
+                cultural_heritage = EXCLUDED.cultural_heritage,
+                underground_floors = EXCLUDED.underground_floors,
+                build_record_area = EXCLUDED.build_record_area,
+                build_record_type = EXCLUDED.build_record_type,
+                common_data_status = EXCLUDED.common_data_status,
+                obj_type = EXCLUDED.obj_type,
+                raw_props = COALESCE(EXCLUDED.raw_props, cad_buildings.raw_props),
+                fetched_at = NOW()
+            """,
+            (
+                cad_num, qcad, b["w"],
+                purpose, name, addr,
+                area, floors,
+                year_built, year_comm,
+                cost, reg_date,
+                status, ownership, cultural, underground,
+                build_rec_area, build_rec_type, common_status, obj_type,
+                raw_props_json,
+            ),
+        )
+        inserted += 1
+    return inserted
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        print("usage: 61_import_nspd_batch.py <jsonl_path>")
+        return 1
+    fp = Path(sys.argv[1])
+    if not fp.exists():
+        print(f"file not found: {fp}")
+        return 1
+
+    quarters: list[dict] = []
+    buildings: list[tuple[dict, str]] = []
+    with fp.open(encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            q = json.loads(line)
+            quarters.append(q)
+            for b in q.get("buildings", []):
+                buildings.append((b, q["cad"]))
+
+    print(f"Parsed: {len(quarters)} quarters, {len(buildings)} buildings")
+    conn = psycopg.connect(PG_DSN, connect_timeout=10)
+    try:
+        cur = conn.cursor()
+        n_q = insert_quarters(cur, quarters)
+        n_b = insert_buildings(cur, buildings)
+        conn.commit()
+        print(f"Inserted: {n_q} quarters, {n_b} buildings")
+        cur.execute("SELECT COUNT(*) FROM cad_quarters_geom")
+        print(f"DB total cad_quarters_geom: {cur.fetchone()[0]}")
+        cur.execute("SELECT COUNT(*) FROM cad_buildings")
+        print(f"DB total cad_buildings: {cur.fetchone()[0]}")
+    finally:
+        conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data/sql/62_scrape_nspd_full.py

@@ -0,0 +1,250 @@
+"""Полный NSPD scrape для всех cad-кварталов ЕКБ с ДДУ-сделками.
+
+Запускать с машины пользователя (РФ-IP, иначе WAF).
+SSH-tunnel к prod БД на localhost:15432.
+
+Что делает:
+1. Берёт список cad-номеров из rosreestr_deals (region=66, ЕКБ, ДДУ, квартиры).
+2. Пропускает уже скрейпнутые (есть в cad_quarters_geom).
+3. Для каждого cad-квартала:
+   a. fetch /api/geoportal/v2/search/geoportal?thematicSearchId=2 → polygon
+   b. fetch /api/geoportal/v2/search/geoportal?thematicSearchId=1 → 20 зданий + земля
+4. INSERT в cad_quarters_geom + cad_buildings (фильтр cad_num LIKE '66:41:%').
+5. После каждого квартала commit + лог. Можно прервать и перезапустить.
+
+Usage:
+    cd backend && uv run python ../data/sql/62_scrape_nspd_full.py [--limit N] [--rate-ms 600]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import ssl
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any
+
+import psycopg
+
+PG_DSN = os.environ.get(
+    "TUNNEL_DSN",
+    "postgresql://gendesign:2J2SBPMKuS998fiwhtQqDhMI@localhost:15432/gendesign",
+)
+
+NSPD_BASE = "https://nspd.gov.ru/api/geoportal/v2/search/geoportal"
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+    "Accept-Language": "ru-RU,ru;q=0.9",
+    "Referer": "https://nspd.gov.ru/map",
+}
+SSL_CTX = ssl._create_unverified_context()
+
+
+def nspd_fetch(thematic_search_id: int, query: str, *, retries: int = 5, timeout: int = 30) -> dict | None:
+    """Fetch NSPD with backoff. WAF banит burst-запросы — на 403 ждём долго."""
+    qs = urllib.parse.urlencode({"thematicSearchId": thematic_search_id, "query": query})
+    url = f"{NSPD_BASE}?{qs}"
+    req = urllib.request.Request(url, headers=HEADERS)
+    last_err = None
+    for attempt in range(retries):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r:
+                return json.loads(r.read().decode("utf-8"))
+        except urllib.error.HTTPError as e:
+            if e.code == 404:
+                return None
+            last_err = e
+            if e.code == 403:
+                # WAF rate-limit — ждём дольше (60 сек)
+                wait = 60 + 30 * attempt
+                print(f"  WAF 403 for {query} (attempt {attempt+1}/{retries}) — sleeping {wait}s", flush=True)
+                time.sleep(wait)
+                continue
+            print(f"  HTTP {e.code} for {query} (attempt {attempt+1})", flush=True)
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            last_err = e
+            print(f"  Network err for {query}: {e} (attempt {attempt+1})", flush=True)
+        time.sleep(min(2 ** attempt, 30))
+    print(f"  FAILED {query}: {last_err}", flush=True)
+    return None
+
+
+def poly_to_wkt(geom: dict | None) -> str | None:
+    if not geom:
+        return None
+    def ring(r):
+        return "(" + ",".join(f"{p[0]} {p[1]}" for p in r) + ")"
+    t = geom.get("type")
+    if t == "Polygon":
+        return "POLYGON(" + ",".join(ring(r) for r in geom["coordinates"]) + ")"
+    if t == "MultiPolygon":
+        return "MULTIPOLYGON(" + ",".join("(" + ",".join(ring(r) for r in p) + ")" for p in geom["coordinates"]) + ")"
+    if t == "Point":
+        return f"POINT({geom['coordinates'][0]} {geom['coordinates'][1]})"
+    return None
+
+
+def get_pending_cads(cur) -> list[str]:
+    """Все cad-номера с ДДУ-сделками в ЕКБ — минус уже scraped."""
+    cur.execute("""
+        SELECT DISTINCT quarter_cad_number
+          FROM rosreestr_deals
+         WHERE region_code = 66
+           AND doc_type = 'ДДУ'
+           AND realestate_type_code = '002001003000'
+           AND (district ILIKE '%Екатеринбург%' OR city ILIKE '%Екатеринбург%')
+           AND quarter_cad_number IS NOT NULL
+           AND quarter_cad_number <> ''
+           AND quarter_cad_number <> '66:41:0000000'
+    """)
+    all_cads = {r[0] for r in cur.fetchall()}
+    cur.execute("SELECT cad_number FROM cad_quarters_geom")
+    done = {r[0] for r in cur.fetchall()}
+    pending = sorted(all_cads - done)
+    return pending
+
+
+def insert_quarter(cur, cad: str, wkt: str) -> None:
+    cur.execute(
+        """
+        INSERT INTO cad_quarters_geom (cad_number, geom, source)
+        VALUES (
+            %s,
+            ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromText(%s), 3857), 4326))::geometry(MultiPolygon, 4326),
+            'nspd'
+        )
+        ON CONFLICT (cad_number) DO UPDATE SET geom = EXCLUDED.geom, fetched_at = NOW()
+        """,
+        (cad, wkt),
+    )
+
+
+def insert_buildings(cur, qcad: str, features: list[dict]) -> int:
+    """Insert 'Здания' filtered by cad LIKE '66:41:%'."""
+    n = 0
+    for f in features:
+        if f.get("properties", {}).get("categoryName") != "Здания":
+            continue
+        opts = f["properties"].get("options", {}) or {}
+        cn = opts.get("cad_num")
+        if not cn or not cn.startswith("66:41:"):
+            continue
+        wkt = poly_to_wkt(f.get("geometry"))
+        if not wkt:
+            continue
+        cur.execute(
+            """
+            INSERT INTO cad_buildings (
+                cad_num, quarter_cad_num, geom, purpose, building_name,
+                readable_address, area, floors, year_built, year_commisioning,
+                cost_value, registration_date
+            )
+            VALUES (
+                %s, %s, ST_Transform(ST_SetSRID(ST_GeomFromText(%s), 3857), 4326),
+                %s, %s, %s, %s, %s, %s, %s, %s, %s
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                cost_value = EXCLUDED.cost_value,
+                area = EXCLUDED.area,
+                fetched_at = NOW()
+            """,
+            (
+                cn, qcad, wkt,
+                opts.get("purpose"), opts.get("building_name"),
+                opts.get("readable_address"), opts.get("area"), opts.get("floors"),
+                int(opts["year_built"]) if opts.get("year_built") else None,
+                int(opts["year_commisioning"]) if opts.get("year_commisioning") else None,
+                opts.get("cost_value"),
+                opts.get("registration_date", "").split(" ")[0] if opts.get("registration_date") else None,
+            ),
+        )
+        n += 1
+    return n
+
+
+def main() -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--limit", type=int, default=None, help="Limit N quarters (smoke)")
+    p.add_argument("--rate-ms", type=int, default=600, help="Pause between cads (ms)")
+    p.add_argument("--commit-every", type=int, default=10, help="Commit every N quarters")
+    args = p.parse_args()
+
+    conn = psycopg.connect(PG_DSN, connect_timeout=10)
+    cur = conn.cursor()
+    pending = get_pending_cads(cur)
+    if args.limit:
+        pending = pending[: args.limit]
+    print(f"Pending: {len(pending)} cad-quarters")
+    if not pending:
+        print("Nothing to do.")
+        return 0
+
+    started = time.time()
+    ok = 0
+    n_buildings = 0
+    failed = 0
+    for i, cn in enumerate(pending, 1):
+        try:
+            j2 = nspd_fetch(2, cn)
+            qf = None
+            if j2:
+                features = j2.get("data", {}).get("features") or []
+                for f in features:
+                    if f.get("properties", {}).get("label") == cn:
+                        qf = f
+                        break
+            wkt = poly_to_wkt(qf.get("geometry") if qf else None) if qf else None
+            if wkt:
+                insert_quarter(cur, cn, wkt)
+
+            j1 = nspd_fetch(1, cn)
+            if j1:
+                features = j1.get("data", {}).get("features") or []
+                n_buildings += insert_buildings(cur, cn, features)
+
+            if wkt:
+                ok += 1
+            else:
+                failed += 1
+
+            if i % args.commit_every == 0:
+                conn.commit()
+                elapsed = time.time() - started
+                rate = ok / elapsed * 60 if elapsed > 0 else 0
+                eta = (len(pending) - i) / max(rate, 1) if rate > 0 else 0
+                print(
+                    f"[{i}/{len(pending)}] {cn} "
+                    f"ok={ok} fail={failed} buildings={n_buildings} "
+                    f"rate={rate:.1f}/min eta={eta:.0f}min",
+                    flush=True,
+                )
+        except Exception as e:  # noqa: BLE001
+            failed += 1
+            conn.rollback()
+            print(f"  ERR {cn}: {e}", flush=True)
+
+        time.sleep(args.rate_ms / 1000.0)
+
+    conn.commit()
+    elapsed = time.time() - started
+    cur.execute("SELECT COUNT(*) FROM cad_quarters_geom")
+    db_q = cur.fetchone()[0]
+    cur.execute("SELECT COUNT(*) FROM cad_buildings")
+    db_b = cur.fetchone()[0]
+    print(f"\nDONE in {elapsed:.0f}s. ok={ok} failed={failed} buildings_inserted={n_buildings}")
+    print(f"DB totals: cad_quarters_geom={db_q}, cad_buildings={db_b}")
+    conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data/sql/63_schema_nspd_runs.sql

@@ -0,0 +1,94 @@
+-- NSPD scrape industrialization: schema baseline.
+--
+-- 1. cad_buildings — формальная DDL для таблицы, которая создавалась
+--    напрямую в проде через 61_import_nspd_batch.py (10590 зданий ЕКБ
+--    на момент 2026-04-30). IF NOT EXISTS — миграция идемпотентна.
+-- 2. nspd_scrape_runs — журнал автоматизированных Celery-запусков
+--    (по образцу kn_scrape_runs из 50_schema_kn_extensions.sql).
+-- 3. nspd_scrape_log — построчный лог стадий (по образцу kn_scrape_log).
+
+CREATE EXTENSION IF NOT EXISTS postgis;
+
+-- ── 1. cad_buildings ─────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS cad_buildings (
+    cad_num             TEXT PRIMARY KEY,
+    quarter_cad_num     TEXT REFERENCES cad_quarters_geom(cad_number) ON DELETE SET NULL,
+    geom                geometry(Geometry, 4326),
+    purpose             TEXT,
+    building_name       TEXT,
+    readable_address    TEXT,
+    area                NUMERIC,
+    floors              INT,
+    year_built          INT,
+    year_commisioning   INT,
+    cost_value          NUMERIC,
+    registration_date   DATE,
+    status              TEXT,
+    ownership_type      TEXT,
+    cultural_heritage   TEXT,
+    underground_floors  INT,
+    build_record_area   NUMERIC,
+    build_record_type   TEXT,
+    common_data_status  TEXT,
+    obj_type            TEXT,
+    raw_props           jsonb,
+    fetched_at          timestamptz NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS cad_buildings_geom_gist
+    ON cad_buildings USING GIST (geom);
+CREATE INDEX IF NOT EXISTS cad_buildings_quarter_idx
+    ON cad_buildings(quarter_cad_num);
+CREATE INDEX IF NOT EXISTS cad_buildings_purpose_idx
+    ON cad_buildings(purpose);
+
+COMMENT ON TABLE cad_buildings IS
+    'Здания (НСПД thematicSearchId=1) внутри cad_quarters_geom. '
+    'Источник cost_value для cross-check с sale_graph.';
+
+-- ── 2. nspd_scrape_runs ──────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS nspd_scrape_runs (
+    run_id            BIGSERIAL PRIMARY KEY,
+    started_at        timestamptz NOT NULL DEFAULT NOW(),
+    finished_at       timestamptz,
+    heartbeat_at      timestamptz,
+    region_code       INT NOT NULL,
+    pending_count     INT,
+    quarters_ok       INT NOT NULL DEFAULT 0,
+    quarters_failed   INT NOT NULL DEFAULT 0,
+    buildings_ok      INT NOT NULL DEFAULT 0,
+    requests_count    INT NOT NULL DEFAULT 0,
+    waf_429_count     INT NOT NULL DEFAULT 0,
+    status            TEXT NOT NULL DEFAULT 'running',  -- running | done | failed | zombie | skipped
+    error             TEXT,
+    triggered_by      TEXT NOT NULL DEFAULT 'beat'      -- beat | manual | resume
+);
+
+CREATE INDEX IF NOT EXISTS nspd_runs_status_idx
+    ON nspd_scrape_runs(status, started_at DESC);
+CREATE INDEX IF NOT EXISTS nspd_runs_region_idx
+    ON nspd_scrape_runs(region_code, started_at DESC);
+
+COMMENT ON TABLE nspd_scrape_runs IS
+    'Журнал NSPD-скрейпов кадастровых кварталов и зданий. Один прогон на регион. '
+    'Beat-расписание: 20-е число февраля/мая/августа/ноября (после кв. публикаций rosreestr).';
+
+-- ── 3. nspd_scrape_log ───────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS nspd_scrape_log (
+    log_id      BIGSERIAL PRIMARY KEY,
+    run_id      BIGINT REFERENCES nspd_scrape_runs(run_id) ON DELETE CASCADE,
+    ts          timestamptz NOT NULL DEFAULT NOW(),
+    level       TEXT NOT NULL DEFAULT 'info',   -- debug | info | warn | error
+    stage       TEXT,                            -- task_received | discover | quarter_fetch | buildings | commit | done | error
+    cad_number  TEXT,
+    message     TEXT
+);
+
+CREATE INDEX IF NOT EXISTS nspd_log_run_idx
+    ON nspd_scrape_log(run_id, ts DESC);
+CREATE INDEX IF NOT EXISTS nspd_log_level_idx
+    ON nspd_scrape_log(level, ts DESC) WHERE level IN ('warn', 'error');
+
+COMMENT ON TABLE nspd_scrape_log IS
+    'Построчный лог NSPD-скрейпа: какие cad-кварталы успешны/упали, '
+    'WAF-403 ретраи, фазы commit. Используется UI /scrape/nspd для мониторинга.';

data/sql/64_v_zk_rosreestr_velocity.sql

@@ -0,0 +1,116 @@
+-- v_zk_rosreestr_velocity refresh: + cad_buildings cost_value для cross-check.
+--
+-- Что добавилось vs 60_v_zk_rosreestr_velocity.sql:
+--   1. quarter_cadastre_avg_th_per_m2 — медианная кадастровая стоимость м²
+--      строений (≥3 этажей) в том же кадастровом квартале.
+--   2. cadastre_vs_market_pct — премиум/дисконт рынка относительно кадастра
+--      (положительные значения = рынок дороже кадастра, что норма).
+--      Аномалии (>+50% или <-30%) пригодны для warning-badge на UI.
+--
+-- Логика: cad_buildings.cost_value хранится в ₽ на здание; делим на area
+-- чтобы получить ₽/м². Берём только residential (purpose ILIKE
+-- '%многокварт%' OR floors >= 3) — низкоэтажки и склады искажают.
+-- Нет фильтра по году ввода: даже старый фонд даёт ориентир для зонирования.
+--
+-- Источник истины cost_value: НСПД thematicSearchId=1 → properties.options.cost_value.
+
+CREATE OR REPLACE VIEW v_zk_rosreestr_velocity AS
+WITH zk_pool AS (
+    SELECT DISTINCT ON (o.obj_id)
+           o.obj_id,
+           o.comm_name,
+           o.dev_name,
+           o.district_name,
+           o.flat_count,
+           o.cad_quarter,
+           o.obj_class
+      FROM domrf_kn_objects o
+     WHERE o.region_cd = 66
+       AND o.cad_quarter IS NOT NULL
+       AND o.flat_count IS NOT NULL
+       AND o.flat_count > 0
+       AND o.site_status = 'Строящиеся'
+     ORDER BY o.obj_id, o.snapshot_date DESC
+),
+quarter_totals AS (
+    SELECT cad_quarter, SUM(flat_count)::numeric AS total_flats
+      FROM zk_pool
+     GROUP BY cad_quarter
+),
+quarter_deals AS (
+    SELECT quarter_cad_number AS cad_quarter,
+           COUNT(*)                                            AS deals_12mo,
+           AVG(price_per_sqm)                                  AS avg_price_pm,
+           PERCENTILE_CONT(0.5) WITHIN GROUP
+               (ORDER BY price_per_sqm)                        AS median_price_pm,
+           AVG(area)                                           AS avg_area_sqm
+      FROM rosreestr_deals
+     WHERE region_code = 66
+       AND doc_type = 'ДДУ'
+       AND realestate_type_code = '002001003000'
+       AND area > 10 AND area <= 200
+       AND price_per_sqm BETWEEN 30000 AND 1000000
+       AND period_start_date >= NOW() - INTERVAL '12 months'
+     GROUP BY quarter_cad_number
+),
+quarter_cadastre AS (
+    -- Медианная кадастровая стоимость ₽/м² по жилым строениям квартала.
+    -- Фильтры:
+    --   * cost_value > 0 и area >= 100 м² (исключаем подсобки)
+    --   * floors >= 3 ИЛИ purpose ILIKE '%многокв%' (отсеиваем ИЖС/гаражи)
+    --   * ₽/м² 5К..500К (уберает мусор и грубые ошибки росреестра)
+    SELECT
+        quarter_cad_num                                         AS cad_quarter,
+        COUNT(*)                                                AS buildings_n,
+        PERCENTILE_CONT(0.5) WITHIN GROUP
+            (ORDER BY cost_value / NULLIF(area, 0))             AS median_cost_per_m2
+      FROM cad_buildings
+     WHERE quarter_cad_num IS NOT NULL
+       AND cost_value IS NOT NULL
+       AND area IS NOT NULL
+       AND area >= 100
+       AND (floors IS NOT NULL AND floors >= 3
+            OR purpose ILIKE '%многокв%')
+       AND (cost_value / NULLIF(area, 0)) BETWEEN 5000 AND 500000
+     GROUP BY quarter_cad_num
+)
+SELECT
+    z.obj_id,
+    z.comm_name,
+    z.dev_name,
+    z.district_name,
+    z.obj_class,
+    z.flat_count,
+    z.cad_quarter,
+    qt.total_flats                                              AS quarter_total_flats,
+    qd.deals_12mo                                               AS quarter_deals_12mo,
+    ROUND(qd.deals_12mo * z.flat_count::numeric / qt.total_flats, 1)
+                                                                AS estimated_deals_12mo,
+    ROUND(qd.deals_12mo * z.flat_count::numeric / qt.total_flats / 12.0, 2)
+                                                                AS estimated_velocity_pm,
+    ROUND(qd.median_price_pm / 1000.0, 1)                       AS median_price_th_per_m2,
+    ROUND(qd.avg_area_sqm, 1)                                   AS avg_deal_area_sqm,
+    -- Кадастровая стоимость и cross-check.
+    qc.buildings_n                                              AS cadastre_buildings_n,
+    ROUND(qc.median_cost_per_m2 / 1000.0, 1)                    AS cadastre_th_per_m2,
+    CASE
+        WHEN qc.median_cost_per_m2 IS NOT NULL
+             AND qd.median_price_pm IS NOT NULL
+             AND qc.median_cost_per_m2 > 0
+        THEN ROUND(
+            (qd.median_price_pm - qc.median_cost_per_m2)
+                / qc.median_cost_per_m2 * 100.0,
+            1
+        )
+        ELSE NULL
+    END                                                         AS cadastre_vs_market_pct
+  FROM zk_pool z
+  JOIN quarter_totals qt ON qt.cad_quarter = z.cad_quarter
+  JOIN quarter_deals qd ON qd.cad_quarter = z.cad_quarter
+  LEFT JOIN quarter_cadastre qc ON qc.cad_quarter = z.cad_quarter
+;
+
+COMMENT ON VIEW v_zk_rosreestr_velocity IS
+    'v2: per-ЖК velocity/цена через rosreestr-сделки квартала + кадастровая '
+    'стоимость м² из cad_buildings (NSPD). cadastre_vs_market_pct = премиум '
+    'рынка над кадастром, аномалии (>+50% / <-30%) — outliers для warning UI.';

docker-compose.prod.yml

@@ -48,7 +48,12 @@ services:
   backend:
     image: ghcr.io/lekss361/gendesign-backend:${IMAGE_TAG:-latest}
     restart: unless-stopped
-    env_file: ./backend/.env
+    # .env.runtime пишется deploy.yml через SSH (SENTRY_RELEASE=$IMAGE_TAG).
+    # required: false — compose не падает если файла нет (первый деплой).
+    env_file:
+      - path: ./backend/.env
+      - path: ./backend/.env.runtime
+        required: false
     depends_on:
       postgres:
         condition: service_healthy
@@ -75,7 +80,10 @@ services:
     # Отдельный chromium-образ (+200 МБ Playwright). См. backend/Dockerfile target=runner-with-chromium.
     image: ghcr.io/lekss361/gendesign-worker:${IMAGE_TAG:-latest}
     restart: unless-stopped
-    env_file: ./backend/.env
+    env_file:
+      - path: ./backend/.env
+      - path: ./backend/.env.runtime
+        required: false
     depends_on:
       postgres:
         condition: service_healthy
@@ -89,7 +97,10 @@ services:
     # Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.
     image: ghcr.io/lekss361/gendesign-backend:${IMAGE_TAG:-latest}
     restart: unless-stopped
-    env_file: ./backend/.env
+    env_file:
+      - path: ./backend/.env
+      - path: ./backend/.env.runtime
+        required: false
     depends_on:
       redis:
         condition: service_started

frontend/src/app/analytics/recommend/page.tsx

@@ -175,7 +175,7 @@ export default function RecommendPage() {
                   >
                     📊
                     {data.scope.mortgage_rate_pct != null
-                      ? ` Ставка ЦБ ${data.scope.mortgage_rate_pct.toFixed(2)}% (${data.scope.mortgage_rate_period})`
+                      ? ` Средневзв. ИЖК ${data.scope.mortgage_rate_pct.toFixed(2)}% (${data.scope.mortgage_rate_period}, со льготами; рыночная ~20%)`
                       : " ставка ЦБ нет данных"}
                     {data.scope.poi_score != null &&
                     data.scope.poi_score_city_avg != null

frontend/src/components/analytics/RecommendVelocityPanel.tsx

@@ -48,28 +48,35 @@ export function RecommendVelocityPanel({
     [priceFactor, elasticity],
   );
 
-  // Aggregate live recompute
+  // Aggregate live recompute. Темп и сроки считаем per-bucket с СВОЕЙ
+  // эластичностью (Tier 3) — потому что Студии и 80+ м² реагируют на цену
+  // по-разному. pfPow выше — fallback для bucket'ов без своей эластичности.
   const totals = useMemo(() => {
     let units = 0;
     let revenue = 0;
     let baseVelocity = 0; // sum of bucket velocity at price_factor=1
-    let weightedSold24 = 0; // weighted sum for liquidity
+    let adjustedVelocity = 0; // sum of bucket velocity × bucket-specific pf^e
+    let weightedSold24 = 0;
     for (const r of derivedRows) {
       const u = r.effective_units ?? 0;
       units += u;
-      // Revenue scales linearly with price_factor (price_median × pf).
       const baseRev = r.effective_revenue_rub ?? 0;
       revenue += baseRev * priceFactor;
       const v = r.velocity_per_month ?? 0;
-      baseVelocity += v * (r.effective_share_pct / Math.max(r.share_pct, 0.01));
+      const shareRatio = r.effective_share_pct / Math.max(r.share_pct, 0.01);
-      const adjustedV = v * pfPow;
+      baseVelocity += v * shareRatio;
+      const be = r.elasticity ?? elasticity;
+      const bucketPfPow = priceFactor > 0 ? priceFactor ** be : 1;
+      const adjustedV = v * bucketPfPow;
+      adjustedVelocity += adjustedV * shareRatio;
       if (u > 0 && adjustedV > 0) {
         const months = u / adjustedV;
         const fracIn24 = Math.min(1, 24 / months);
         weightedSold24 += fracIn24 * u;
       }
     }
-    const tempo = baseVelocity * pfPow;
+    const tempo =
+      adjustedVelocity > 0 ? adjustedVelocity : baseVelocity * pfPow;
     const monthsToSellout = tempo > 0 && units > 0 ? units / tempo : null;
     const liquidity = units > 0 ? (weightedSold24 / units) * 100 : null;
     const avgTicket = units > 0 && revenue > 0 ? revenue / units : null;
@@ -81,7 +88,7 @@ export function RecommendVelocityPanel({
       liquidity,
       avgTicket,
     };
-  }, [derivedRows, priceFactor, pfPow]);
+  }, [derivedRows, priceFactor, pfPow, elasticity]);
 
   const liquidityColor =
     totals.liquidity == null
@@ -243,6 +250,132 @@ export function RecommendVelocityPanel({
         </div>
       ) : null}
 
+      {/* Cadastre vs market cross-check (NSPD ↔ rosreestr) */}
+      {scope.cadastre_median_per_m2 != null &&
+      scope.cadastre_vs_market_pct != null
+        ? (() => {
+            const pct = scope.cadastre_vs_market_pct;
+            const isAnomaly = pct > 50 || pct < -30;
+            const bg = isAnomaly ? "#fef2f2" : "#f0fdf4";
+            const border = isAnomaly ? "#fecaca" : "#bbf7d0";
+            const fg = isAnomaly ? "#b3261e" : "#0a7a3a";
+            return (
+              <div
+                style={{
+                  marginTop: 12,
+                  padding: 8,
+                  background: bg,
+                  border: `1px solid ${border}`,
+                  borderRadius: 6,
+                  fontSize: 12,
+                  lineHeight: 1.4,
+                }}
+                title={`Спред NSPD-кадастра и rosreestr-сделок. Норма: 0..+50% (рынок справедливо дороже кадастра, тк кадастр обычно отстаёт). Аномалии: >+50% (переоценка рынка) или <-30% (рынок дешевле кадастра, странно).`}
+              >
+                <strong>Кадастр vs Рынок: </strong>
+                кадастровая медиана{" "}
+                <strong>
+                  {(scope.cadastre_median_per_m2 / 1000).toFixed(0)} тыс ₽/м²
+                </strong>{" "}
+                (по {scope.cadastre_buildings_n} зданиям NSPD), рынок{" "}
+                <strong>
+                  {scope.district_median_price_per_m2 != null
+                    ? `${(scope.district_median_price_per_m2 / 1000).toFixed(0)} тыс ₽/м²`
+                    : "—"}
+                </strong>
+                {" → "}
+                <strong style={{ color: fg }}>
+                  {pct > 0 ? "+" : ""}
+                  {pct.toFixed(0)}%
+                </strong>
+                {isAnomaly ? (
+                  <span
+                    style={{
+                      marginLeft: 6,
+                      padding: "1px 6px",
+                      background: "#fff",
+                      border: `1px solid ${border}`,
+                      borderRadius: 3,
+                      color: fg,
+                      fontWeight: 600,
+                    }}
+                  >
+                    ⚠ Аномалия
+                  </span>
+                ) : null}
+              </div>
+            );
+          })()
+        : null}
+
+      {/* Per-bucket elasticity breakdown — Tier 3 */}
+      {scope.elasticity_per_bucket &&
+      Object.keys(scope.elasticity_per_bucket).length > 0 ? (
+        <div
+          style={{
+            marginTop: 12,
+            padding: 8,
+            background: "#f8fafc",
+            border: "1px solid #e6e8ec",
+            borderRadius: 6,
+            fontSize: 11,
+          }}
+        >
+          <div
+            style={{
+              fontWeight: 600,
+              color: "#374151",
+              marginBottom: 4,
+              textTransform: "uppercase",
+              letterSpacing: 0.4,
+            }}
+          >
+            Эластичность по сегментам
+          </div>
+          <div
+            style={{
+              display: "grid",
+              gridTemplateColumns: "repeat(auto-fit, minmax(140px, 1fr))",
+              gap: 6,
+            }}
+          >
+            {Object.entries(scope.elasticity_per_bucket).map(([b, info]) => {
+              const isRegr = info.source === "regression";
+              return (
+                <div
+                  key={b}
+                  style={{
+                    background: "#fff",
+                    padding: "4px 6px",
+                    borderRadius: 4,
+                    border: `1px solid ${isRegr ? "#bbf7d0" : "#e6e8ec"}`,
+                  }}
+                  title={
+                    isRegr
+                      ? `regression: R²=${info.r2.toFixed(2)}, n=${info.n}`
+                      : `fallback (общая эластичность ${elasticity}); n=${info.n} мало для регрессии`
+                  }
+                >
+                  <span style={{ color: "#5b6066" }}>{b}: </span>
+                  <strong style={{ color: isRegr ? "#0a7a3a" : "#9a6700" }}>
+                    {info.elasticity.toFixed(2)}
+                  </strong>
+                  {isRegr ? (
+                    <span style={{ color: "#9ca3af", marginLeft: 4 }}>
+                      (n={info.n})
+                    </span>
+                  ) : (
+                    <span style={{ color: "#9ca3af", marginLeft: 4 }}>
+                      (fb)
+                    </span>
+                  )}
+                </div>
+              );
+            })}
+          </div>
+        </div>
+      ) : null}
+
       {/* Methodology note */}
       <div
         style={{
@@ -257,7 +390,12 @@ export function RecommendVelocityPanel({
         {scope.elasticity_source === "regression"
           ? `регрессия sale_graph: R²=${scope.elasticity_r2.toFixed(2)}, n=${scope.elasticity_n}`
           : `по умолчанию — sale_graph недостаточно (n=${scope.elasticity_n})`}
-        ). Базовый темп{" "}
+        )
+        {scope.elasticity_weighted != null &&
+        Math.abs(scope.elasticity_weighted - elasticity) > 0.05
+          ? ` · взвешенная по бакетам: ${scope.elasticity_weighted.toFixed(2)}`
+          : ""}
+        . Базовый темп{" "}
         <strong>{scope.market_velocity_per_month?.toFixed(1) ?? "—"}</strong>{" "}
         кв/мес (
         {scope.velocity_source === "sale_graph"

frontend/src/types/analytics.ts

@@ -220,6 +220,19 @@ export interface RecommendBucket {
   revenue_planned_rub: number | null;
   velocity_per_month: number | null;
   months_to_sellout: number | null;
+  elasticity?: number;
+  elasticity_r2?: number;
+  elasticity_n?: number;
+  elasticity_source?: "regression" | "fallback_global";
+}
+
+export interface ElasticityPerBucket {
+  [bucketPretty: string]: {
+    elasticity: number;
+    r2: number;
+    n: number;
+    source: "regression" | "fallback_global";
+  };
 }
 
 export interface RecommendComparable {
@@ -270,6 +283,11 @@ export interface RecommendMixOutput {
     elasticity_r2: number;
     elasticity_n: number;
     elasticity_source: "regression" | "fallback";
+    elasticity_weighted: number | null;
+    elasticity_per_bucket: ElasticityPerBucket;
+    cadastre_median_per_m2: number | null;
+    cadastre_buildings_n: number;
+    cadastre_vs_market_pct: number | null;
     price_factor_applied: number;
     required_price_factor: number | null;
     target_months: number | null;