lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

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

Merged
bot-backend merged 1 commit from feat/analytics-velocity-alerts-ddu-indicator into main 2026-06-13 17:14:10 +00:00
Conversation 0 Commits 1 Files changed 10 +1228
10 changed files with 1228 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
backend/app/api/v1/analytics.py
View file

@ -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 -------------------------------------------------------

50
backend/app/api/v1/market.py Normal file
View file

@ -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)

2
backend/app/main.py
View file

@ -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"])

47
backend/app/schemas/market.py Normal file
View file

@ -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]

6
backend/app/services/analytics/__init__.py Normal file
View file

@ -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).
"""

278
backend/app/services/analytics/ddu_price_indicator.py Normal file
View file

@ -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

292
backend/app/services/analytics/velocity_alerts.py Normal file
View file

@ -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,
}

192
backend/tests/sql/test_ddu_price_indicator.py Normal file
View file

@ -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")

154
backend/tests/sql/test_velocity_alerts.py Normal file
View file

@ -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"

177
data/sql/152_mv_ddu_price_indicator.sql Normal file
View file

@ -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;