lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

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

Merged
lekss361 merged 2 commits from feat/velocity-mv-61 into main 2026-06-17 19:43:15 +00:00
Conversation 0 Commits 2 Files changed 5 +349
5 changed files with 349 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

95
backend/app/services/site_finder/sales_tracker_mv_refresh.py Normal file
View file

@ -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/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).
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

17
backend/app/workers/beat_schedule.py
View file

@ -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

1
backend/app/workers/celery_app.py
View file

@ -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"

52
backend/app/workers/tasks/mv_sales_tracker_refresh.py Normal file
View file

@ -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/164_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()

184
data/sql/164_mv_sales_tracker_velocity_absorption.sql Normal file
View file

@ -0,0 +1,184 @@
-- 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;