diff --git a/tradein-mvp/backend/app/tasks/refresh_search_matview.py b/tradein-mvp/backend/app/tasks/refresh_search_matview.py index 29fc1f8d..485035fa 100644 --- a/tradein-mvp/backend/app/tasks/refresh_search_matview.py +++ b/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: - 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) + for mv in matviews: + mv_start = time.monotonic() + try: + with conn.cursor() as cur: + 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. diff --git a/tradein-mvp/backend/data/sql/139_premium_houses.sql b/tradein-mvp/backend/data/sql/139_premium_houses.sql new file mode 100644 index 00000000..b2983bf1 --- /dev/null +++ b/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;