From 96a9c575b44d9566f54711dd2d4d10f906891939 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 17 Jun 2026 21:48:11 +0300 Subject: [PATCH 1/2] feat(sql): sales-tracker velocity + absorption MVs for Site Finder (#61) --- .../site_finder/sales_tracker_mv_refresh.py | 95 +++++++++ backend/app/workers/beat_schedule.py | 17 ++ backend/app/workers/celery_app.py | 1 + .../workers/tasks/mv_sales_tracker_refresh.py | 52 +++++ ...1_mv_sales_tracker_velocity_absorption.sql | 184 ++++++++++++++++++ 5 files changed, 349 insertions(+) create mode 100644 backend/app/services/site_finder/sales_tracker_mv_refresh.py create mode 100644 backend/app/workers/tasks/mv_sales_tracker_refresh.py create mode 100644 data/sql/161_mv_sales_tracker_velocity_absorption.sql diff --git a/backend/app/services/site_finder/sales_tracker_mv_refresh.py b/backend/app/services/site_finder/sales_tracker_mv_refresh.py new file mode 100644 index 00000000..fa2afed1 --- /dev/null +++ b/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] , 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 diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 0252d9b7..104c6a5c 100644 --- a/backend/app/workers/beat_schedule.py +++ b/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 diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index cec38d12..86fa867c 100644 --- a/backend/app/workers/celery_app.py +++ b/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" diff --git a/backend/app/workers/tasks/mv_sales_tracker_refresh.py b/backend/app/workers/tasks/mv_sales_tracker_refresh.py new file mode 100644 index 00000000..a38790be --- /dev/null +++ b/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() diff --git a/data/sql/161_mv_sales_tracker_velocity_absorption.sql b/data/sql/161_mv_sales_tracker_velocity_absorption.sql new file mode 100644 index 00000000..3ffeea5e --- /dev/null +++ b/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; From 2cf6261005e7bb3d234485e5cfb061a4367ad52e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 17 Jun 2026 22:36:39 +0300 Subject: [PATCH 2/2] =?UTF-8?q?chore(sql):=20renumber=20sales-tracker=20MV?= =?UTF-8?q?=20migration=20161=E2=86=92164=20(avoid=20collision)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/services/site_finder/sales_tracker_mv_refresh.py | 2 +- backend/app/workers/tasks/mv_sales_tracker_refresh.py | 2 +- ...orption.sql => 164_mv_sales_tracker_velocity_absorption.sql} | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) rename data/sql/{161_mv_sales_tracker_velocity_absorption.sql => 164_mv_sales_tracker_velocity_absorption.sql} (99%) diff --git a/backend/app/services/site_finder/sales_tracker_mv_refresh.py b/backend/app/services/site_finder/sales_tracker_mv_refresh.py index fa2afed1..f5943bb9 100644 --- a/backend/app/services/site_finder/sales_tracker_mv_refresh.py +++ b/backend/app/services/site_finder/sales_tracker_mv_refresh.py @@ -2,7 +2,7 @@ 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: +data/sql/164_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). diff --git a/backend/app/workers/tasks/mv_sales_tracker_refresh.py b/backend/app/workers/tasks/mv_sales_tracker_refresh.py index a38790be..badee03f 100644 --- a/backend/app/workers/tasks/mv_sales_tracker_refresh.py +++ b/backend/app/workers/tasks/mv_sales_tracker_refresh.py @@ -11,7 +11,7 @@ Both MVs are built from the Объектив sales-tracker ("шахматки") (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. +MV-source migration: data/sql/164_mv_sales_tracker_velocity_absorption.sql. """ from __future__ import annotations diff --git a/data/sql/161_mv_sales_tracker_velocity_absorption.sql b/data/sql/164_mv_sales_tracker_velocity_absorption.sql similarity index 99% rename from data/sql/161_mv_sales_tracker_velocity_absorption.sql rename to data/sql/164_mv_sales_tracker_velocity_absorption.sql index 3ffeea5e..5ce58fd8 100644 --- a/data/sql/161_mv_sales_tracker_velocity_absorption.sql +++ b/data/sql/164_mv_sales_tracker_velocity_absorption.sql @@ -1,4 +1,4 @@ --- 161_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. --