diff --git a/backend/app/api/v1/analytics.py b/backend/app/api/v1/analytics.py index ada580c6..e748737c 100644 --- a/backend/app/api/v1/analytics.py +++ b/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 ------------------------------------------------------- diff --git a/backend/app/api/v1/market.py b/backend/app/api/v1/market.py new file mode 100644 index 00000000..b2204523 --- /dev/null +++ b/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) diff --git a/backend/app/main.py b/backend/app/main.py index 2604c737..82910f2b 100644 --- a/backend/app/main.py +++ b/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"]) diff --git a/backend/app/schemas/market.py b/backend/app/schemas/market.py new file mode 100644 index 00000000..9e4cde1e --- /dev/null +++ b/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] diff --git a/backend/app/services/analytics/__init__.py b/backend/app/services/analytics/__init__.py new file mode 100644 index 00000000..bb713590 --- /dev/null +++ b/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). +""" diff --git a/backend/app/services/analytics/ddu_price_indicator.py b/backend/app/services/analytics/ddu_price_indicator.py new file mode 100644 index 00000000..3958d4b2 --- /dev/null +++ b/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 diff --git a/backend/app/services/analytics/velocity_alerts.py b/backend/app/services/analytics/velocity_alerts.py new file mode 100644 index 00000000..61160284 --- /dev/null +++ b/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, + } diff --git a/backend/tests/sql/test_ddu_price_indicator.py b/backend/tests/sql/test_ddu_price_indicator.py new file mode 100644 index 00000000..df9bd034 --- /dev/null +++ b/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") diff --git a/backend/tests/sql/test_velocity_alerts.py b/backend/tests/sql/test_velocity_alerts.py new file mode 100644 index 00000000..44b34bfe --- /dev/null +++ b/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" diff --git a/data/sql/152_mv_ddu_price_indicator.sql b/data/sql/152_mv_ddu_price_indicator.sql new file mode 100644 index 00000000..6fdef0be --- /dev/null +++ b/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;