feat(tradein/db): premium_houses materialized view from listing concentration (#2002)

tradein-mvp/backend/app/tasks/refresh_search_matview.py

@@ -8,6 +8,7 @@ in tradein-mvp/backend/app/). This module provides both:
 NEEDS COORDINATION: main session must decide whether to bootstrap Celery in tradein-mvp
 or run via external scheduler (systemd timer / OS cron). Until then, no Beat schedule.
 """
+
 from __future__ import annotations
 
 import logging
@@ -21,20 +22,35 @@ logger = logging.getLogger(__name__)
 
 
 def refresh_search_matview() -> None:
-    """REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3, sync).
+    """REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv + premium_houses.
 
-    Logs duration. Idempotent. Safe to run during read traffic (CONCURRENTLY).
+    psycopg v3, sync. Logs duration. Idempotent. Safe to run during read traffic
+    (CONCURRENTLY). premium_houses (#2002) derives from `listings`, so it is refreshed
+    right after the search MV in the same session. Each MV is refreshed independently —
+    a failure on one is logged and does not abort the other (autocommit, so a failed
+    REFRESH does not leave the connection in an aborted transaction).
     """
     start = time.monotonic()
     # DATABASE_URL is SQLAlchemy dialect form (postgresql+psycopg://...) in tradein-mvp
     # (see tradein-mvp/docker-compose.prod.yml). libpq / psycopg.connect() accepts only
     # postgresql:// or postgres:// — strip the +psycopg dialect prefix.
     dsn = settings.database_url.replace("postgresql+psycopg://", "postgresql://", 1)
+    # CONCURRENTLY requires a UNIQUE index on each MV (listings_search_mv,
+    # premium_houses_house_id_uidx). Order matters: premium_houses depends on the same
+    # `listings` rows, so refresh it after the search MV is fresh.
+    matviews = ("listings_search_mv", "premium_houses")
     with psycopg.connect(dsn, autocommit=True) as conn:
+        for mv in matviews:
+            mv_start = time.monotonic()
+            try:
                 with conn.cursor() as cur:
-            cur.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv")
-    elapsed = time.monotonic() - start
-    logger.info("listings_search_mv refresh completed in %.2fs", elapsed)
+                    cur.execute(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {mv}")
+            except Exception:
+                # Don't let one MV's failure skip the others; surface for alerting.
+                logger.exception("%s refresh failed", mv)
+            else:
+                logger.info("%s refresh completed in %.2fs", mv, time.monotonic() - mv_start)
+    logger.info("matview refresh batch completed in %.2fs", time.monotonic() - start)
 
 
 # Celery-task wrapper — only registers if celery_app exists.

tradein-mvp/backend/data/sql/139_premium_houses.sql

@@ -0,0 +1,56 @@
+-- 139_premium_houses.sql
+-- Purpose: premium_houses materialized view — premium / elite EKB buildings
+--          identified by listing concentration (#2002, premium-building identification).
+--
+-- Rationale (#2002):
+--   We flag a house as "premium" by the concentration of expensive listings on it.
+--   Validated query returns ~297 houses that together cover ~65% of all >=20M RUB
+--   elite listings. A house qualifies when it has a meaningful sample (n >= 4 active
+--   listings with sane price_per_m2) AND either:
+--     - a high median price_per_m2 (>= 220000 RUB/m2), OR
+--     - at least 3 listings priced >= 20M RUB (n_elite >= 3).
+--   price_per_m2 is clamped to [100000, 800000] to drop scrape garbage / mis-parsed
+--   areas before aggregating, so the median is robust.
+--
+-- house_id dupes: house_dedup is imperfect, so the same physical building can appear
+--   under several house_id values. That is FINE here — we deliberately keep every
+--   variant in the set; dropping dupes would only shrink coverage, never improve it.
+--   The unique index is on house_id (each MV row is one house_id), not on a physical
+--   building, so distinct house_id rows never collide.
+--
+-- Refresh: refreshed by the existing MV-refresh task (app/tasks/refresh_search_matview.py,
+--   scheduled as `refresh_search_matview` in scrape_schedules) right AFTER
+--   listings_search_mv — premium_houses derives from `listings`, so refreshing it once
+--   the search MV is fresh keeps both in step. REFRESH ... CONCURRENTLY requires the
+--   UNIQUE index below.
+--
+-- Dependencies: 002_core_tables.sql (listings: house_id_fk, price_rub, price_per_m2,
+--               area_m2, is_active).
+-- Deploy order: after 002_core_tables.sql. No backend schema coupling.
+-- Re-run safe: CREATE MATERIALIZED VIEW IF NOT EXISTS + CREATE UNIQUE INDEX IF NOT EXISTS.
+--   No DROP — dropping would break in-flight REFRESH ... CONCURRENTLY and momentarily
+--   delete the object during deploy.
+
+BEGIN;
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS premium_houses AS
+WITH lh AS (
+    SELECT l.house_id_fk AS house_id,
+        count(*) AS n,
+        count(*) FILTER (WHERE l.price_rub >= 20000000) AS n_elite,
+        percentile_cont(0.5) WITHIN GROUP (ORDER BY l.price_per_m2)
+            FILTER (WHERE l.price_per_m2 BETWEEN 100000 AND 800000) AS med_ppm2
+    FROM listings l
+    WHERE l.is_active AND l.house_id_fk IS NOT NULL AND l.area_m2 > 0
+          AND l.price_per_m2 BETWEEN 100000 AND 800000
+    GROUP BY l.house_id_fk
+)
+SELECT house_id, n, n_elite, round(med_ppm2)::int AS med_ppm2
+FROM lh
+WHERE n >= 4 AND (med_ppm2 >= 220000 OR n_elite >= 3);
+
+-- UNIQUE index is mandatory for REFRESH MATERIALIZED VIEW CONCURRENTLY.
+CREATE UNIQUE INDEX IF NOT EXISTS premium_houses_house_id_uidx
+    ON premium_houses (house_id);
+
+COMMIT;