gendesign/data/sql/164_mv_sales_tracker_velocity_absorption.sql

-- 164_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;