feat(analytics): mv_layout_velocity MV + refresh helper (#113 PR B) (#194)

backend/app/services/site_finder/layout_velocity_refresh.py

@@ -0,0 +1,57 @@
+"""Refresh helper for mv_layout_velocity (Issue #113 PR B).
+
+Not scheduled automatically in this PR — intended for manual invocation or
+a Celery beat task in a follow-up issue.
+
+Usage example (manual, via psql-connected shell or admin endpoint):
+    from sqlalchemy.orm import Session
+    from app.services.site_finder.layout_velocity_refresh import refresh_layout_velocity
+
+    count = refresh_layout_velocity(db)
+    # logs: "mv_layout_velocity refreshed: 459 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_layout_velocity(db: Session, *, concurrently: bool = True) -> int:
+    """REFRESH MATERIALIZED VIEW mv_layout_velocity.
+
+    Args:
+        db: SQLAlchemy Session (sync).
+        concurrently: When True, uses REFRESH CONCURRENTLY — non-blocking but
+            requires the unique index mv_layout_velocity_pk to exist (created
+            by 94_mv_layout_velocity.sql).  Pass False only for the very first
+            populate or when the MV was just recreated.
+
+    Returns:
+        Row count in the MV after refresh (for observability / alerting).
+    """
+    try:
+        if concurrently:
+            db.execute(text("REFRESH MATERIALIZED VIEW CONCURRENTLY mv_layout_velocity"))
+        else:
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_layout_velocity"))
+        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"
+            )
+            db.rollback()
+            db.execute(text("REFRESH MATERIALIZED VIEW mv_layout_velocity"))
+            db.commit()
+        else:
+            raise
+    row = db.execute(text("SELECT COUNT(*) FROM mv_layout_velocity")).first()
+    count = int(row[0]) if row else 0
+    logger.info("mv_layout_velocity refreshed: %d rows", count)
+    return count

data/sql/94_mv_layout_velocity.sql

@@ -0,0 +1,90 @@
+-- 94_mv_layout_velocity.sql
+-- Issue #113 PR B — Materialized view per (obj_id, room_bucket) for top-layouts ranking.
+--
+-- Sources:
+--   objective_corpus_room_month  — monthly per-corpus-per-room deals (19 738 rows, 2025-05..2026-05)
+--   objective_complex_mapping    — project_name → domrf_obj_id mapping (114 mapped objects)
+--
+-- Join key:
+--   objective_corpus_room_month.project_name = objective_complex_mapping.objective_complex_name
+--   (verified: one project_name → at most one domrf_obj_id)
+--
+-- room_bucket normalisation:
+--   cyrillic 'студия' → ASCII 'studio' for consistency with layout_signature.py (PR A, Issue #113)
+--   values '1', '2', '3', '4+' are kept as-is
+--
+-- Window: last 24 months (rolling, anchored to NOW() at refresh time)
+-- Expected rows after aggregation: ~459 (obj_id × room_bucket combinations)
+--
+-- REFRESH CONCURRENTLY is safe after this migration because a UNIQUE index
+-- on (obj_id, room_bucket) is created immediately.
+-- Subsequent refreshes: layout_velocity_refresh.py helper (PR B) — not scheduled yet.
+--
+-- Deploy: auto-applied by deploy.yml via _schema_migrations tracking.
+-- Dependencies: none (no views depend on this MV at creation time).
+--
+-- WARN: Re-apply этого файла (DR / lost _schema_migrations entry / dev local)
+--       снесёт MV + CASCADE удалит зависимости (e.g. future /best-layouts views).
+--       После re-apply ПЕРВЫЙ refresh должен быть concurrently=False
+--       (CONCURRENTLY требует уже populated MV — упадёт на пустой).
+--       Tracking table `_schema_migrations` нормально предотвращает re-apply,
+--       но добавлен comment как safety reminder.
+
+BEGIN;
+
+DROP MATERIALIZED VIEW IF EXISTS mv_layout_velocity CASCADE;
+
+CREATE MATERIALIZED VIEW mv_layout_velocity AS
+WITH last24mo AS (
+    SELECT
+        ocm.project_name,
+        CASE
+            WHEN ocm.room_bucket = 'студия' THEN 'studio'
+            ELSE ocm.room_bucket
+        END                                                         AS room_bucket,
+        ocm.deals_total_count,
+        ocm.deals_total_avg_area_m2,
+        ocm.deals_total_avg_price_thousand_rub_per_m2,
+        ocm.deals_total_vol_m2,
+        ocm.report_month
+    FROM objective_corpus_room_month ocm
+    WHERE ocm.report_month >= (NOW() - INTERVAL '24 months')::date
+)
+SELECT
+    cm.domrf_obj_id                                                 AS obj_id,
+    l.room_bucket,
+    SUM(l.deals_total_count)::int                                   AS total_deals_24mo,
+    AVG(l.deals_total_avg_area_m2)::numeric(10, 2)                 AS avg_area_m2,
+    AVG(l.deals_total_avg_price_thousand_rub_per_m2)::numeric(12, 2)
+                                                                    AS avg_price_thousand_rub_per_m2,
+    SUM(l.deals_total_vol_m2)::numeric(12, 2)                      AS total_vol_m2,
+    MIN(l.report_month)                                             AS window_start,
+    MAX(l.report_month)                                             AS window_end,
+    COUNT(DISTINCT l.report_month)::int                            AS months_with_data
+FROM last24mo l
+JOIN objective_complex_mapping cm
+    ON cm.objective_complex_name = l.project_name
+WHERE l.room_bucket IS NOT NULL
+  AND cm.domrf_obj_id IS NOT NULL
+  AND cm.objective_group = 'Екатеринбург'  -- защита от cross-region Cartesian при future multi-city
+GROUP BY cm.domrf_obj_id, l.room_bucket
+WITH NO DATA;
+
+-- UNIQUE index required for REFRESH CONCURRENTLY (future periodic refreshes)
+-- Created on empty MV → instant, no lock duration on data
+CREATE UNIQUE INDEX mv_layout_velocity_pk
+    ON mv_layout_velocity (obj_id, room_bucket);
+
+-- Lookup index used by /best-layouts endpoint (PR C)
+CREATE INDEX mv_layout_velocity_obj_idx
+    ON mv_layout_velocity (obj_id);
+
+-- Initial populate (non-concurrent — MV was just created, CONCURRENTLY not allowed on empty MV)
+REFRESH MATERIALIZED VIEW mv_layout_velocity;
+
+COMMENT ON MATERIALIZED VIEW mv_layout_velocity IS
+  'Per-(obj_id, room_bucket) deals aggregation за last 24 months. '
+  'Source: objective_corpus_room_month × objective_complex_mapping (EKB only). '
+  'Refresh via layout_velocity_refresh.py (concurrently=True after initial populate).';
+
+COMMIT;