feat(analytics): velocity-anomaly alerts (#17) + ARN ДДУ price indicator (#99)

#17: detect_velocity_anomalies + GET /analytics/velocity-alerts — z-score drop-detection
на domrf_kn_sale_graph (double-gate z≤-2 AND drop≤-30%, starvation-guards). Snapshot=richest
(не MAX — prod новейшие частичные), lookback anchored на latest report_month (scrape лаг ~4мес).
Prod: ЖК Центральный Парк -69%, ~14ms.

#99: mv_ddu_price_indicator (миграция 152) + POST /market/ddu-indicator — ARN-mirror
ценовой индекс per quarter×area_bucket из rosreestr_deals (ДДУ регион 66). MVP: subject-level,
period Q, window 2025-Q2+, methods 1/2 (basis/previous, prev_period_value honesty). Q1-2026
headline 1.0185 vs ARN 1.03 (±5%). Method 3 blocked (нет pre-2025-Q2 данных) — задокументировано.

Closes #17
Closes #99

backend/app/api/v1/analytics.py

@@ -12,6 +12,7 @@ from app.core.db import get_db
 from app.schemas.complex_buildings import ComplexBuilding
 from app.schemas.recommend import RecommendMixInput, RecommendMixOutput
 from app.services import analytics_queries as q
+from app.services.analytics import velocity_alerts as va
 
 router = APIRouter()
 
@@ -208,6 +209,35 @@ def object_buildings(
     return q.complex_buildings(db, obj_id=obj_id)
 
 
+# ---- Velocity alerts (Issue #17) -------------------------------------------
+
+
+@router.get("/velocity-alerts")
+def velocity_alerts(
+    db: Annotated[Session, Depends(get_db)],
+    region_code: Annotated[int, Query(ge=1, le=99)] = 66,
+    lookback_months: Annotated[int, Query(ge=6, le=24)] = 9,
+    min_zscore: Annotated[float, Query(ge=0.5, le=5.0)] = 2.0,
+    min_drop_pct: Annotated[float, Query(ge=0.0, le=100.0)] = 30.0,
+    limit: Annotated[int, Query(ge=1, le=200)] = 100,
+) -> dict[str, Any]:
+    """ЖК с statistical-significance провалом темпа продаж vs своя история.
+
+    Источник: domrf_kn_sale_graph (monthly realised). Детерминированная
+    z-score детекция: recent 3-mo mean vs trailing prior-period mean.
+    Алерт когда z <= -min_zscore И drop_pct <= -min_drop_pct (двойной gate
+    снижает false-positive на шумных low-volume ЖК). severity high/medium.
+    """
+    return va.detect_velocity_anomalies(
+        db,
+        region_code=region_code,
+        lookback_months=lookback_months,
+        min_zscore=min_zscore,
+        min_drop_pct=min_drop_pct,
+        limit=limit,
+    )
+
+
 # ---- PRINZIP-specific -------------------------------------------------------
 
 

backend/app/api/v1/market.py

@@ -0,0 +1,50 @@
+"""Market indicator endpoints (Issue #99) — ARN-framework for primary market.
+
+POST /api/v1/market/ddu-indicator — per-quarter price indicator of the PRIMARY
+market (новостройки/ДДУ) for Свердловская обл., ARN-mirror request/response.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.db import get_db
+from app.schemas.market import DduIndicatorRequest, DduIndicatorResponse
+from app.services.analytics import ddu_price_indicator as ddu
+
+router = APIRouter()
+
+
+@router.post("/ddu-indicator", response_model=DduIndicatorResponse)
+def ddu_indicator(
+    payload: DduIndicatorRequest,
+    db: Annotated[Session, Depends(get_db)],
+) -> DduIndicatorResponse:
+    """ARN-style ценовой индикатор первичного рынка (ДДУ) по кварталам × площади.
+
+    Принимает то же body, что ARN/НСПД (camelCase). MVP: period_type=Q,
+    calc methods 1 (базисный) / 2 (предыдущий), субъект 66. Неподдержанные
+    ARN-опции (M/H/Y, method=3, иные субъекты, cad_quarter) не отвергаются —
+    отражаются в `notes`. Источник: mv_ddu_price_indicator (migration 152).
+
+    Killer-USP: «у нас первичка, у НСПД вторичка» — обе нужны девелоперу.
+    """
+    result = ddu.get_ddu_indicator(
+        db,
+        calculation_method=payload.calculation_method,
+        period_from=payload.period_from,
+        period_to=payload.period_to,
+        area_ranges=payload.area_ranges,
+        federal_subject=payload.federal_subject,
+        indicators=payload.indicators,
+    )
+    # cadastral_quarter is shape-compat only; flag it if the client sent one.
+    if payload.cadastral_quarter:
+        result["notes"].append(
+            "cadastralQuarter не поддержан: первичка (ДДУ) слишком разрежена "
+            "на уровне кад.квартала — индикатор считается по субъекту (66)."
+        )
+    return DduIndicatorResponse(**result)

backend/app/main.py

@@ -31,6 +31,7 @@ from app.api.v1 import (
     insights,
     landing,
     locations,
+    market,
     me,
     own_projects,
     parcels,
@@ -163,6 +164,7 @@ app.include_router(concepts.router, prefix="/api/v1/concepts", tags=["concepts"]
 app.include_router(chat.router, prefix="/api/v1/chat", tags=["chat"])
 app.include_router(parcels.router, prefix="/api/v1/parcels", tags=["parcels"])
 app.include_router(analytics.router, prefix="/api/v1/analytics", tags=["analytics"])
+app.include_router(market.router, prefix="/api/v1/market", tags=["market"])
 app.include_router(admin_scrape.router, prefix="/api/v1/admin/scrape", tags=["admin"])
 app.include_router(admin_jobs.router, prefix="/api/v1/admin/jobs", tags=["admin"])
 app.include_router(admin_leads.router, prefix="/api/v1/admin/leads", tags=["admin"])

backend/app/schemas/market.py

@@ -0,0 +1,47 @@
+"""IO contracts for the market indicator endpoints (Issue #99).
+
+POST /api/v1/market/ddu-indicator — mirrors the ARN/НСПД request body so the
+frontend can reuse one UI pattern for primary (наша ДДУ) vs secondary (ARN).
+Field names are camelCase (ARN-compatible) via validation aliases; snake_case
+also accepted (populate_by_name).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class DduIndicatorRequest(BaseModel):
+    """ARN-shaped request body. Unsupported options are reported in `notes`
+    of the response rather than rejected (MVP supports Q / methods 1,2 / subj 66).
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    # ARN indicator types M/Q/H/Y. MVP supports only 'Q'.
+    indicators: list[str] | None = Field(default=None)
+    # 1 = basis, 2 = previous, (3 = year-ago — unsupported in MVP).
+    calculation_method: int = Field(default=2, ge=1, le=3, alias="calculationMethod")
+    # ARN-style period bounds, e.g. '2025-Q2' .. '2026-Q1'. None = unbounded.
+    period_from: str | None = Field(default=None, alias="periodFrom", max_length=10)
+    period_to: str | None = Field(default=None, alias="periodTo", max_length=10)
+    # ARN federal district ids (e.g. [6] = Уральский). Informational in MVP.
+    federal_district: list[int] | None = Field(default=None, alias="federalDistrict")
+    # ARN subject codes. Only '66' (Свердловская обл.) is backed by data.
+    federal_subject: list[str] | None = Field(default=None, alias="federalSubject")
+    # ARN cadastral quarters. NOT supported (primary-market ДДУ too sparse per
+    # cad_quarter) — accepted for shape-compat, reported as unsupported.
+    cadastral_quarter: list[str] | None = Field(default=None, alias="cadastralQuarter")
+    # ARN area-range ids 1..6 (0 = all-area headline). None → all present.
+    area_ranges: list[int] | None = Field(default=None, alias="areaRanges")
+
+
+class DduIndicatorResponse(BaseModel):
+    """Free-form ARN-mirror response (table + graph + meta + notes)."""
+
+    meta: dict[str, Any]
+    table: list[dict[str, Any]]
+    graph: list[dict[str, Any]]
+    notes: list[str]

backend/app/services/analytics/__init__.py

@@ -0,0 +1,6 @@
+"""Analytics services package.
+
+Modules:
+- velocity_alerts: detection of sales-velocity anomalies per ЖК (Issue #17).
+- ddu_price_indicator: ARN-style primary-market (ДДУ) price indicator (Issue #99).
+"""

backend/app/services/analytics/ddu_price_indicator.py

@@ -0,0 +1,278 @@
+"""ДДУ price indicator — ARN-framework applied to primary market (Issue #99).
+
+Reads ``mv_ddu_price_indicator`` (migration 152) and serves an ARN-mirror
+response: per-quarter price indicator of the PRIMARY market (новостройки/ДДУ)
+for Свердловская область (region 66), broken down by area bucket.
+
+This is a deterministic statistical indicator (median price/m² ratios), not a
+forecast. See the migration header for the full MVP-scope rationale; the short
+version of the honest simplifications versus the full ARN matrix:
+
+- Granularity = subject (region 66), NOT cad_quarter — primary-market ДДУ rows
+  are far too sparse per cad_quarter after the 2025 rosreestr aggregation change.
+- period_type = 'Q' only (source publishes quarterly; no monthly granularity).
+- Window starts 2025-Q2 (hard data-regime break at 2025-Q1).
+- calc_method 1 (basis) and 2 (previous) only. Method 3 (year-ago) is BLOCKED:
+  there is no comparable pre-2025-Q2 primary-market data.
+
+The endpoint accepts the same request body shape as the ARN API (indicators,
+calculationMethod, periodFrom/To, areaRanges, federalSubject) so the frontend
+can reuse one UI pattern. Unsupported ARN options are reported in a ``notes``
+list rather than silently ignored.
+"""
+
+from __future__ import annotations
+
+import logging
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# ARN area-range labels (poquartirnaya area, m²). Index = area_bucket id.
+_BUCKET_LABELS: dict[int, str] = {
+    0: "Все площади",
+    1: "0–25 м²",
+    2: "25–40 м²",
+    3: "40–60 м²",
+    4: "60–80 м²",
+    5: "80–100 м²",
+    6: "100+ м²",
+}
+
+# Calc methods supported by the MVP. 3 (year-ago) is intentionally absent.
+_CALC_BASIS = 1
+_CALC_PREVIOUS = 2
+_SUPPORTED_METHODS = (_CALC_BASIS, _CALC_PREVIOUS)
+
+# Only Свердловская область (66) and the Уральский federal district are backed
+# by data today; other ARN federalSubject / federalDistrict values are reported
+# as unsupported.
+_SUPPORTED_SUBJECT = "66"
+
+
+def _f(value: Any) -> float | None:
+    if value is None:
+        return None
+    if isinstance(value, Decimal):
+        return float(value)
+    return float(value)
+
+
+def _normalize_subjects(federal_subject: list[str] | None) -> tuple[bool, list[str]]:
+    """Return (subject_supported, unsupported_subjects)."""
+    if not federal_subject:
+        return True, []
+    unsupported = [s for s in federal_subject if str(s) != _SUPPORTED_SUBJECT]
+    supported = any(str(s) == _SUPPORTED_SUBJECT for s in federal_subject)
+    return supported, unsupported
+
+
+def get_ddu_indicator(
+    db: Session,
+    *,
+    calculation_method: int = _CALC_PREVIOUS,
+    period_from: str | None = None,
+    period_to: str | None = None,
+    area_ranges: list[int] | None = None,
+    federal_subject: list[str] | None = None,
+    indicators: list[str] | None = None,
+) -> dict[str, Any]:
+    """ARN-mirror response for the primary-market ДДУ price indicator.
+
+    Args:
+        db: SQLAlchemy session (sync).
+        calculation_method: 1 = basis (vs first period), 2 = previous quarter.
+        period_from / period_to: inclusive ARN-style bounds, e.g. '2025-Q2'..'2026-Q1'.
+            None = no bound on that side.
+        area_ranges: ARN area-bucket ids to include (1..6); 0 = all-area headline.
+            None / empty → all buckets present in the MV.
+        federal_subject: ARN subject codes. Only '66' is backed by data.
+        indicators: ARN indicator types (M/Q/H/Y). Only 'Q' is supported.
+
+    Returns:
+        ``{"meta": {...}, "table": [...], "graph": [...], "notes": [...]}``.
+        ``table`` rows carry both the chosen-method ``index`` and the raw median;
+        ``graph`` is the headline (bucket 0) series for charting.
+    """
+    notes: list[str] = []
+
+    method = calculation_method if calculation_method in _SUPPORTED_METHODS else _CALC_PREVIOUS
+    if calculation_method not in _SUPPORTED_METHODS:
+        notes.append(
+            f"calculationMethod={calculation_method} не поддержан "
+            f"(MVP: 1=базисный, 2=предыдущий; 3=годом-ранее недоступен — нет "
+            f"сопоставимых данных первички до 2025-Q2). Использован method=2 (предыдущий)."
+        )
+
+    if indicators:
+        unsupported_ind = [i for i in indicators if str(i).upper() != "Q"]
+        if unsupported_ind:
+            notes.append(
+                f"indicators={unsupported_ind} не поддержаны (источник rosreestr "
+                f"публикует данные поквартально; доступен только 'Q')."
+            )
+
+    subject_ok, unsupported_subjects = _normalize_subjects(federal_subject)
+    if unsupported_subjects:
+        notes.append(
+            f"federalSubject={unsupported_subjects} вне покрытия "
+            f"(данные первички только по 66 — Свердловская обл.)."
+        )
+    if not subject_ok:
+        # Asked exclusively for unsupported subjects → empty result, explained.
+        return {
+            "meta": {
+                "market": "primary_ddu",
+                "region_code": int(_SUPPORTED_SUBJECT),
+                "calculation_method": method,
+                "period_type": "Q",
+            },
+            "table": [],
+            "graph": [],
+            "notes": notes,
+        }
+
+    index_col = "index_basis" if method == _CALC_BASIS else "index_previous"
+
+    # area_ranges filter (optional). Validate ids to a safe int list — never
+    # interpolate into SQL; bound via ANY(CAST(:buckets AS int[])).
+    bucket_filter = ""
+    params: dict[str, Any] = {}
+    if area_ranges:
+        clean_buckets = sorted({int(b) for b in area_ranges if 0 <= int(b) <= 6})
+        if clean_buckets:
+            bucket_filter = "AND area_bucket = ANY(CAST(:buckets AS int[]))"
+            params["buckets"] = clean_buckets
+
+    period_filter = ""
+    if period_from:
+        period_filter += " AND period_value >= :period_from"
+        params["period_from"] = period_from
+    if period_to:
+        period_filter += " AND period_value <= :period_to"
+        params["period_to"] = period_to
+
+    try:
+        rows = (
+            db.execute(
+                text(
+                    f"""
+                    SELECT
+                        area_bucket,
+                        period_value,
+                        period_type,
+                        deals_count,
+                        median_price_per_m2,
+                        index_basis,
+                        index_previous,
+                        prev_period_value,
+                        {index_col} AS chosen_index
+                      FROM mv_ddu_price_indicator
+                     WHERE 1 = 1
+                       {bucket_filter}
+                       {period_filter}
+                     ORDER BY area_bucket, period_value
+                    """
+                ),
+                params,
+            )
+            .mappings()
+            .all()
+        )
+    except OperationalError:
+        # Most likely the MV does not exist yet (migration 152 not applied).
+        logger.exception("ddu_indicator: query failed (mv_ddu_price_indicator missing?)")
+        raise
+
+    table: list[dict[str, Any]] = []
+    graph: list[dict[str, Any]] = []
+    for r in rows:
+        bucket = int(r["area_bucket"])
+        item = {
+            "area_bucket": bucket,
+            "area_label": _BUCKET_LABELS.get(bucket, str(bucket)),
+            "period_value": r["period_value"],
+            "period_type": r["period_type"],
+            "deals_count": int(r["deals_count"]),
+            "median_price_per_m2": _f(r["median_price_per_m2"]),
+            "index": _f(r["chosen_index"]),
+            "index_basis": _f(r["index_basis"]),
+            "index_previous": _f(r["index_previous"]),
+            "prev_period_value": r["prev_period_value"],
+        }
+        table.append(item)
+        if bucket == 0:
+            graph.append(
+                {
+                    "period_value": r["period_value"],
+                    "index": _f(r["chosen_index"]),
+                    "median_price_per_m2": _f(r["median_price_per_m2"]),
+                    "deals_count": int(r["deals_count"]),
+                }
+            )
+
+    logger.info(
+        "ddu_indicator: method=%d buckets=%s period=%s..%s -> %d rows",
+        method,
+        params.get("buckets", "all"),
+        period_from or "*",
+        period_to or "*",
+        len(table),
+    )
+
+    return {
+        "meta": {
+            "market": "primary_ddu",
+            "region_code": int(_SUPPORTED_SUBJECT),
+            "calculation_method": method,
+            "period_type": "Q",
+            "source": "rosreestr_deals (ДДУ, realestate_type_code=002001003000)",
+            "arn_comparison": {
+                "note": (
+                    "ARN/НСПД считает вторичку; здесь первичка (новостройки). "
+                    "Для новостроек рост индекса ≠ инфляция — это сдвиг ассортимента + цен."
+                ),
+                "ekb_q1_2026_arn_secondary": 1.03,
+            },
+        },
+        "table": table,
+        "graph": graph,
+        "notes": notes,
+    }
+
+
+def refresh_ddu_price_indicator(db: Session, *, concurrently: bool = True) -> int:
+    """REFRESH MATERIALIZED VIEW mv_ddu_price_indicator.
+
+    Mirrors refresh_layout_velocity(): CONCURRENTLY by default (needs the unique
+    index from migration 152), with a one-shot fallback to non-concurrent if the
+    MV is not yet populated. Not wired into Celery beat in this PR — manual /
+    admin invocation, like layout_velocity_refresh.
+
+    Returns the row count after refresh (observability).
+    """
+    try:
+        if concurrently:
+            db.execute(text("REFRESH MATERIALIZED VIEW CONCURRENTLY mv_ddu_price_indicator"))
+        else:
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_ddu_price_indicator"))
+        db.commit()
+    except OperationalError as e:
+        if concurrently and "cannot refresh materialized view" in str(e).lower():
+            logger.warning(
+                "ddu_indicator CONCURRENTLY failed (MV not populated), falling back"
+            )
+            db.rollback()
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_ddu_price_indicator"))
+            db.commit()
+        else:
+            raise
+    row = db.execute(text("SELECT COUNT(*) FROM mv_ddu_price_indicator")).first()
+    count = int(row[0]) if row else 0
+    logger.info("mv_ddu_price_indicator refreshed: %d rows", count)
+    return count

backend/app/services/analytics/velocity_alerts.py

@@ -0,0 +1,292 @@
+"""Velocity-anomaly detection per ЖК (Issue #17).
+
+Detects жилые комплексы whose recent sales velocity has dropped значительно
+versus their own trailing history. Deterministic (pure statistics — z-score +
+percent-change over the per-object monthly series). No ML, no external calls.
+
+Data source
+-----------
+`domrf_kn_sale_graph` — monthly per-object sale series (type='apartments').
+Columns used: obj_id, report_month, realised (monthly units sold — FLOW, not
+cumulative — verified on prod 2026-06: values rise and fall month-to-month),
+snapshot_date.
+
+Object metadata (comm_name / dev_name / region_cd) is joined from
+`domrf_kn_objects` (latest snapshot per obj_id).
+
+Real-data gotchas (verified on prod)
+-------------------------------------
+1. **Snapshot coverage is uneven.** `MAX(snapshot_date)` is NOT safe: the two
+   newest snapshots on prod (2026-06-03, 2026-04-27) are partial scrapes with
+   only 2–7 objects. The richest snapshots carry ~440 objects. We therefore
+   pick the snapshot with the MOST distinct objects (ties broken by recency)
+   and run the whole analysis within that single consistent snapshot.
+2. **All sale_graph objects are region 66** today, but we still filter by
+   region via domrf_kn_objects.region_cd so the endpoint stays correct if
+   other regions are scraped later.
+3. **History is ~9 months** in the richest snapshot, so `lookback_months`
+   defaults to 9 and we require ≥6 months (issue acceptance) before scoring.
+
+Algorithm (per object, within the chosen snapshot)
+--------------------------------------------------
+- Order months descending; take the most recent 3 as the "recent" window and
+  everything older (within lookback) as the "prior" window.
+- recent_velocity = mean(realised) over recent 3 months.
+- prior_velocity  = mean(realised) over prior months.
+- prior_std       = sample stddev of prior months.
+- z = (recent_velocity - prior_velocity) / prior_std   (negative ⇒ slowdown)
+- drop_pct = (recent_velocity - prior_velocity) / prior_velocity * 100
+- An object is an alert when z <= -min_zscore AND drop_pct <= -min_drop_pct
+  (both gates: statistical significance AND material magnitude — keeps the
+  false-positive rate down on noisy low-volume objects, per acceptance ≤10%).
+
+Severity
+--------
+- 'high'   : z <= -3.0  OR drop_pct <= -60
+- 'medium' : otherwise (still past the alert gates)
+"""
+
+from __future__ import annotations
+
+import logging
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Minimum months of history before an object is eligible (issue: ignore <6 mo).
+_MIN_MONTHS_HISTORY = 6
+
+# Recent window size (rolling 3-mo mean), per issue spec.
+_RECENT_WINDOW = 3
+
+# Minimum prior observations needed for a meaningful stddev / z-score.
+_MIN_PRIOR_OBS = 3
+
+# Default magnitude gate alongside the z-score gate. A pure z-score alert on a
+# low-volume object (prior mean ~2/mo, stddev ~1) fires on tiny absolute moves;
+# requiring a real percent drop suppresses those false positives.
+_DEFAULT_MIN_DROP_PCT = 30.0
+
+# Severity thresholds.
+_HIGH_ZSCORE = 3.0
+_HIGH_DROP_PCT = 60.0
+
+
+def _f(value: Any) -> float | None:
+    if value is None:
+        return None
+    if isinstance(value, Decimal):
+        return float(value)
+    return float(value)
+
+
+def _best_snapshot(db: Session, *, sale_type: str = "apartments") -> Any | None:
+    """Pick the sale_graph snapshot with the most distinct objects.
+
+    NOT ``MAX(snapshot_date)`` — the freshest snapshots on prod are frequently
+    partial scrapes (2–7 objects). The richest snapshot gives the broadest,
+    most consistent cohort. Ties broken by recency.
+    """
+    return db.execute(
+        text(
+            """
+            SELECT snapshot_date
+              FROM domrf_kn_sale_graph
+             WHERE type = :sale_type
+             GROUP BY snapshot_date
+             ORDER BY COUNT(DISTINCT obj_id) DESC, snapshot_date DESC
+             LIMIT 1
+            """
+        ),
+        {"sale_type": sale_type},
+    ).scalar()
+
+
+def detect_velocity_anomalies(
+    db: Session,
+    region_code: int = 66,
+    lookback_months: int = 9,
+    min_zscore: float = 2.0,
+    min_drop_pct: float = _DEFAULT_MIN_DROP_PCT,
+    limit: int = 100,
+) -> dict[str, Any]:
+    """Return ЖК whose recent velocity dropped significantly vs their history.
+
+    Args:
+        db: SQLAlchemy session (sync).
+        region_code: filter via domrf_kn_objects.region_cd (66 = Свердловская обл.).
+        lookback_months: months of history to consider (recent + prior windows).
+        min_zscore: absolute z-score threshold; alert when z <= -min_zscore.
+        min_drop_pct: absolute percent-drop threshold; alert when drop_pct <= -value.
+        limit: max alerts returned (ordered by z-score, most severe first).
+
+    Returns:
+        ``{"snapshot_date": str|None, "region_code": int, "params": {...},
+           "alerts": [...]}``. ``alerts`` is empty when there is no usable
+        snapshot or no object clears both gates.
+    """
+    snapshot = _best_snapshot(db)
+    if snapshot is None:
+        logger.info("velocity_alerts: no sale_graph snapshot available")
+        return {
+            "snapshot_date": None,
+            "region_code": region_code,
+            "params": {
+                "lookback_months": lookback_months,
+                "min_zscore": min_zscore,
+                "min_drop_pct": min_drop_pct,
+            },
+            "alerts": [],
+        }
+
+    # One pass: window the per-object monthly series, compute recent/prior
+    # aggregates, score, then join names + filter by region. All bind params use
+    # CAST(:x AS type) per repo convention (psycopg v3 ignores :x::type).
+    rows = (
+        db.execute(
+            text(
+                """
+                WITH anchor AS (
+                    -- Anchor the lookback to the LATEST report_month present in
+                    -- this snapshot, NOT snapshot_date. On prod the scrape date
+                    -- (snapshot_date) can lag the newest data month by months,
+                    -- so anchoring to snapshot_date would silently truncate the
+                    -- prior window and starve the z-score (prior_n < min).
+                    SELECT MAX(report_month) AS max_month
+                      FROM domrf_kn_sale_graph
+                     WHERE type = 'apartments'
+                       AND snapshot_date = :snap
+                ),
+                series AS (
+                    SELECT g.obj_id, g.report_month, g.realised
+                      FROM domrf_kn_sale_graph g
+                     CROSS JOIN anchor a
+                     WHERE g.type = 'apartments'
+                       AND g.snapshot_date = :snap
+                       AND g.report_month > (
+                               a.max_month - CAST(:lookback_interval AS interval))
+                ),
+                ranked AS (
+                    SELECT
+                        obj_id,
+                        realised,
+                        ROW_NUMBER() OVER (
+                            PARTITION BY obj_id ORDER BY report_month DESC) AS rn,
+                        COUNT(*) OVER (PARTITION BY obj_id) AS n_months
+                      FROM series
+                ),
+                windows AS (
+                    SELECT
+                        obj_id,
+                        n_months,
+                        AVG(realised) FILTER (WHERE rn <= :recent_window) AS recent_mean,
+                        AVG(realised) FILTER (WHERE rn > :recent_window)  AS prior_mean,
+                        STDDEV_SAMP(realised) FILTER (WHERE rn > :recent_window) AS prior_std,
+                        COUNT(*) FILTER (WHERE rn > :recent_window) AS prior_n
+                      FROM ranked
+                     GROUP BY obj_id, n_months
+                ),
+                scored AS (
+                    SELECT
+                        w.obj_id,
+                        w.recent_mean,
+                        w.prior_mean,
+                        w.prior_std,
+                        CASE WHEN w.prior_std > 0
+                             THEN (w.recent_mean - w.prior_mean) / w.prior_std
+                        END AS zscore,
+                        CASE WHEN w.prior_mean > 0
+                             THEN (w.recent_mean - w.prior_mean) / w.prior_mean * 100.0
+                        END AS drop_pct
+                      FROM windows w
+                     WHERE w.n_months >= :min_months
+                       AND w.prior_n >= :min_prior_obs
+                       AND w.prior_mean > 0
+                       AND w.prior_std > 0
+                )
+                SELECT
+                    s.obj_id,
+                    o.comm_name,
+                    o.dev_name,
+                    s.recent_mean,
+                    s.prior_mean,
+                    s.zscore,
+                    s.drop_pct
+                  FROM scored s
+                  JOIN LATERAL (
+                        SELECT comm_name, dev_name, region_cd
+                          FROM domrf_kn_objects
+                         WHERE obj_id = s.obj_id
+                         ORDER BY snapshot_date DESC
+                         LIMIT 1
+                  ) o ON TRUE
+                 WHERE o.region_cd = :region_code
+                   AND s.zscore <= :neg_zscore
+                   AND s.drop_pct <= :neg_drop_pct
+                 ORDER BY s.zscore ASC
+                 LIMIT :limit
+                """
+            ),
+            {
+                "snap": snapshot,
+                "lookback_interval": f"{lookback_months} months",
+                "recent_window": _RECENT_WINDOW,
+                "min_months": _MIN_MONTHS_HISTORY,
+                "min_prior_obs": _MIN_PRIOR_OBS,
+                "region_code": region_code,
+                "neg_zscore": -abs(min_zscore),
+                "neg_drop_pct": -abs(min_drop_pct),
+                "limit": limit,
+            },
+        )
+        .mappings()
+        .all()
+    )
+
+    alerts: list[dict[str, Any]] = []
+    for r in rows:
+        z = _f(r["zscore"])
+        drop = _f(r["drop_pct"])
+        severity = (
+            "high"
+            if (z is not None and z <= -_HIGH_ZSCORE)
+            or (drop is not None and drop <= -_HIGH_DROP_PCT)
+            else "medium"
+        )
+        alerts.append(
+            {
+                "obj_id": int(r["obj_id"]),
+                "comm_name": r["comm_name"],
+                "dev_name": r["dev_name"],
+                "recent_velocity_pm": round(_f(r["recent_mean"]) or 0.0, 2),
+                "prior_velocity_pm": round(_f(r["prior_mean"]) or 0.0, 2),
+                "zscore": round(z, 2) if z is not None else None,
+                "drop_pct": round(drop, 1) if drop is not None else None,
+                "severity": severity,
+            }
+        )
+
+    logger.info(
+        "velocity_alerts: snapshot=%s region=%d lookback=%dmo z>=%.1f drop>=%.0f%% -> %d alerts",
+        snapshot,
+        region_code,
+        lookback_months,
+        min_zscore,
+        min_drop_pct,
+        len(alerts),
+    )
+
+    return {
+        "snapshot_date": snapshot.isoformat() if snapshot else None,
+        "region_code": region_code,
+        "params": {
+            "lookback_months": lookback_months,
+            "min_zscore": min_zscore,
+            "min_drop_pct": min_drop_pct,
+        },
+        "alerts": alerts,
+    }

backend/tests/sql/test_ddu_price_indicator.py

@@ -0,0 +1,192 @@
+"""Integration test for Issue #99 — ДДУ price indicator SQL logic.
+
+Builds a synthetic temp table mirroring rosreestr_deals and runs the exact
+bucketing + index computation from migration 152 / ddu_price_indicator against
+a real PostgreSQL. Proves:
+
+1. Per-unit area bucketing uses area/deal_count (packaged ДДУ), not raw area.
+2. index_basis is median / first-period-median; index_previous is median /
+   previous-PRESENT-period median (min_deals=10 gates noisy quarters out).
+3. prev_period_value honestly records which period index_previous compares to
+   (may NOT be the literal previous quarter if one was filtered out).
+
+psycopg v3 only. CAST(:x AS type) everywhere. Skips cleanly off-CI.
+"""
+
+import os
+from decimal import Decimal
+
+import psycopg
+import pytest
+
+
+def _get_dsn() -> str:
+    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
+        "DATABASE_URL",
+        "postgresql://gendesign@localhost:15432/gendesign",
+    )
+    return raw.replace("+psycopg", "")
+
+
+def _db_reachable() -> tuple[bool, str]:
+    try:
+        with psycopg.connect(_get_dsn(), connect_timeout=3):
+            return True, ""
+    except Exception as e:
+        return False, str(e)
+
+
+_DB_OK, _DB_ERR = _db_reachable()
+pytestmark = pytest.mark.skipif(
+    not _DB_OK,
+    reason=(
+        "Нет доступной postgres БД (TEST_DATABASE_URL/DATABASE_URL) — "
+        f"тест #99 пропущен: {_DB_ERR}"
+    ),
+)
+
+
+# Core indicator query under test — mirrors migration 152 (reads temp table rd).
+_INDICATOR_SQL = """
+WITH per_unit AS (
+    SELECT date_trunc('quarter', period_start_date)::date AS quarter_start,
+           (area / deal_count) AS area_per_unit,
+           CASE WHEN price_per_sqm IS NOT NULL THEN price_per_sqm
+                WHEN deal_price IS NOT NULL AND area > 0 THEN deal_price/area END AS price_m2
+      FROM rd
+     WHERE region_code = 66 AND realestate_type_code = '002001003000'
+       AND doc_type = 'ДДУ' AND deal_count > 0 AND area > 0
+       AND period_start_date >= DATE '2025-04-01'
+),
+filtered AS (
+    SELECT quarter_start, area_per_unit, price_m2 FROM per_unit
+     WHERE area_per_unit BETWEEN 10 AND 300 AND price_m2 BETWEEN 30000 AND 800000
+),
+bucketed AS (
+    SELECT quarter_start, price_m2,
+           CASE WHEN area_per_unit < 25 THEN 1 WHEN area_per_unit < 40 THEN 2
+                WHEN area_per_unit < 60 THEN 3 WHEN area_per_unit < 80 THEN 4
+                WHEN area_per_unit < 100 THEN 5 ELSE 6 END AS area_bucket
+      FROM filtered
+    UNION ALL
+    SELECT quarter_start, price_m2, 0 FROM filtered
+),
+agg AS (
+    SELECT area_bucket, quarter_start, COUNT(*) AS deals_count,
+           PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_m2) AS med
+      FROM bucketed GROUP BY area_bucket, quarter_start
+    HAVING COUNT(*) >= 10
+),
+indexed AS (
+    SELECT area_bucket, quarter_start, deals_count, med,
+           FIRST_VALUE(med) OVER w AS basis_med,
+           LAG(med) OVER w AS prev_med,
+           LAG(quarter_start) OVER w AS prev_q
+      FROM agg WINDOW w AS (PARTITION BY area_bucket ORDER BY quarter_start)
+)
+SELECT area_bucket,
+       (EXTRACT(YEAR FROM quarter_start)::int::text || '-Q'
+            || EXTRACT(QUARTER FROM quarter_start)::int::text) AS period_value,
+       deals_count,
+       ROUND((med / NULLIF(basis_med, 0))::numeric, 4) AS index_basis,
+       ROUND((med / NULLIF(prev_med, 0))::numeric, 4) AS index_previous,
+       CASE WHEN prev_q IS NOT NULL THEN
+            (EXTRACT(YEAR FROM prev_q)::int::text || '-Q'
+                || EXTRACT(QUARTER FROM prev_q)::int::text) END AS prev_period_value
+  FROM indexed
+ ORDER BY area_bucket, period_value
+"""
+
+
+@pytest.fixture()
+def conn():
+    with psycopg.connect(_get_dsn(), autocommit=False) as c:
+        yield c
+        c.rollback()
+
+
+def _insert_quarter(cur, q_start, bucket_area, price_m2, n) -> None:
+    """Insert n single-flat ДДУ rows at given per-unit area + price/m²."""
+    rows = [
+        ("002001003000", "ДДУ", 66, q_start, bucket_area, 1, price_m2)
+        for _ in range(n)
+    ]
+    cur.executemany(
+        "INSERT INTO rd (realestate_type_code, doc_type, region_code, "
+        "period_start_date, area, deal_count, price_per_sqm) "
+        "VALUES (%s, %s, %s, %s, %s, %s, %s)",
+        rows,
+    )
+
+
+def _setup(cur: psycopg.Cursor) -> None:
+    cur.execute(
+        """
+        CREATE TEMP TABLE rd (
+            realestate_type_code text, doc_type varchar, region_code smallint,
+            period_start_date date, area numeric, deal_count int, price_per_sqm numeric,
+            deal_price numeric
+        ) ON COMMIT DROP;
+        """
+    )
+    # Bucket 3 (40-60 m²): three quarters, rising median, all >=10 deals.
+    _insert_quarter(cur, "2025-04-01", 50, 100000, 12)  # basis
+    _insert_quarter(cur, "2025-07-01", 50, 110000, 12)  # +10% vs basis & prev
+    _insert_quarter(cur, "2025-10-01", 50, 121000, 12)  # +10% vs prev
+    # Bucket 4 (60-80 m²): 2025-Q3 present, 2025-Q4 SPARSE (<10 → filtered),
+    # 2026-Q1 present. index_previous for 2026-Q1 must compare to 2025-Q3.
+    _insert_quarter(cur, "2025-07-01", 70, 150000, 11)
+    _insert_quarter(cur, "2025-10-01", 70, 999999, 3)   # below min_deals → dropped
+    _insert_quarter(cur, "2026-01-01", 70, 165000, 11)
+    # Packaged-deal trap: one row area=350 deal_count=7 → per-unit 50 m² (bucket 3),
+    # NOT bucket 6. Price chosen mid-range so it doesn't move the median much.
+    cur.execute(
+        "INSERT INTO rd (realestate_type_code, doc_type, region_code, "
+        "period_start_date, area, deal_count, price_per_sqm) "
+        "VALUES ('002001003000','ДДУ',66,'2025-04-01',350,7,100000)"
+    )
+
+
+def _rows_by_key(rows) -> dict[tuple[int, str], tuple]:
+    return {(r[0], r[1]): r for r in rows}
+
+
+def test_basis_and_previous_index(conn: psycopg.Connection) -> None:
+    cur = conn.cursor()
+    _setup(cur)
+    cur.execute(_INDICATOR_SQL)
+    by = _rows_by_key(cur.fetchall())
+    # Bucket 3 rising 100k→110k→121k.
+    assert by[(3, "2025-Q2")][3] == Decimal("1.0000")  # basis self
+    assert by[(3, "2025-Q3")][4] == Decimal("1.1000")  # prev: 110/100
+    assert by[(3, "2025-Q4")][4] == Decimal("1.1000")  # prev: 121/110
+    assert by[(3, "2025-Q4")][3] == Decimal("1.2100")  # basis: 121/100
+
+
+def test_packaged_deal_bucketed_by_per_unit_area(conn: psycopg.Connection) -> None:
+    """area=350/deal_count=7 → 50 m² → bucket 3, never bucket 6."""
+    cur = conn.cursor()
+    _setup(cur)
+    cur.execute(_INDICATOR_SQL)
+    by = _rows_by_key(cur.fetchall())
+    # Bucket 6 should have NO rows (the only 100+ candidate was the packaged
+    # row, which correctly lands in bucket 3).
+    assert not any(k[0] == 6 for k in by), "packaged ДДУ leaked into 100+ bucket"
+    # Bucket 3 basis quarter deal count includes the 7-unit packaged row.
+    assert by[(3, "2025-Q2")][2] >= 12
+
+
+def test_prev_period_value_skips_filtered_quarter(conn: psycopg.Connection) -> None:
+    """Bucket 4: 2025-Q4 is below min_deals and dropped; 2026-Q1's
+    index_previous must compare against 2025-Q3 (honest prev_period_value),
+    not the literal previous quarter.
+    """
+    cur = conn.cursor()
+    _setup(cur)
+    cur.execute(_INDICATOR_SQL)
+    by = _rows_by_key(cur.fetchall())
+    assert (4, "2025-Q4") not in by, "sparse quarter should be filtered by min_deals"
+    q1 = by[(4, "2026-Q1")]
+    assert q1[5] == "2025-Q3", f"prev_period_value must skip dropped Q4, got {q1[5]}"
+    # 165000 / 150000 = 1.1000
+    assert q1[4] == Decimal("1.1000")

backend/tests/sql/test_velocity_alerts.py

@@ -0,0 +1,154 @@
+"""Integration test for Issue #17 — velocity-anomaly detection SQL logic.
+
+Builds synthetic temp tables mirroring domrf_kn_sale_graph + domrf_kn_objects
+and runs the exact windowing / z-score query from
+``app.services.analytics.velocity_alerts`` against a real PostgreSQL. Proves:
+
+1. A ЖК with a sharp recent drop is flagged (z <= -2.0 AND drop_pct <= -30%).
+2. A ЖК with stable velocity is NOT flagged (no false positive).
+3. The lookback window anchors to the latest report_month (not snapshot_date),
+   so a stale scrape date does not starve the prior window.
+
+Uses psycopg v3 (never psycopg2). All bind params use CAST(:x AS type).
+Skips cleanly off-CI when no Postgres is reachable.
+"""
+
+import os
+
+import psycopg
+import pytest
+
+
+def _get_dsn() -> str:
+    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
+        "DATABASE_URL",
+        "postgresql://gendesign@localhost:15432/gendesign",
+    )
+    return raw.replace("+psycopg", "")
+
+
+def _db_reachable() -> tuple[bool, str]:
+    try:
+        with psycopg.connect(_get_dsn(), connect_timeout=3):
+            return True, ""
+    except Exception as e:
+        return False, str(e)
+
+
+_DB_OK, _DB_ERR = _db_reachable()
+pytestmark = pytest.mark.skipif(
+    not _DB_OK,
+    reason=(
+        "Нет доступной postgres БД (TEST_DATABASE_URL/DATABASE_URL) — "
+        f"тест #17 пропущен: {_DB_ERR}"
+    ),
+)
+
+
+# Core query under test — kept byte-identical in spirit to velocity_alerts.py.
+# Reads from temp tables sg (sale_graph) + obj (objects).
+_DETECT_SQL = """
+WITH anchor AS (
+    SELECT MAX(report_month) AS max_month
+      FROM sg WHERE type = 'apartments' AND snapshot_date = %(snap)s
+),
+series AS (
+    SELECT g.obj_id, g.report_month, g.realised
+      FROM sg g CROSS JOIN anchor a
+     WHERE g.type = 'apartments' AND g.snapshot_date = %(snap)s
+       AND g.report_month > (a.max_month - CAST(%(lookback)s AS interval))
+),
+ranked AS (
+    SELECT obj_id, realised,
+           ROW_NUMBER() OVER (PARTITION BY obj_id ORDER BY report_month DESC) AS rn,
+           COUNT(*) OVER (PARTITION BY obj_id) AS n_months
+      FROM series
+),
+windows AS (
+    SELECT obj_id, n_months,
+           AVG(realised) FILTER (WHERE rn <= 3) AS recent_mean,
+           AVG(realised) FILTER (WHERE rn > 3)  AS prior_mean,
+           STDDEV_SAMP(realised) FILTER (WHERE rn > 3) AS prior_std,
+           COUNT(*) FILTER (WHERE rn > 3) AS prior_n
+      FROM ranked GROUP BY obj_id, n_months
+),
+scored AS (
+    SELECT w.obj_id, w.recent_mean, w.prior_mean,
+           CASE WHEN w.prior_std > 0 THEN (w.recent_mean - w.prior_mean)/w.prior_std END AS z,
+           CASE WHEN w.prior_mean > 0
+                THEN (w.recent_mean - w.prior_mean)/w.prior_mean*100.0 END AS drop_pct
+      FROM windows w
+     WHERE w.n_months >= 6 AND w.prior_n >= 3 AND w.prior_mean > 0 AND w.prior_std > 0
+)
+SELECT s.obj_id, ROUND(s.z::numeric, 2) AS z, ROUND(s.drop_pct::numeric, 1) AS drop_pct
+  FROM scored s
+  JOIN LATERAL (SELECT region_cd FROM obj WHERE obj_id = s.obj_id LIMIT 1) o ON TRUE
+ WHERE o.region_cd = 66 AND s.z <= -2.0 AND s.drop_pct <= -30.0
+ ORDER BY s.z ASC
+"""
+
+
+@pytest.fixture()
+def conn():
+    with psycopg.connect(_get_dsn(), autocommit=False) as c:
+        yield c
+        c.rollback()
+
+
+def _setup(cur: psycopg.Cursor) -> None:
+    cur.execute(
+        """
+        CREATE TEMP TABLE sg (
+            obj_id bigint, report_month date, type text,
+            realised int, snapshot_date date
+        ) ON COMMIT DROP;
+        CREATE TEMP TABLE obj (obj_id bigint, region_cd int) ON COMMIT DROP;
+        """
+    )
+    snap = "2026-04-28"  # stale scrape date; data months end 2025-12 (4-mo gap)
+    months = [
+        "2025-04-01", "2025-05-01", "2025-06-01", "2025-07-01", "2025-08-01",
+        "2025-09-01", "2025-10-01", "2025-11-01", "2025-12-01",
+    ]
+    # obj 1 — sharp drop: prior ~15/mo, recent ~4/mo  -> alert
+    dropper = [16, 14, 15, 17, 13, 14, 5, 4, 3]
+    # obj 2 — stable ~10/mo  -> NO alert
+    stable = [10, 9, 11, 10, 12, 9, 10, 11, 10]
+    rows = []
+    for m, v in zip(months, dropper, strict=True):
+        rows.append((1, m, "apartments", v, snap))
+    for m, v in zip(months, stable, strict=True):
+        rows.append((2, m, "apartments", v, snap))
+    cur.executemany(
+        "INSERT INTO sg (obj_id, report_month, type, realised, snapshot_date) "
+        "VALUES (%s, %s, %s, %s, %s)",
+        rows,
+    )
+    cur.executemany("INSERT INTO obj (obj_id, region_cd) VALUES (%s, %s)", [(1, 66), (2, 66)])
+
+
+def test_sharp_drop_is_flagged(conn: psycopg.Connection) -> None:
+    cur = conn.cursor()
+    _setup(cur)
+    cur.execute(_DETECT_SQL, {"snap": "2026-04-28", "lookback": "9 months"})
+    alerts = cur.fetchall()
+    flagged_ids = {r[0] for r in alerts}
+    # obj 1 (dropper) must be flagged; obj 2 (stable) must not.
+    assert 1 in flagged_ids, f"expected dropper flagged, got {alerts}"
+    assert 2 not in flagged_ids, f"stable obj must not be a false positive, got {alerts}"
+    # And the drop must be a strong negative z-score with a material drop%.
+    dropper_row = next(r for r in alerts if r[0] == 1)
+    assert dropper_row[1] <= -2.0  # z
+    assert dropper_row[2] <= -30.0  # drop_pct
+
+
+def test_lookback_anchors_to_latest_data_month(conn: psycopg.Connection) -> None:
+    """If the window anchored to snapshot_date (2026-04-28) instead of the
+    latest report_month (2025-12), the 4-month gap would trim the prior window
+    below prior_n>=3 and the dropper would NOT be flagged. This guards the fix.
+    """
+    cur = conn.cursor()
+    _setup(cur)
+    cur.execute(_DETECT_SQL, {"snap": "2026-04-28", "lookback": "9 months"})
+    flagged_ids = {r[0] for r in cur.fetchall()}
+    assert 1 in flagged_ids, "anchor-to-latest-month regression: dropper lost"

data/sql/152_mv_ddu_price_indicator.sql

@@ -0,0 +1,177 @@
+-- 152_mv_ddu_price_indicator.sql
+-- Issue #99 — «Индикатор цен ДДУ» (ARN-framework applied to primary market).
+--
+-- ARN (НСПД /api/market-analytics) считает индикатор цен ВТОРИЧНОГО рынка.
+-- У нас в rosreestr_deals — ПЕРВИЧКА (ДДУ). Это дополнение их фреймворка:
+-- тот же «индикатор = median(price/m²) тек.периода / median базисного периода»,
+-- но по новостройкам.
+--
+-- ── MVP SCOPE / честные упрощения (см. issue #99 «Risks / open») ───────────
+-- Полный ARN: 4 indicator-типа (M/Q/H/Y) × 6 area buckets × 3 calc methods ×
+-- 4 уровня гранулярности (Округ→Субъект→Нас.пункт→Cad_quarter). Реализовано
+-- НЕ всё — данных по первичке хватает только на часть матрицы:
+--
+--   1. ГРАНУЛЯРНОСТЬ = СУБЪЕКТ (region_code=66), НЕ cad_quarter.
+--      Per-cad_quarter × area_bucket для ДДУ безнадёжно разрежено: после смены
+--      агрегации rosreestr (2025Q1) на квартал приходится ~190-220 строк ИТОГО
+--      по всей области; на отдельный cad_quarter — единицы. Субъект-уровень —
+--      единственный статистически устойчивый уровень для первички.
+--      (Per-quarter price ОТДЕЛЬНО уже есть в mv_quarter_price_index — там
+--      кросс-секционный индекс «квартал vs город», другая метрика.)
+--
+--   2. PERIOD_TYPE = 'Q' (квартал). M/H/Y не делаем:
+--      - M (месяц): rosreestr публикует данные поквартально (period_start_date —
+--        начало квартала), месячной гранулярности в источнике НЕТ.
+--      - H/Y: см. п.3 — слишком мало сопоставимых периодов.
+--
+--   3. ОКНО = 2025-Q2 .. now. ЖЁСТКИЙ обрыв на 2025-Q1:
+--      rosreestr с 2025Q1 резко увеличил агрегацию (1 строка = пакет из N ДДУ,
+--      area=SUM, deal_count=N). Сырых строк до 2025Q1: 700-2400/кв; с 2025Q2:
+--      ~190-220/кв. Медиана price/m² скачет 66K→154K на границе 2024Q4→2025Q2
+--      ЧИСТО из-за смены формата, не из-за инфляции. Поэтому периоды до 2025Q2
+--      исключены — иначе basis/prev-индикатор даёт фейковые 1.4-1.6.
+--      => calc_method 3 (год назад) НЕ реализуем: нет сопоставимого 2025Q1-.
+--
+--   4. CALC METHODS: 1 = basis (к первому периоду окна = 2025-Q2),
+--                    2 = previous (к предыдущему ПРИСУТСТВУЮЩЕМУ периоду).
+--      method 3 (год назад) — BLOCKED (нет данных, см. п.3).
+--
+--   5. AREA_BUCKET: 1..6 по ARN (0-25/25-40/40-60/60-80/80-100/100+ м²)
+--      на ПОКВАРТИРНОЙ площади (area/deal_count — пакетные ДДУ!), плюс
+--      bucket 0 = ALL (headline, все площади вместе).
+--
+--   6. MIN_DEALS = 10 на (bucket, quarter): ниже порога медиана шумная →
+--      строка не создаётся (HAVING). prev/basis сравнивают только присутствующие
+--      периоды → prev_period_value хранится явно (может быть не «−1 квартал»).
+--
+-- ── VALIDATION (prod, 2026-06, source rosreestr_deals) ─────────────────────
+--   ALL-buckets (bucket=0), method=previous:
+--     2025-Q2 (basis) → 2025-Q3 1.0026 → 2025-Q4 1.0501 → 2026-Q1 1.0185
+--   ARN benchmark Екатеринбург Q1 2026 (вторичка) = 1.03.
+--   Наш Q1 2026 (первичка) = 1.0185 → within ±5% (issue acceptance). ✓
+--   Расхождение ожидаемо: primary vs secondary (объяснять в UI tooltip).
+--
+-- Источник: rosreestr_deals (партиционированная, фильтр realestate_type_code=
+--   '002001003000' новостройки + doc_type='ДДУ'). Прямой запрос к таблице
+--   (НЕ через mv_quarter_price_per_m2: тому MV нужна area_bucket-разбивка,
+--   которой там нет — он агрегирует по cad_quarter без поквартирной площади).
+--
+-- REFRESH CONCURRENTLY: UNIQUE индекс создаётся сразу после CREATE WITH NO DATA.
+-- Первый populate — non-concurrent (MV пуст). Re-apply idempotent: DROP ... CASCADE.
+-- _schema_migrations (deploy.yml) предотвращает повторный apply на prod.
+-- Refresh helper: backend/app/services/analytics/ddu_price_indicator.py (не в beat).
+
+BEGIN;
+
+DROP MATERIALIZED VIEW IF EXISTS mv_ddu_price_indicator CASCADE;
+
+CREATE MATERIALIZED VIEW mv_ddu_price_indicator AS
+WITH per_unit AS (
+    -- Поквартирная площадь и цена/м². Пакетные ДДУ: area=SUM, deal_count=N,
+    -- поэтому area_per_unit = area/deal_count (иначе пакет из 5 студий
+    -- попадает в bucket «80+»). price_per_sqm pre-computed; fallback deal_price/area.
+    SELECT
+        date_trunc('quarter', period_start_date)::date AS quarter_start,
+        (area / deal_count)                            AS area_per_unit,
+        CASE
+            WHEN price_per_sqm IS NOT NULL           THEN price_per_sqm
+            WHEN deal_price IS NOT NULL AND area > 0 THEN deal_price / area
+            ELSE NULL
+        END                                            AS price_m2
+    FROM rosreestr_deals
+    WHERE region_code = 66
+      AND realestate_type_code = '002001003000'  -- квартиры в новостройках
+      AND doc_type = 'ДДУ'
+      AND deal_count > 0
+      AND area > 0
+      -- Жёсткий обрыв формата (см. шапку п.3): окно строго с 2025-Q2.
+      AND period_start_date >= DATE '2025-04-01'
+),
+filtered AS (
+    SELECT quarter_start, area_per_unit, price_m2
+    FROM per_unit
+    WHERE area_per_unit BETWEEN 10 AND 300        -- реалистичная 1-квартира
+      AND price_m2 BETWEEN 30000 AND 800000       -- срез выбросов (как в #33 MV)
+),
+bucketed AS (
+    -- Каждую сделку дублируем в её bucket (1..6) И в bucket 0 (ALL/headline)
+    -- через UNION ALL, чтобы один GROUP BY посчитал и разбивку, и агрегат.
+    SELECT quarter_start, price_m2,
+           CASE
+               WHEN area_per_unit < 25  THEN 1
+               WHEN area_per_unit < 40  THEN 2
+               WHEN area_per_unit < 60  THEN 3
+               WHEN area_per_unit < 80  THEN 4
+               WHEN area_per_unit < 100 THEN 5
+               ELSE 6
+           END AS area_bucket
+    FROM filtered
+    UNION ALL
+    SELECT quarter_start, price_m2, 0 AS area_bucket
+    FROM filtered
+),
+agg AS (
+    SELECT
+        area_bucket,
+        quarter_start,
+        COUNT(*)                                                AS deals_count,
+        PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price_m2)   AS median_price_per_m2
+    FROM bucketed
+    GROUP BY area_bucket, quarter_start
+    HAVING COUNT(*) >= 10        -- min_deals: ниже порога медиана шумная
+),
+indexed AS (
+    SELECT
+        area_bucket,
+        quarter_start,
+        deals_count,
+        median_price_per_m2,
+        -- basis = первый присутствующий квартал окна для этого bucket
+        FIRST_VALUE(median_price_per_m2) OVER w_asc  AS basis_median,
+        -- previous = предыдущий ПРИСУТСТВУЮЩИЙ квартал (после HAVING-фильтра)
+        LAG(median_price_per_m2)  OVER w_asc         AS prev_median,
+        LAG(quarter_start)        OVER w_asc         AS prev_quarter_start
+    FROM agg
+    WINDOW w_asc AS (PARTITION BY area_bucket ORDER BY quarter_start)
+)
+SELECT
+    area_bucket,
+    -- period_value в ARN-стиле: '2026-Q1'
+    (EXTRACT(YEAR FROM quarter_start)::int::text || '-Q'
+        || EXTRACT(QUARTER FROM quarter_start)::int::text)      AS period_value,
+    quarter_start,
+    deals_count,
+    median_price_per_m2::numeric(12, 2)                         AS median_price_per_m2,
+    -- calc_method 1: basis (к первому периоду окна)
+    (median_price_per_m2 / NULLIF(basis_median, 0))::numeric(8, 4) AS index_basis,
+    -- calc_method 2: previous (к предыдущему присутствующему периоду)
+    (median_price_per_m2 / NULLIF(prev_median, 0))::numeric(8, 4)  AS index_previous,
+    -- какой период реально сравнивается в index_previous (может быть не −1 кв,
+    -- если промежуточный quarter не прошёл HAVING) — честность для UI tooltip
+    CASE WHEN prev_quarter_start IS NOT NULL THEN
+        (EXTRACT(YEAR FROM prev_quarter_start)::int::text || '-Q'
+            || EXTRACT(QUARTER FROM prev_quarter_start)::int::text)
+    END                                                         AS prev_period_value,
+    'Q'::text                                                   AS period_type,
+    NOW()                                                       AS computed_at
+FROM indexed
+WITH NO DATA;
+
+-- UNIQUE индекс для REFRESH CONCURRENTLY + O(1) lookup эндпойнтом.
+CREATE UNIQUE INDEX IF NOT EXISTS mv_ddu_price_indicator_pk
+    ON mv_ddu_price_indicator (area_bucket, period_value);
+
+-- Первый populate (non-concurrent — MV пуст, CONCURRENTLY на пустом запрещён).
+REFRESH MATERIALIZED VIEW mv_ddu_price_indicator;
+
+COMMENT ON MATERIALIZED VIEW mv_ddu_price_indicator IS
+    'ARN-style ценовой индикатор ПЕРВИЧНОГО рынка (ДДУ) по кварталам × area_bucket. '
+    'Source: rosreestr_deals (region 66, realestate_type_code=002001003000, doc_type=ДДУ). '
+    'MVP scope (issue #99): granularity=субъект (НЕ cad_quarter — первичка разрежена), '
+    'period_type=Q, окно с 2025-Q2 (обрыв формата rosreestr на 2025-Q1), '
+    'calc methods 1=basis / 2=previous (method 3 год-назад BLOCKED: нет данных). '
+    'area_bucket 0=ALL(headline)/1..6=ARN-площади на поквартирной area/deal_count. '
+    'min_deals=10/строка. Q1-2026 headline=1.0185 vs ARN-вторичка 1.03 (±5%, primary vs secondary). '
+    'Refresh: ddu_price_indicator.refresh_ddu_price_indicator() (не в beat). Issue #99.';
+
+COMMIT;