Compare commits
No commits in common. "81e708b953b6fd38576200a537bc95237bf4c887" and "8a24b057ac1b48693b72fb4091589ab467a2d8ab" have entirely different histories.
81e708b953
...
8a24b057ac
5 changed files with 0 additions and 349 deletions
|
|
@ -1,95 +0,0 @@
|
||||||
"""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
|
|
||||||
|
|
@ -522,21 +522,4 @@ def build_beat_schedule() -> dict:
|
||||||
"options": {"queue": "celery"},
|
"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
|
return schedule
|
||||||
|
|
|
||||||
|
|
@ -82,7 +82,6 @@ celery_app = Celery(
|
||||||
"app.workers.tasks.izyatie_ocr_ingest",
|
"app.workers.tasks.izyatie_ocr_ingest",
|
||||||
"app.workers.tasks.developer_registry_refresh",
|
"app.workers.tasks.developer_registry_refresh",
|
||||||
"app.workers.tasks.refresh_layout_velocity",
|
"app.workers.tasks.refresh_layout_velocity",
|
||||||
"app.workers.tasks.mv_sales_tracker_refresh",
|
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
celery_app.conf.timezone = "Europe/Moscow"
|
celery_app.conf.timezone = "Europe/Moscow"
|
||||||
|
|
|
||||||
|
|
@ -1,52 +0,0 @@
|
||||||
"""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()
|
|
||||||
|
|
@ -1,184 +0,0 @@
|
||||||
-- 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;
|
|
||||||
Loading…
Add table
Reference in a new issue