feat(sql): sales-tracker velocity + absorption MVs for Site Finder (#61)

backend/app/services/site_finder/sales_tracker_mv_refresh.py

@@ -0,0 +1,95 @@
+"""Refresh helper for the sales-tracker MVs (Issue #61).
+
+Two independent materialized views built from the Объектив sales-tracker
+("шахматки") snapshots (objective_lots / objective_lots_history), created by
+data/sql/161_mv_sales_tracker_velocity_absorption.sql:
+
+  1. mv_sales_tracker_velocity_by_district  — per (district, month) sold/total/
+     avg-sold-price. Feeds the Site Finder Velocity Score (4th scoring criterion).
+  2. mv_sales_tracker_absorption_curves     — cumulative sold% as f(months from
+     sales_start_date) per (rooms_int, area_bucket). Foundation for recommend_mix
+     + sellout forecast.
+
+The two MVs do not depend on each other, so refresh order is irrelevant; both
+are refreshed in the same call.
+
+Scheduled via Celery beat hardcoded entry in workers/beat_schedule.py
+('mv-sales-tracker-refresh-weekly', Mon 04:30 MSK).
+
+Usage example (manual, via psql-connected shell or admin endpoint):
+    from app.services.site_finder.sales_tracker_mv_refresh import refresh_sales_tracker_mvs
+
+    counts = refresh_sales_tracker_mvs(db)
+    # logs: "mv_sales_tracker_velocity_by_district refreshed: 70 rows", etc.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.exc import DatabaseError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+_MV_NAMES: tuple[str, ...] = (
+    "mv_sales_tracker_velocity_by_district",
+    "mv_sales_tracker_absorption_curves",
+)
+
+
+def _refresh_mv(db: Session, mv_name: str, *, concurrently: bool) -> int:
+    """Run REFRESH MATERIALIZED VIEW [CONCURRENTLY] <mv_name>, return row count.
+
+    Falls back to non-concurrent on the known "cannot refresh concurrently"
+    error (MV empty or no UNIQUE index — should not happen in prod since the
+    migration creates the UNIQUE index and populates the MV, but provides a
+    safe recovery path for first-run / post-recreation edge cases).
+    """
+    try:
+        if concurrently:
+            db.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {mv_name}"))
+        else:
+            db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
+        db.commit()
+    except DatabaseError as e:
+        # PostgreSQL emits "CONCURRENTLY cannot be used when the materialized
+        # view ... is not populated" (matview.c, SQLSTATE 55000), surfaced by
+        # psycopg3 as an InternalError (a DatabaseError sibling).
+        if concurrently and "concurrently cannot be used" in str(e).lower():
+            logger.warning(
+                "%s: CONCURRENTLY failed (MV likely not populated), "
+                "falling back to non-concurrent refresh",
+                mv_name,
+            )
+            db.rollback()
+            db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
+            db.commit()
+        else:
+            raise
+
+    row = db.execute(text(f"SELECT COUNT(*) FROM {mv_name}")).first()
+    count = int(row[0]) if row else 0
+    logger.info("%s refreshed: %d rows", mv_name, count)
+    return count
+
+
+def refresh_sales_tracker_mvs(db: Session, *, concurrently: bool = True) -> dict[str, int]:
+    """Refresh both sales-tracker MVs.
+
+    Args:
+        db: SQLAlchemy Session (sync).
+        concurrently: When True, uses REFRESH CONCURRENTLY (non-blocking —
+            readers continue). Requires the per-MV UNIQUE indexes
+            (mv_sales_tracker_velocity_by_district_pk,
+            mv_sales_tracker_absorption_curves_pk) and the MVs to be already
+            populated. Pass False only for first populate or after recreation.
+
+    Returns:
+        Mapping mv_name -> row count after refresh (for observability).
+    """
+    counts: dict[str, int] = {}
+    for mv_name in _MV_NAMES:
+        counts[mv_name] = _refresh_mv(db, mv_name, concurrently=concurrently)
+    return counts

backend/app/workers/beat_schedule.py

@@ -522,4 +522,21 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # Sales-tracker MVs (#61) — питают Site Finder Velocity Score (4-й критерий) +
+    # recommend_mix / sellout-forecast. Оба MV (mv_sales_tracker_velocity_by_district,
+    # mv_sales_tracker_absorption_curves) рефрешатся CONCURRENTLY (non-blocking, требуют
+    # unique-индексы из миграции 161). Источник — objective_lots / objective_lots_history
+    # (Объектив-шахматки), наполняются objective_sync (Mon 04:15 МСК по умолчанию).
+    #
+    # Понедельник 04:30 МСК (Celery conf.timezone=Europe/Moscow → crontab в МСК, #1233) —
+    # ПОСЛЕ objective_sync (04:15), чтобы агрегаты считались по свежему снапшоту; в
+    # окне до тяжёлого monday-кластера site_finder-рефрешей (ird 05:00, gknspecial 05:30,
+    # supply-layers 06:00). Refresh лёгкий (~6с на 1.1M lots). Техническая infra-задача,
+    # не в job_settings (как refresh-quarter-price-index / refresh-layout-velocity).
+    schedule["mv-sales-tracker-refresh-weekly"] = {
+        "task": "tasks.mv_sales_tracker_refresh.refresh_sales_tracker_mvs",
+        "schedule": _parse_cron("30 4 * * mon"),  # 04:30 MSK, понедельник
+        "options": {"queue": "celery"},
+    }
+
     return schedule

backend/app/workers/celery_app.py

@@ -82,6 +82,7 @@ celery_app = Celery(
         "app.workers.tasks.izyatie_ocr_ingest",
         "app.workers.tasks.developer_registry_refresh",
         "app.workers.tasks.refresh_layout_velocity",
+        "app.workers.tasks.mv_sales_tracker_refresh",
     ],
 )
 celery_app.conf.timezone = "Europe/Moscow"

backend/app/workers/tasks/mv_sales_tracker_refresh.py

@@ -0,0 +1,52 @@
+"""Celery task: refresh the sales-tracker MVs (Issue #61).
+
+Scheduled via hardcoded beat entry in workers/beat_schedule.py:
+  'mv-sales-tracker-refresh-weekly' — weekly on Monday at 04:30 MSK.
+
+Refreshes (both CONCURRENTLY, non-blocking):
+  - mv_sales_tracker_velocity_by_district  (Site Finder Velocity Score, 4th criterion)
+  - mv_sales_tracker_absorption_curves     (recommend_mix + sellout forecast foundation)
+
+Both MVs are built from the Объектив sales-tracker ("шахматки") snapshots
+(objective_lots / objective_lots_history). Source data refreshes via the
+objective_sync beat job, so a weekly MV refresh keeps the aggregates current.
+
+MV-source migration: data/sql/161_mv_sales_tracker_velocity_absorption.sql.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.core.db import SessionLocal
+from app.services.site_finder.sales_tracker_mv_refresh import refresh_sales_tracker_mvs
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.mv_sales_tracker_refresh.refresh_sales_tracker_mvs",
+    max_retries=2,
+)
+def refresh_sales_tracker_mvs_task(self: Any) -> dict[str, Any]:
+    """REFRESH both sales-tracker MVs (#61).
+
+    Both MVs are refreshed CONCURRENTLY (non-blocking, require their UNIQUE
+    indexes created by migration 161); the service falls back to non-concurrent
+    if an MV is found unpopulated (first-run edge case).
+
+    Returns result dict for the Celery task result store / logging.
+    """
+    db = SessionLocal()
+    try:
+        counts = refresh_sales_tracker_mvs(db, concurrently=True)
+        logger.info("refresh_sales_tracker_mvs: completed, rows=%s", counts)
+        return {"status": "ok", "rows": counts}
+    except Exception as e:
+        logger.exception("refresh_sales_tracker_mvs failed: %s", e)
+        raise
+    finally:
+        db.close()

data/sql/161_mv_sales_tracker_velocity_absorption.sql

@@ -0,0 +1,184 @@
+-- 161_mv_sales_tracker_velocity_absorption.sql
+-- Issue #61 — Velocity materialized views for Site Finder Velocity Score (4th scoring
+-- criterion) + recommend_mix smart unit-mix. Foundation for sellout forecast.
+--
+-- B2-1 data source ("шахматки" / sales-tracker): the Объектив scraper
+-- (backend/app/workers/tasks/scrape_objective.py) → tables:
+--   objective_lots          — 1.12M rows, one row per tracked lot (current state),
+--                             carries district / rooms_int / area_pd / sales_start_date /
+--                             is_sold / registration_date / contract_date / price_per_m2_rub.
+--   objective_lots_history  — 974k rows, daily-ish per-lot snapshots
+--                             (snapshot_date, is_sold, status, prices).
+-- Snapshot history depth (as of 2026-06-17): 3 captures 2026-05-17 / 05-19 / 06-03 (spans
+-- >2 weeks, sold count moved 193188->194893 => measurable absorption). Cohort/absorption
+-- resolution improves automatically as the weekly scraper accumulates more snapshots.
+--
+-- -- MV 1: mv_sales_tracker_velocity_by_district --------------------------------------
+--   Grain: (district, sale_month). One row per district per month.
+--   Dedup: a lot appears in multiple snapshots within a month -> we keep that lot's LATEST
+--          snapshot within the month (DISTINCT ON lot, snapshot_date DESC) before
+--          aggregating, so total_count is lots-tracked-that-month (not snapshot rows).
+--   Metrics: total_count, sold_count, avg_sold_price_per_m2, avg_sold_price_total,
+--            sold_share (velocity proxy for SF Velocity Score).
+--
+-- -- MV 2: mv_sales_tracker_absorption_curves ----------------------------------------
+--   Grain: (rooms_int, area_bucket, months_since_start). Cumulative sold% as f(months
+--          from first_seen). "first_seen" = objective_lots.sales_start_date (true sales
+--          launch — richer/longer than the 3-snapshot window). Sold-month anchor =
+--          COALESCE(registration_date, contract_date). months_since_start clamped >= 0
+--          (712 noise rows have anchor < start). 99.98% of sold lots carry both dates.
+--          cohort_size = all lots in (rooms, area_bucket) cohort; cum_sold = sold lots
+--          whose months_since_start <= the row's bucket; cum_sold_pct = cum_sold/cohort.
+--   This is snapshot-sparsity-independent (driven by registration dates, not snapshots),
+--          so the curve is usable today and the foundation for sellout forecast.
+--
+-- REFRESH CONCURRENTLY: both MVs get a UNIQUE index on their full grain immediately after
+--   creation (on empty MV -> instant), enabling non-blocking weekly REFRESH CONCURRENTLY.
+--   Scheduled via Celery beat `mv-sales-tracker-refresh-weekly` (Mon 04:30 MSK) ->
+--   task app.workers.tasks.mv_sales_tracker_refresh.refresh_sales_tracker_mvs.
+--
+-- Deploy: auto-applied by deploy.yml via _schema_migrations tracking (one-shot, NN order).
+-- Dependencies on existing objects: objective_lots, objective_lots_history (read-only).
+-- No views depend on these MVs at creation time.
+--
+-- WARN: re-apply (DR / lost _schema_migrations / dev local) DROP ... CASCADE снесёт MV +
+--       зависимости. После re-apply ПЕРВЫЙ refresh = non-concurrent (CONCURRENTLY падает
+--       на пустой/не-populated MV). _schema_migrations нормально предотвращает re-apply.
+
+BEGIN;
+
+-- ====================================================================================
+-- MV 1: velocity by district x month
+-- ====================================================================================
+DROP MATERIALIZED VIEW IF EXISTS mv_sales_tracker_velocity_by_district CASCADE;
+
+CREATE MATERIALIZED VIEW mv_sales_tracker_velocity_by_district AS
+WITH lot_month AS (
+    -- One row per (lot, month): the lot's latest snapshot within that month.
+    SELECT DISTINCT ON (h.objective_lot_id, date_trunc('month', h.snapshot_date))
+        l.district                                          AS district,
+        date_trunc('month', h.snapshot_date)::date          AS sale_month,
+        h.objective_lot_id,
+        h.is_sold,
+        h.price_per_m2_rub,
+        h.price_calculated_total_rub
+    FROM objective_lots_history h
+    JOIN objective_lots l ON l.objective_lot_id = h.objective_lot_id
+    WHERE l.district IS NOT NULL
+    ORDER BY h.objective_lot_id,
+             date_trunc('month', h.snapshot_date),
+             h.snapshot_date DESC
+)
+SELECT
+    district,
+    sale_month,
+    count(*)::int                                                       AS total_count,
+    count(*) FILTER (WHERE is_sold)::int                                AS sold_count,
+    round(
+        count(*) FILTER (WHERE is_sold)::numeric
+        / NULLIF(count(*), 0), 4
+    )                                                                    AS sold_share,
+    round(avg(price_per_m2_rub) FILTER (WHERE is_sold), 2)              AS avg_sold_price_per_m2,
+    round(avg(price_calculated_total_rub) FILTER (WHERE is_sold), 2)    AS avg_sold_price_total
+FROM lot_month
+GROUP BY district, sale_month
+WITH NO DATA;
+
+-- UNIQUE index on full grain -> enables REFRESH CONCURRENTLY (created on empty MV = instant)
+CREATE UNIQUE INDEX mv_sales_tracker_velocity_by_district_pk
+    ON mv_sales_tracker_velocity_by_district (district, sale_month);
+
+CREATE INDEX mv_sales_tracker_velocity_district_idx
+    ON mv_sales_tracker_velocity_by_district (district);
+
+REFRESH MATERIALIZED VIEW mv_sales_tracker_velocity_by_district;
+
+COMMENT ON MATERIALIZED VIEW mv_sales_tracker_velocity_by_district IS
+  'Issue #61. Per (district, month) sold/total/avg-sold-price from objective_lots_history '
+  'snapshots (Obektiv shahmatka), deduped to latest snapshot per lot per month. '
+  'Feeds Site Finder Velocity Score. Refresh weekly CONCURRENTLY.';
+
+-- ====================================================================================
+-- MV 2: absorption curves by room_count x area_bucket x months-from-first-seen
+-- ====================================================================================
+DROP MATERIALIZED VIEW IF EXISTS mv_sales_tracker_absorption_curves CASCADE;
+
+CREATE MATERIALIZED VIEW mv_sales_tracker_absorption_curves AS
+WITH base AS (
+    -- One row per lot. area_bucket from area_pd; months_since_start = whole months between
+    -- sales_start_date and the sold anchor (reg/contract). Unsold lots have NULL anchor.
+    SELECT
+        l.rooms_int,
+        CASE
+            WHEN l.area_pd < 30  THEN '<30'
+            WHEN l.area_pd < 45  THEN '30-45'
+            WHEN l.area_pd < 60  THEN '45-60'
+            WHEN l.area_pd < 80  THEN '60-80'
+            ELSE '80+'
+        END                                                             AS area_bucket,
+        l.is_sold,
+        CASE
+            WHEN l.is_sold
+             AND l.sales_start_date IS NOT NULL
+             AND COALESCE(l.registration_date, l.contract_date) IS NOT NULL
+            THEN GREATEST(
+                0,
+                (date_part('year',  age(COALESCE(l.registration_date, l.contract_date),
+                                         l.sales_start_date)) * 12
+                 + date_part('month', age(COALESCE(l.registration_date, l.contract_date),
+                                          l.sales_start_date)))::int
+            )
+        END                                                             AS months_since_start
+    FROM objective_lots l
+    WHERE l.rooms_int IS NOT NULL
+      AND l.area_pd  IS NOT NULL
+      AND l.sales_start_date IS NOT NULL
+),
+cohort AS (
+    SELECT rooms_int, area_bucket, count(*)::int AS cohort_size
+    FROM base
+    GROUP BY rooms_int, area_bucket
+),
+sold_at_month AS (
+    SELECT rooms_int, area_bucket, months_since_start, count(*)::int AS sold_in_month
+    FROM base
+    WHERE is_sold AND months_since_start IS NOT NULL
+    GROUP BY rooms_int, area_bucket, months_since_start
+)
+SELECT
+    s.rooms_int,
+    s.area_bucket,
+    s.months_since_start,
+    c.cohort_size,
+    -- cumulative sold up to and including this month-offset (per cohort)
+    SUM(s.sold_in_month) OVER (
+        PARTITION BY s.rooms_int, s.area_bucket
+        ORDER BY s.months_since_start
+        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+    )::int                                                              AS cum_sold,
+    round(
+        SUM(s.sold_in_month) OVER (
+            PARTITION BY s.rooms_int, s.area_bucket
+            ORDER BY s.months_since_start
+            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+        )::numeric / NULLIF(c.cohort_size, 0), 4
+    )                                                                    AS cum_sold_pct
+FROM sold_at_month s
+JOIN cohort c ON c.rooms_int = s.rooms_int AND c.area_bucket = s.area_bucket
+WITH NO DATA;
+
+-- UNIQUE index on full grain -> enables REFRESH CONCURRENTLY
+CREATE UNIQUE INDEX mv_sales_tracker_absorption_curves_pk
+    ON mv_sales_tracker_absorption_curves (rooms_int, area_bucket, months_since_start);
+
+CREATE INDEX mv_sales_tracker_absorption_cohort_idx
+    ON mv_sales_tracker_absorption_curves (rooms_int, area_bucket);
+
+REFRESH MATERIALIZED VIEW mv_sales_tracker_absorption_curves;
+
+COMMENT ON MATERIALIZED VIEW mv_sales_tracker_absorption_curves IS
+  'Issue #61. Cumulative sold-pct as f(months from sales_start_date) per (rooms_int, '
+  'area_bucket). Anchor = COALESCE(registration_date, contract_date) from objective_lots. '
+  'Foundation for recommend_mix + sellout forecast. Refresh weekly CONCURRENTLY.';
+
+COMMIT;