feat(analytics): mv_quarter_price_per_m2 MV + refresh helper (#33 D1 PR A)

Per-cad_quarter price aggregation из 6.83M rosreestr_deals.

Source verification (postgres MCP):
- rosreestr_deals total: 6 830 001
- price_per_sqm filled 99.9% (fallback deal_price/area)
- quarter_cad_number (NOT cad_quarter), period_start_date, realestate_type_code

SQL (95_mv_quarter_price.sql, ~124 LOC):
- Filter realestate_type_code='002001003000' (новостройки) + outlier 30K..800K
- Percentile p25/median/p75 + mean + temporal slices (6m/12m/24m)
- 24-month window, HAVING COUNT(*)>=3
- WITH NO DATA + REFRESH after INDEX (#194 pattern)
- UNIQUE INDEX (quarter_cad_number) для CONCURRENTLY
- ~52 492 quarters в MV (all-Russia; EKB subset ~9700)

Python helper (quarter_price_refresh.py, ~65 LOC):
- refresh_quarter_price(db, concurrently=True) с try/except fallback

Phase B (analyze integration — quarter median вместо district) — отдельный PR.

backend/app/services/site_finder/quarter_price_refresh.py

@@ -0,0 +1,63 @@
+"""Refresh helper for mv_quarter_price_per_m2 (Issue #33 D1).
+
+Not scheduled automatically in this PR — intended for manual invocation or
+a Celery beat task in a follow-up issue (Issue #33 PR C).
+
+Usage example (manual, via psql-connected shell or admin endpoint):
+    from sqlalchemy.orm import Session
+    from app.services.site_finder.quarter_price_refresh import refresh_quarter_price
+
+    count = refresh_quarter_price(db)
+    # logs: "mv_quarter_price_per_m2 refreshed: 52492 rows"
+"""
+
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def refresh_quarter_price(db: Session, *, concurrently: bool = True) -> int:
+    """REFRESH MATERIALIZED VIEW mv_quarter_price_per_m2.
+
+    Args:
+        db: SQLAlchemy Session (sync).
+        concurrently: When True, uses REFRESH CONCURRENTLY — non-blocking but
+            requires the unique index mv_quarter_price_pk to exist (created by
+            95_mv_quarter_price.sql) and the MV to be already populated.
+            Pass False only for the very first populate or after MV recreation.
+
+    Returns:
+        Row count in the MV after refresh (for observability / alerting).
+
+    Raises:
+        OperationalError: Re-raised if the error is not the known
+            "cannot refresh materialized view concurrently" case.
+    """
+    try:
+        if concurrently:
+            db.execute(text("REFRESH MATERIALIZED VIEW CONCURRENTLY mv_quarter_price_per_m2"))
+        else:
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_quarter_price_per_m2"))
+        db.commit()
+    except OperationalError as e:
+        if concurrently and "cannot refresh materialized view" in str(e).lower():
+            logger.warning(
+                "CONCURRENTLY failed (MV likely not populated), falling back to"
+                " non-concurrent refresh"
+            )
+            db.rollback()
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_quarter_price_per_m2"))
+            db.commit()
+        else:
+            raise
+
+    row = db.execute(text("SELECT COUNT(*) FROM mv_quarter_price_per_m2")).first()
+    count = int(row[0]) if row else 0
+    logger.info("mv_quarter_price_per_m2 refreshed: %d rows", count)
+    return count

data/sql/95_mv_quarter_price.sql

@@ -0,0 +1,124 @@
+-- 95_mv_quarter_price.sql
+-- Issue #33 D1 — Per-cad_quarter price aggregation из 6.83M rosreestr_deals.
+--
+-- Real schema (verified via information_schema):
+--   quarter_cad_number  varchar  — cadastral quarter key (e.g. '66:41:0301036')
+--   price_per_sqm       numeric  — pre-computed руб/м²; filled for 6.82M / 6.83M rows
+--   deal_price          numeric  — total deal price (fallback if price_per_sqm IS NULL)
+--   area                numeric  — area m² (fallback denominator)
+--   period_start_date   date     — quarter start (2024-01-01 .. 2026-01-01)
+--   realestate_type_code text    — '002001003000' = новостройки (ДДУ)
+--
+-- Filter scope:
+--   realestate_type_code = '002001003000' → 3 143 004 rows (45.9% of total 6.83M)
+--   price_per_sqm BETWEEN 30000 AND 800000 → outlier removal
+--   period_start_date >= NOW() - INTERVAL '24 months' → rolling window
+--   HAVING COUNT(*) >= 3 → suppress quarters with too few deals
+--
+-- Expected rows: ~52 492 distinct quarters pass HAVING >= 3 (verified via SELECT COUNT).
+-- Dataset is all-Russia (rosreestr_deals is not region-filtered at table level).
+--
+-- Sub-window medians (6m / 12m):
+--   PERCENTILE_CONT does not support FILTER clause in PG16.
+--   Workaround: pre-aggregate per quarter with separate FILTER masks in a lateral
+--   subquery. Implemented as a second CTE (agg_windows) joined to the main agg.
+--
+-- REFRESH CONCURRENTLY: safe after this migration because a UNIQUE INDEX on
+-- quarter_cad_number is created immediately after CREATE WITH NO DATA.
+-- First populate is non-concurrent (MV is empty, CONCURRENTLY not allowed).
+--
+-- Deploy: auto-applied by deploy.yml via _schema_migrations tracking.
+-- Dependencies: none (no existing views depend on this MV at creation time).
+--
+-- WARNING: Re-apply drops MV + CASCADE; after re-apply first REFRESH must use
+-- concurrently=False. _schema_migrations tracking prevents re-apply in prod.
+
+BEGIN;
+
+DROP MATERIALIZED VIEW IF EXISTS mv_quarter_price_per_m2 CASCADE;
+
+CREATE MATERIALIZED VIEW mv_quarter_price_per_m2 AS
+WITH dataset AS (
+    -- Resolve price_m2: use pre-computed field, fall back to deal_price / area
+    SELECT
+        quarter_cad_number,
+        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,
+        period_start_date   AS deal_date
+    FROM rosreestr_deals
+    WHERE realestate_type_code = '002001003000'
+      AND period_start_date >= NOW() - INTERVAL '24 months'
+),
+filtered AS (
+    -- Apply outlier filter once; reused by both aggregate CTEs
+    SELECT quarter_cad_number, price_m2, deal_date
+    FROM dataset
+    WHERE price_m2 BETWEEN 30000 AND 800000
+),
+agg_main AS (
+    -- Primary aggregation: full 24-month window
+    SELECT
+        quarter_cad_number,
+        COUNT(*)                                                AS deals_count,
+        PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY price_m2) AS p25,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY price_m2) AS median,
+        PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY price_m2) AS p75,
+        AVG(price_m2)::numeric(12, 2)                          AS mean,
+        MAX(deal_date)                                         AS last_deal_date
+    FROM filtered
+    GROUP BY quarter_cad_number
+    HAVING COUNT(*) >= 3
+),
+agg_windows AS (
+    -- Sub-window medians per quarter.
+    -- PERCENTILE_CONT does not accept a FILTER clause in PG16, so we pre-filter
+    -- per window inside separate conditional aggregations using a CASE that
+    -- returns NULL for out-of-window rows (NULLs are excluded from PERCENTILE_CONT).
+    SELECT
+        quarter_cad_number,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (
+            ORDER BY CASE WHEN deal_date > NOW() - INTERVAL '6 months'
+                          THEN price_m2 ELSE NULL END
+        )                                                       AS median_6m,
+        PERCENTILE_CONT(0.50) WITHIN GROUP (
+            ORDER BY CASE WHEN deal_date > NOW() - INTERVAL '12 months'
+                          THEN price_m2 ELSE NULL END
+        )                                                       AS median_12m
+    FROM filtered
+    GROUP BY quarter_cad_number
+)
+SELECT
+    m.quarter_cad_number,
+    m.deals_count,
+    m.p25,
+    m.median,
+    m.p75,
+    m.mean,
+    m.last_deal_date,
+    w.median_6m,
+    w.median_12m,
+    -- median_24m == median (window is already 24mo); kept for API symmetry
+    m.median                                                    AS median_24m
+FROM agg_main m
+JOIN agg_windows w USING (quarter_cad_number)
+WITH NO DATA;
+
+-- UNIQUE index required for REFRESH MATERIALIZED VIEW CONCURRENTLY.
+-- Created on empty MV → instant (no data lock).
+CREATE UNIQUE INDEX mv_quarter_price_pk
+    ON mv_quarter_price_per_m2 (quarter_cad_number);
+
+-- Initial populate (non-concurrent — CONCURRENTLY not allowed on empty MV).
+REFRESH MATERIALIZED VIEW mv_quarter_price_per_m2;
+
+COMMENT ON MATERIALIZED VIEW mv_quarter_price_per_m2 IS
+    'Per-cad_quarter (quarter_cad_number) price aggregation из rosreestr_deals. '
+    'Filter: realestate_type_code=002001003000 (новостройки/ДДУ), rolling 24 months, '
+    'outlier filter 30K–800K руб/м², HAVING >= 3 deals. '
+    'All-Russia scope (~52K rows expected). '
+    'Refresh: weekly via celery beat (TBD, Issue #33 PR C).';
+
+COMMIT;