From e1a2a779e467a38512e5c3e0309057f0e9605ad8 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 16 May 2026 14:06:57 +0300 Subject: [PATCH] feat(analytics): mv_quarter_price_per_m2 MV + refresh helper (#33 D1 PR A) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../site_finder/quarter_price_refresh.py | 63 +++++++++ data/sql/95_mv_quarter_price.sql | 124 ++++++++++++++++++ 2 files changed, 187 insertions(+) create mode 100644 backend/app/services/site_finder/quarter_price_refresh.py create mode 100644 data/sql/95_mv_quarter_price.sql diff --git a/backend/app/services/site_finder/quarter_price_refresh.py b/backend/app/services/site_finder/quarter_price_refresh.py new file mode 100644 index 00000000..97e83891 --- /dev/null +++ b/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 diff --git a/data/sql/95_mv_quarter_price.sql b/data/sql/95_mv_quarter_price.sql new file mode 100644 index 00000000..9e5d5c03 --- /dev/null +++ b/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;